DEV Community: Boaz Leleina

FinOps Sentinel: AWS Cost Protection Agent

Boaz Leleina — Thu, 30 Jul 2026 07:08:24 +0000

An autonomous agent that finds cloud waste, prices it, explains it, and removes it, but only after a human says yes.

Demo

The problem

Cloud waste is not an event. It is accumulation.

An engineer terminates an instance; the EBS volume it booted from survives, unattached, billing monthly. Someone reserves an Elastic IP for a migration that finished last quarter. A staging database gets stopped rather than deleted, so it is still billing for allocated storage, and scheduled by AWS to restart itself after seven days whether anyone wants it or not.

Every item is individually too small to chase. Together they are the line on the bill nobody can account for.

The awkward part is that this is not a detection problem. An unattached volume is one API call from obvious. The hard part is that acting on it means deleting infrastructure in a live account, where the cost of one wrong deletion dwarfs months of savings.

So tooling splits into two unsatisfying camps: dashboards that report waste and leave a human to do all the work anyway, and automation that acts confidently and occasionally deletes the wrong thing.

FinOps Sentinel is an attempt at the middle position. The agent finds, prices, explains, and executes the mechanics. A human decides, always.

What it does

Four verbs:

Command	What happens
`sentinel scan`	Inventory every region, evaluate nine rules, persist findings, alert on new ones
Slack button / `POST /decisions/{id}`	A human approves or denies; guardrails re-checked; playbook runs
`sentinel digest`	Advisory patterns: right-sizing suggestions and spend anomalies. No buttons
`sentinel expire`	Findings nobody decided within 72 hours close themselves

The nine rules

Rule	Detects	Actionable?
`ebs_unattached`	Volumes in `available` state	Yes: snapshot, then delete
`eip_orphaned`	Elastic IPs with no association	Yes: release
`ec2_stopped`	Instances stopped beyond threshold	Yes: terminate
`ebs_old_snapshot`	Snapshots past retention, or whose source volume is gone	Yes: delete
`ec2_idle`	Running instances, CPU and network flatlined	No: advisory
`rds_idle`	Databases with no connections across the window	No: advisory
`rds_stopped`	Databases in `stopped` state	No: advisory
`s3_no_lifecycle`	Buckets over a size threshold with no lifecycle policy	No: advisory
`s3_incomplete_multipart`	Multipart uploads abandoned past threshold	Yes: abort uploads

The actionable/advisory split is not a maturity gradient. It is a safety boundary, and it is enforced in two independent places.

A real scan:

Regions: us-east-1, eu-west-1, ap-southeast-2 (3)
Loaded 8 scanners per region.
Findings database: data/sentinel.db

Scan completed in 3.15s
Inventory Discovered: 45 resources

Findings Generated: 24 violations
Notifications Sent: 15 (via slack)

                          Optimization Opportunities
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━┓
┃ Resource ID                ┃ Region         ┃ Rule             ┃     $/mo ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━┩
│ vol-d6e748bf37ac6ccec      │ us-east-1      │ ebs_unattached   │    $0.80 │
│ [P] vol-645a783cf3f376d0d  │ us-east-1      │ ebs_unattached   │    $1.60 │
│ eipalloc-db037e1ade3a739fc │ us-east-1      │ eip_orphaned     │    $3.65 │
│ i-da0a5c483e98127c9        │ us-east-1      │ ec2_idle         │   $70.08 │
│ i-a91ac43e9f2689209        │ eu-west-1      │ ec2_idle         │  $280.32 │
│ finops-demo-unmanaged-eu-… │ eu-west-1      │ s3_no_lifecycle  │   $18.40 │
└────────────────────────────┴────────────────┴──────────────────┴──────────┘
[P] protected by tag: reported, never notified, never remediated, and excluded from the totals below.

Total Potential Monthly Savings: $540.02

Design: ports and adapters

The system is built hexagonally. Business rules live in domain/ and import nothing but pydantic: no boto3, no SQLAlchemy, no HTTP client, no Slack SDK.

src/finops_sentinel/
├── domain/         # Pure business logic. No AWS, no HTTP, no SQL.
│   ├── models.py       # Entities + the state machine table
│   ├── rules.py        # Guardrails: protection tags, playbook allowlist
│   ├── services.py     # Use cases: run_scan, approve, deny, digest
│   ├── summaries.py    # Deterministic prose (the LLM fallback floor)
│   ├── rightsizing.py  # Pure: is this instance over-provisioned?
│   └── anomaly.py      # Pure: is today's waste statistically unusual?
├── ports/          # Abstract interfaces: the domain's vocabulary
│   ├── cloud.py        # CloudGateway
│   ├── repository.py   # FindingsRepository
│   ├── notifier.py     # Notifier
│   ├── advisor.py      # Advisor
│   ├── pricing.py      # Pricing
│   └── scanner.py      # Scanner
├── adapters/       # Concrete implementations
│   ├── aws/            # boto3 gateway, price tables, six scanners
│   ├── persistence/    # SQLAlchemy + SQLite
│   ├── notifications/  # Slack (Block Kit), Console
│   ├── advisor/        # Ollama (local LLM), Template
│   └── inbound/        # Typer CLI, FastAPI
├── bootstrap.py    # Composition root: the only file that wires
└── config.py       # Settings, one per env var

The dependency rule is enforced in CI, not by discipline:

[[tool.importlinter.contracts]]
name = "Domain is pure Python (pydantic only)"
type = "forbidden"
source_modules = ["finops_sentinel.domain"]
forbidden_modules = [
    "boto3", "botocore", "fastapi", "sqlalchemy", "alembic",
    "slack_sdk", "typer", "rich", "uvicorn", "httpx",
]

Import boto3 into the domain and the build fails with a named contract violation.

This changed the design at least once. The spec for anomaly detection said "deterministic pandas, in domain." But that contract is literally named "Domain is pure Python (pydantic only)", and adding pandas would have made the name a lie. A rolling z-score turned out to be twenty lines of statistics. A 60MB dependency avoided because the architecture was enforced rather than merely described.

How a scan works

Two passes, and the reason for two is specific.

Pass 1: discovery. Every scanner in every region returns (Resource, raw_provider_dict) pairs. These are upserted; anything not seen this scan is marked DELETED.

Pass 2: evaluation. Each scanner receives its own region's inventory and emits findings.

They are separate because evaluation sometimes needs inventory the scanner did not discover. The ec2_stopped rule prices a stopped instance by summing its still-billing EBS volumes: volumes that a different scanner found. Pass 1 builds the shared picture; pass 2 reads it.

class Scanner(ABC):
    @abstractmethod
    def discover(self, gateway: CloudGateway) -> list[tuple[Resource, dict[str, Any]]]:
        """Pass 1: Discover resources from the cloud provider."""

    @abstractmethod
    def evaluate(self, resources: list[tuple[Resource, dict[str, Any]]]) -> list[Finding]:
        """Pass 2: Evaluate a list of resources to generate findings."""

Three isolation boundaries
Boundary The failure it prevents Per-region One region's outage or missing IAM grant blinding all others Per-scanner One unavailable service discarding every other scanner's findings in that region DELETED sweep scoped to succeeded regions A failed region's inventory being marked deleted, silently disarming every finding it owns
The per-scanner boundary was added after a real incident. Adding RDS scanners broke sentinel scan outright on LocalStack: RDS raised, the region was marked failed, and with one region configured the entire scan aborted. Every other scanner's findings vanished.

On real AWS, a single missing IAM permission does exactly the same thing, and "no findings" is indistinguishable from a clean account. That is the most expensive failure mode this system has, so it is now structurally impossible:

for index, scanner in enumerate(target.scanners):
    try:
        discovered.extend(scanner.discover(target.gateway))
    except Exception as exc:  # one bad scanner must not end the region
        failures[index] = f"{type(exc).__name__}: {exc}"
        logger.warning("Scanner %s failed in region %s: %s", ...)

And when every region fails, the scan raises rather than returning an empty list:

if targets and not inventory_by_region:
    raise RuntimeError("Every region failed to scan: " + ...)

Returning "no findings" there would read as a clean account. That is the most expensive lie the system could tell.

The state machine

Findings move through a lifecycle, and the transition table lives in the domain next to the enum it governs:

TRANSITIONS: dict[FindingStatus, set[FindingStatus]] = {
    FindingStatus.OPEN:     {FindingStatus.NOTIFIED},
    FindingStatus.NOTIFIED: {FindingStatus.APPROVED, FindingStatus.DENIED,
                             FindingStatus.EXPIRED},
    FindingStatus.APPROVED: {FindingStatus.REMEDIATED, FindingStatus.FAILED},
    # DENIED / REMEDIATED / FAILED / EXPIRED are terminal in v1
}

                    ┌──────────────── EXPIRED (72h, terminal)
                    │    OPEN ──────▶ NOTIFIED ──────▶ DENIED (terminal)
    │               │     │
    │               └──────▶ APPROVED ──┬──▶ REMEDIATED (terminal)
    │                                   └──▶ FAILED (terminal)
    │     └── protected findings stay here forever, by design

Four properties fall out of this:

A re-scan can never resurrect a decided finding. save_finding() updates evidence, cost, and last_seen_at, but never touches status. -_ Alerts fire exactly once_, on OPEN → NOTIFIED. This is why a second scan sends no Slack messages, by design, not by bug.
Protected findings never leave OPEN. Reported in the table, excluded from every total.
Dry-run stops at APPROVED. Only real execution reaches REMEDIATED, so the status itself distinguishes "a human said yes" from "the cloud changed."

Transitions are atomic compare-and-swap:

def transition_finding(self, finding_id, expected, new) -> bool:
    """UPDATE findings SET status=:new WHERE id=:id AND status=:expected.
    Returns True only if THIS call performed the transition (rowcount 1)."""

A Slack double-click, a race against the expiry job, or two concurrent scans cannot execute a remediation twice. The second caller gets rowcount 0.

The safety model

Five independent layers stand between a detection and a deletion. No single failure removes more than one.

1. Tag protection.finops:protected=true: never notified, never actionable. Re-checked at approval time against current tags, so if someone tags a resource after the alert goes out, the newer intent wins.

2. The playbook allowlist, keyed by resource type:

PLAYBOOK_ALLOWLIST: dict[ResourceType, str] = {
    ResourceType.EBS_VOLUME:   "snapshot_then_delete_volume",
    ResourceType.ELASTIC_IP:   "release_eip",
    ResourceType.EC2_INSTANCE: "terminate_stopped_instance",
    ResourceType.EBS_SNAPSHOT: "delete_ebs_snapshot",
    ResourceType.S3_BUCKET:    "abort_incomplete_multipart_uploads",
}

No entry, no action. RDS_INSTANCE is absent deliberately.

3. Notify-only rules, catching what a type-keyed allowlist cannot:

NOTIFY_ONLY_RULES: frozenset[str] = frozenset(
    {"ec2_idle", "rds_idle", "rds_stopped", "s3_no_lifecycle"}
)

Two concrete disasters this prevents:

An ec2_idle finding is on a running instance. The type-keyed allowlist would hand it terminate_stopped_instance, which does not check state.
s3_no_lifecycle and s3_incomplete_multipart share S3_BUCKET. Approving "this bucket has no lifecycle policy" would abort multipart uploads, because that is the playbook registered for the type. 4. Human approval. Nothing is remediated automatically, ever. Every execution traces to a named actor in the decisions table.

5. DRY_RUN, default true. Playbooks log what they would do and return {"dry_run": True}.

On top of that, destructive playbooks build their own recovery path:

def _snapshot_then_delete_volume(self, volume_id: str) -> dict[str, Any]:
    response = self.client.create_snapshot(VolumeId=volume_id, ...)
    snapshot_id = response["SnapshotId"]

    waiter = self.client.get_waiter("snapshot_completed")
    waiter.wait(SnapshotIds=[snapshot_id], WaiterConfig={"Delay": 15, "MaxAttempts": 40})

    self.client.delete_volume(VolumeId=volume_id)
    return {"snapshot_id": snapshot_id, "deleted_volume": volume_id}

The waiter is the point. Delete-first-and-hope-the-snapshot-lands makes the operation irreversible on failure.

Where the LLM fits (and where it does not)

A locally hosted model (Ollama, qwen3:30b-a3b by default) writes the explanation attached to each notification. It occupies a deliberately small role:_ it never decides anything and never computes a number._

class Advisor(ABC):
    """The advisor is advisory only: it never decides, never remediates, and
    never gates a notification. Implementations MUST NOT raise: a summary is
    a nice-to-have, so an unreachable or misbehaving backend has to degrade to
    a deterministic template rather than block the pipeline."""

The never-raise contract lives in the port, not in each adapter, because a summary is optional and a notification is not.

Four defences, because the prompt contains user-controlled AWS tags:

An evidence allowlist. Only ~12 named keys reach the prompt.

Field relabelling. The model read observation_days as "running for 14 days," so it is renamed to metric_window_days in the prompt only, while stored evidence keeps the scanner's keys.
Structured output. The Pydantic schema is handed to Ollama's format field and used to validate the reply, so a model ignoring the constraint is caught rather than trusted.
Output is untrusted display copy. Rendered in its own Slack block, never parsed, never used to build action values. Buttons carry system-generated IDs.

def summarize(self, finding: Finding, resource: Resource) -> str:
    fallback = render_template_summary(finding, resource)
    try:
        parsed = self.advise(finding, resource)
    except (httpx.HTTPError, json.JSONDecodeError, ValidationError, KeyError) as exc:
        logger.warning("Ollama advisor failed for finding %s (%s); using template", ...)
        return fallback
    return f"{parsed.summary} (risk: {parsed.risk}) Next step: {parsed.recommended_action}"

Kill Ollama mid-run and the pipeline does not notice: the deterministic template writes the same explanation.

Two rules worth explaining

Right-sizing uses peak CPU, never average

if len(cpu_series) < min_datapoints:
    return None                      # no history is not evidence

max_cpu = max(cpu_series)
if max_cpu >= cpu_headroom_percent:
    return None                      # it gets busy; leave it alone

An instance that idles all day and pegs 90% CPU once an hour has a mean around 4%. A mean-based tool recommends halving the machine that carries the actual workload._ Averages are how right-sizing tools produce advice that takes production down._

The 40% default pairs with a candidate table that steps down at most one size: halving vCPU roughly doubles utilisation, so a 40% peak lands near 80% on the target. Raising the threshold without shortening the candidate list breaks that pairing silently, so both places say so.

Anomaly detection refuses to guess

ordered  = sorted(snapshots, key=lambda s: s.snapshot_date)
latest   = ordered[-1]
baseline = ordered[-(window_days + 1):-1]      # candidate day EXCLUDED

if len(baseline) < min_history_days: return None

stdev = statistics.stdev(values)
if stdev == 0: return None

z = (value - mean) / stdev
if abs(z) < z_threshold: return None

Every guard returns None (no verdict) rather than a weak one. An anomaly alert that fires on three days of history trains people to ignore it, which costs more than the alert it replaced.

Excluding the candidate day from its own baseline is load-bearing: with seven points, a spike included in the mean and stdev it is measured against inflates both enough to hide itself.

And the honesty constraint: there is no billing data in this system, so what is measured is estimated waste, not spend. Every user-facing string says so:

No spend anomaly (or too little history to judge: needs 7 days of scans).

"Too little history to judge" and "no anomaly" are different statements, and only one of them is reassuring.

Cost estimation, honestly

Every rate lives in one file with sources cited, behind a Pricing port. Three caveats are documented at the top because they bound how far the numbers can be trusted:

Other regions cost more: the table applies us-east-1 rates everywhere.
List price is not your price. Reserved Instances and Savings Plans reduce it, and a commitment-covered instance saves nothing when deleted.
Snapshots bill on incremental blocks, so snapshot estimates are upper bounds.

Defaults are mid-range, never cheapest: an unknown instance type must not be dismissed as negligible, but must not manufacture savings that dwarf real findings either.

The most interesting case is S3:

# Only a fraction of a bucket is ever cold enough to tier or expire.
# Reporting the whole bucket as recoverable would let one large bucket
# dominate the savings total and make every number in the report
# untrustworthy. The inputs are in evidence so the estimate is auditable.
addressable = full_cost * Decimal(str(self.addressable_fraction))

A bucket without a lifecycle policy is not 100% waste. So the finding reports 20% of its storage cost, and the fraction, the raw size, and the full cost all go into evidence so the number is auditable rather than magic.

Testing

277 tests. 96% overall coverage, 100% on the domain layer.

Layer Tooling What it proves Domain In-memory fakes Guardrails, state machine, pure rules: no AWS at all Adapters moto Scanners against a simulated AWS API LLM respx Every advisor failure: timeout, transport error, bad JSON, schema violation Integration LocalStack Seed → scan → approve → resource actually gone → audit complete
The gate is split rather than flat:

pytest tests/ --cov=src/finops_sentinel --cov-fail-under=90
coverage report --include='*/domain/*' --fail-under=95

A flat 90% would let the domain rot to 85% as long as adapters compensated, which is backwards, since the domain is the part that must never break. A flat 95% global would push toward thin mock-boto3 tests written purely to move a number.

The test that justifies the architecture runs the entire approve-and-remediate flow with a fake gateway, an in-memory repository, and a fake notifier. Zero AWS, zero HTTP, zero Slack. If it passes, swapping Slack for Telegram cannot break the approval logic.

Three bugs that only appeared when I ran it

**1. CloudWatch dimension sets. **The metric port originally took a single dimension name/value pair. S3's BucketSizeBytes is published against BucketName and StorageType, and CloudWatch matches dimension sets exactly: a query naming one returns nothing.

Every bucket would have read as "size unknown" in production, and no S3 finding would ever have fired. Neither moto nor LocalStack publishes that metric, so no end-to-end test could have caught it: empty is also what "no data" legitimately looks like.

2. The migration that silently kept the old constraint. Widening a CHECK constraint on SQLite requires rebuilding the table. The first version used table_args, which adds to a reflected definition rather than replacing it, so the rebuilt table carried both the old four-value constraint and the new six-value one, and the old one still rejected every new row. It ran without error. Only dumping the schema revealed it.

3. Alembic migrating a different database than the app used. env.py read SENTINEL_DB_PATH with os.getenv; the app read it through pydantic-settings, which also reads .env. Since the documented setup puts the path in .env and not the shell, every migration ever run landed on the wrong file. No symptom for two months, because earlier migrations only created tables SQLAlchemy would have created anyway. The first migration whose table nothing else creates surfaced it instantly.

The fix was giving the URL exactly one owner:

def database_url() -> str:
    """Both the application and Alembic call this. They used to build it
    independently, and drifted."""
    return f"sqlite:///{settings.sentinel_db_path}"

All three argue the same thing: a green test suite is not the same as a working system. Each was found by running the thing against a real environment.

Running it

git clone https://github.com/boazleleina/finops-sentinel.git
cd finops-sentinel
cp .env.example .env

docker compose --profile dev up -d          # LocalStack + API
alembic upgrade head
python3 scripts/seed_localstack.py          # seeds realistic waste

sentinel scan
sentinel digest

The seed script creates unattached volumes (one tagged protected), an orphaned Elastic IP, a stopped instance, orphaned snapshots, an idle running instance with flat CloudWatch metrics, and four S3 buckets: one unmanaged, one with a lifecycle policy that must not be flagged, one protected, one holding an abandoned multipart upload.

LocalStack and real AWS are interchangeable via one environment variable. No code path differs.

What I would build next

Kubernetes deployment: CronJob for scanning, Deployment for the API, Helm chart for thresholds and secrets.

Terraform + real AWS in read-only DRY_RUN=true mode, which is also when the RDS scanners finally get end-to-end coverage (LocalStack's RDS is Pro-tier).
Cost Explorer adapter. The anomaly detector monitors estimated waste today. A Cost Explorer implementation of the same port would swap in real billing data without the z-score logic changing at all, the port design paying off.
Mutation testing on the domain layer. Coverage measures execution, not assertion; mutmut deletes assertions and checks the tests fail.

The takeaway

The interesting engineering here is not the AWS integration. It is that the rules deciding whether to delete infrastructure are pure functions, testable without a network, enforced by CI to stay that way, and wrapped in five independent guardrails, so that the scariest thing the system does is also the part that is easiest to reason about.

Repository:

boazleleina / finops-sentinel

An automated, event-driven AWS Cost Optimization Agent engineered to continuously scan, evaluate, and remediate wasted cloud resources with Human-in-the-Loop Slack approvals.

FinOps Sentinel

FinOps Sentinel is an automated, event-driven AWS Cost Optimization Agent engineered to continuously scan, evaluate, and remediate wasted resources across AWS environments.

By applying strict FinOps principles, it identifies cloud waste (e.g., unattached EBS volumes, orphaned Elastic IPs, stopped EC2 instances, idle running instances), calculates potential monthly savings, and facilitates automated or Human-in-the-Loop (HITL) remediation via Slack.

System Architecture

FinOps Sentinel is built on Hexagonal Architecture (Ports & Adapters) and Domain-Driven Design (DDD). The core business rules are strictly decoupled from external libraries, databases, and AWS interfaces.

Tip

ARCHITECTURE.md is the full engineering document — every port and adapter explained, the database design, the safety model, end-to-end execution flows, and the reasoning behind each significant decision. Read it if you are working on the code rather than running it.

<div class="js-render-enrichment-target" data-json='{"data":"flowchart TD\n %% Define Styles\n classDef core fill:#2b3a42,stroke:#3f51b5,stroke-width:2px,color:#fff\n classDef port fill:#1a2b3c,stroke:#00bcd4,stroke-width:2px,color:#fff,stroke-dasharray: 5 5\n classDef adapter fill:#37474f,stroke:#4caf50,stroke-width:2px,color:#fff\n classDef external fill:#eceff1,stroke:#607d8b,stroke-width:1px,color:#000\n\n subgraph External[\"External World\"]\n CLI[\"CLI Client\"]:::external\n SlackWeb[\"Slack Interface\"]:::external\n AWS[\"AWS (Boto3)\"]:::external\n SQLite[\"SQLite Database\"]:::external\n Ollama[\"Ollama (local LLM)\"]:::external\n end\n\n subgraph Adapters[\"Inbound Adapters\"]\n Typer[\"Typer (cli.py)\"]:::adapter\n FastAPI[\"FastAPI (fastapi_app.py)\"]:::adapter\n end\n\n subgraph Ports[\"Ports (Interfaces)\"]\n CloudGateway[\"CloudGateway\"]:::port\n FindingsRepo[\"FindingsRepository\"]:::port\n Notifier[\"Notifier\"]:::port\n ScannerPort[\"Scanner\"]:::port\n AdvisorPort[\"Advisor\"]:::port\n PricingPort[\"Pricing\"]:::port\n end\n\n subgraph Domain[\"Core Domain\"]\n Models[\"Entity Models\"]:::core\n Services[\"State Machine (services.py)\"]:::core\n end\n\n subgraph OutboundAdapters[\"Outbound Adapters\"]\n BotoAdapter[\"Boto3CloudGateway\"]:::adapter\n SQLAdapter[\"SQLAlchemyRepository\"]:::adapter\n SlackAdapter[\"SlackNotifier\"]:::adapter\n OllamaAdapter[\"OllamaAdvisor\"]:::adapter\n PricingAdapter[\"StaticPricing\"]:::adapter\n end\n\n %% Inbound Flow\n CLI --> Typer\n SlackWeb --> FastAPI\n Typer -.->|Invokes| Services\n FastAPI -.->|Invokes| Services\n\n %% Domain Flow\n Services --> Models\n Services -.->|Depends On| CloudGateway\n Services -.->|Depends On| FindingsRepo\n Services -.->|Depends On| Notifier\n Services -.->|Depends On| ScannerPort\n Services -.->|Depends On| AdvisorPort\n Services -.->|Depends On| PricingPort\n\n %% Outbound Implementation\n BotoAdapter -.->|Implements| CloudGateway\n SQLAdapter -.->|Implements| FindingsRepo\n SlackAdapter -.->|Implements| Notifier\n OllamaAdapter -.->|Implements| AdvisorPort\n PricingAdapter -.->|Implements| PricingPort\n\n %% Outbound Flow\n BotoAdapter --> AWS\n SQLAdapter --> SQLite\n SlackAdapter --> SlackWeb\n OllamaAdapter --> Ollama\n"}' data-plain='flowchart TD
%% Define Styles
classDef core fill:#2b3a42,stroke:#3f51b5,stroke-width:2px,color:#fff
classDef port fill:#1a2b3c,stroke:#00bcd4,stroke-width:2px,color:#fff,stroke-dasharray: 5 5
classDef adapter fill:#37474f,stroke:#4caf50,stroke-width:2px,color:#fff
classDef external fill:#eceff1,stroke:#607d8b,stroke-width:1px,color:#000
subgraph External["External World"]
CLI["CLI Client"]:::external
SlackWeb["Slack Interface"]:::external
AWS["AWS (Boto3)"]:::external
SQLite["SQLite Database"]:::external
Ollama["Ollama (local LLM)"]:::external
end
subgraph Adapters["Inbound Adapters"]
Typer["Typer (cli.py)"]:::adapter
FastAPI["FastAPI (fastapi_app.py)"]:::adapter
end

subgraph Ports["Ports (Interfaces)"]
    CloudGateway["CloudGateway"]:::port
    FindingsRepo["FindingsRepository"]:::port
    Notifier["Notifier"]:::port
    ScannerPort["Scanner"]:::port
    AdvisorPort["Advisor"]:::port
    PricingPort["Pricing"]:::port
end

subgraph Domain["Core Domain"]
    Models["Entity Models"]:::core
    Services["State Machine (services.py)"]:::core
end

subgraph OutboundAdapters["Outbound Adapters"]
    BotoAdapter["Boto3CloudGateway"]:::adapter
    SQLAdapter["SQLAlchemyRepository"]:::adapter
    SlackAdapter["SlackNotifier"]:::adapter
    OllamaAdapter["OllamaAdvisor"]:::adapter
    PricingAdapter["StaticPricing"]:::adapter
end

%% Inbound Flow
CLI --&gt; Typer
SlackWeb --&gt; FastAPI
Typer -.-&gt;|Invokes| Services
FastAPI -.-&gt;|Invokes| Services

%% Domain Flow
Services --&gt; Models
Services -.-&gt;|Depends On| CloudGateway
Services -.-&gt;|Depends On| FindingsRepo
Services -.-&gt;|Depends On| Notifier
Services -.-&gt;|Depends On| ScannerPort
Services -.-&gt;|Depends On| AdvisorPort
Services -.-&gt;|Depends On| PricingPort

%% Outbound Implementation
BotoAdapter -.-&gt;|Implements| CloudGateway
SQLAdapter -.-&gt;|Implements| FindingsRepo
SlackAdapter -.-&gt;|Implements| Notifier
OllamaAdapter -.-&gt;|Implements| AdvisorPort
PricingAdapter -.-&gt;|Implements| PricingPort

%% Outbound Flow
BotoAdapter --&gt; AWS
SQLAdapter --&gt; SQLite
SlackAdapter --&gt; SlackWeb
OllamaAdapter --&gt; Ollama

' dir="auto">

flowchart TD

    %% Define Styles

    classDef core fill:#2b3a42,stroke:#3f51b5,stroke-width:2px,color:#fff

    classDef port fill:#1a2b3c,stroke:#00bcd4,stroke-width:2px,color:#fff,stroke-dasharray: 5 5

    classDef adapter fill:#37474f,stroke:#4caf50,stroke-width:2px,color:#fff

    classDef external

…

View on GitHub

Full engineering documentation, including every port, adapter, and design decision, is in ARCHITECTURE.md.

Building a Slack Approval Workflow That Deletes Cloud Infrastructure

Boaz Leleina — Thu, 30 Jul 2026 06:17:55 +0000

Block Kit, signature verification, and the design decisions that stop a button click from becoming an incident.

That screenshot is a bot asking permission to delete an EBS volume. Clicking
Approve Remediation snapshots the volume, waits for the snapshot to
complete, deletes the volume, and edits the message to say what happened.

Getting that to work is mostly plumbing. Getting it to work safely, so that a
stale click, a replayed request, or a resource someone protected in the
meantime cannot cause damage, is the interesting part.

This walks through both, using the Slack adapter from
FinOps Sentinel.

The shape of the problem

Slack interactivity is two separate channels that only look like a conversation:

Your app ──── incoming webhook ────▶ Slack channel
                                          │
                                     user clicks
                                          │
Your app ◀─── HTTP POST ──────────────────┘
             (a completely new request, from Slack's servers)

The click arrives as an unauthenticated POST from the public internet to
whatever URL you registered. Nothing about the request proves it came from
Slack, or that a human clicked anything. That is the security problem in one
sentence, and everything below follows from it.

Part 1: Setting up the Slack app

Create the app and get a webhook

api.slack.com/apps → Create New App → From scratch
Name it, pick your workspace
Incoming Webhooks → toggle On → Add New Webhook to Workspace
Choose a channel, click Allow, copy the URL

SLACK_WEBHOOK_URL=https://hooks.slack.com/services/TXXXXX/BXXXXX/XXXXXXXX

Webhooks post to exactly one channel and cannot read anything. For a
notification bot that is the right amount of privilege: no OAuth flow, no bot
token, no scopes to review.

Enable interactivity

Interactivity & Shortcuts → toggle On → set the Request URL:

https://your-domain.example/callbacks/slack

Locally you need a tunnel:

ngrok http 8000
# → https://a1b2c3d4.ngrok.app
# Request URL: https://a1b2c3d4.ngrok.app/callbacks/slack

The free ngrok URL changes on every restart, and you must update Slack each
time. Save yourself the confusion: if buttons "do nothing," check this first.

Get the signing secret

Basic Information → App Credentials → Signing Secret → Show.

SLACK_SIGNING_SECRET=your_signing_secret_here

This is what makes callbacks trustworthy. Without it your endpoint will delete
infrastructure for anyone who sends it a well-formed POST.

Part 2: Sending a message worth acting on

Block Kit messages are JSON arrays of blocks. The naive version works
immediately:

from slack_sdk.webhook import WebhookClient

WebhookClient(webhook_url).send(text=f"Idle volume found: {volume_id}")

But an alert an engineer has to act on needs to answer what, where, how much,
and what happens if I click this, in about ten seconds, often on a phone.

Here is the real implementation:

def send_finding_alert(self, finding: Finding, resource: Resource) -> str | None:
    webhook_url = settings.slack_webhook_url
    if not webhook_url:
        raise RuntimeError("SLACK_WEBHOOK_URL is not configured")

    remediable = is_remediable(finding.rule)
    header = (
        "*FinOps Alert: Waste Detected*"
        if remediable
        else "*FinOps Advisory: Possible Idle Resource*"
    )

    blocks: list[dict[str, Any]] = [
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": (
                    f"{header}\n\n"
                    f"*Rule:* {finding.rule}\n"
                    f"*Resource:* `{resource.resource_id}` ({resource.resource_type})\n"
                    # Region is not decoration: with several regions scanned, it is
                    # the first thing an approver needs to know where to look, and
                    # remediation runs there.
                    f"*Region:* `{resource.region}`\n"
                    f"*Cost Impact:* ${finding.est_monthly_cost_usd}/mo"
                ),
            },
        }
    ]

The region line was added after a real confusion. Scanning three regions
produces three near-identical alerts differing only by resource ID. Without the
region an approver cannot tell which account corner they are about to change.

The LLM summary goes in its own block

    if finding.llm_summary:
        # Advisor output is untrusted display copy (it summarizes
        # user-controlled tags), so it goes in its own context block and is
        # never used to build action values.
        blocks.append({
            "type": "context",
            "elements": [{"type": "mrkdwn", "text": finding.llm_summary}],
        })

A local LLM writes a plain-language explanation of each finding. The prompt
contains AWS tags, which anyone with tagging permission can write. So model
output is treated as untrusted display copy: rendered, never parsed, and
never used to build anything the system acts on.

Buttons are conditional, and that is the whole design

    if remediable:
        blocks.append({
            "type": "actions",
            "elements": [
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Approve Remediation"},
                    "style": "primary",
                    "value": f"approve_{finding.id}",
                    "action_id": "approve_remediation",
                },
                {
                    "type": "button",
                    "text": {"type": "plain_text", "text": "Deny"},
                    "style": "danger",
                    "value": f"deny_{finding.id}",
                    "action_id": "deny_remediation",
                },
            ],
        })
    else:
        # Metric-inferred: no playbook is allowed to run, so offering an
        # Approve button would promise an action the domain refuses.
        blocks.append({
            "type": "context",
            "elements": [{
                "type": "mrkdwn",
                "text": ("_Advisory only: inferred from CloudWatch metrics. "
                         "No automated remediation is available for this rule._"),
            }],
        })

Some findings are advisory by design. An idle EC2 instance might be a warm
standby, a batch worker between runs, or a license server, and low CPU is
evidence, not proof. The domain refuses to remediate those rules.

The early version rendered buttons on every alert. Clicking Approve on an
advisory finding returned a refusal. A button that does nothing is worse than
no button: it trains people to distrust every button. So the adapter asks the
domain first and renders accordingly.

Notification preview text

    response = WebhookClient(webhook_url).send(
        # Notification preview text. The region belongs here too, since
        # this is all a phone lock screen shows.
        text=(f"FinOps Alert: {finding.rule} on {resource.resource_id} "
              f"in {resource.region}"),
        blocks=blocks,
    )
    if response.status_code != 200:
        raise RuntimeError(f"Slack webhook returned {response.status_code}: {response.body}")

    logger.info("Sent Slack alert for finding %s", finding.id)
    # Incoming webhooks return no message timestamp; edits happen via the
    # interaction payload's response_url instead.
    return None

Easy to forget: when blocks is present, text becomes the push-notification
preview. Omit it and phones show "This content can't be displayed."

Note the raise. A failed send is not swallowed. The caller leaves the
finding unnotified so the next scan retries it. Swallowing would lose the alert
silently, which for a cost alert means the money keeps burning and nobody knows.

Part 3: Receiving the click safely

Verify before you parse

def parse_callback(
    self, raw_body: bytes, headers: Mapping[str, str]
) -> tuple[Decision, dict[str, Any]]:
    self._verify_signature(raw_body, headers)     # ← first line, always
    ...

Signature verification is the first thing that happens. Parse-then-verify
means you've already run a JSON decoder on hostile input.

CALLBACK_MAX_AGE_SECONDS = 60 * 5

def _verify_signature(self, raw_body: bytes, headers: Mapping[str, str]) -> None:
    secret = settings.slack_signing_secret
    if not secret:
        # No secret configured: verification bypassed (local testing).
        return

    timestamp = headers.get("x-slack-request-timestamp") or headers.get(
        "X-Slack-Request-Timestamp"
    )
    signature = headers.get("x-slack-signature") or headers.get("X-Slack-Signature")
    if not timestamp or not signature:
        raise PermissionError("Missing Slack signature headers")

    try:
        age = abs(time.time() - int(timestamp))
    except ValueError as exc:
        raise PermissionError("Invalid Slack timestamp header") from exc
    if age > CALLBACK_MAX_AGE_SECONDS:
        raise PermissionError("Slack request timestamp expired")

    verifier = SignatureVerifier(secret)
    if not verifier.is_valid(body=raw_body, timestamp=timestamp, signature=signature):
        raise PermissionError("Invalid Slack signature")

Three checks, three different attacks:

Check	Stops
Headers present	Casual probing of a public endpoint
Timestamp within 5 minutes	Replay: a captured valid request resent later
HMAC valid	Forgery

The replay check is the one people skip. A signature stays valid forever unless
you bound its age; capture one legitimate approve callback and you can replay it
indefinitely. Slack sends the timestamp precisely so you can reject stale
requests.

SignatureVerifier comes from slack_sdk and does constant-time comparison,
worth using rather than hand-rolling HMAC and comparing with ==.

The if not secret: return bypass is a deliberate local-testing affordance,
and it is a landmine. It means an unconfigured deployment accepts anything.
Fine on a laptop, dangerous anywhere reachable. If I were hardening this for
production I would make it fail closed unless an explicit
ALLOW_UNSIGNED_CALLBACKS=true were set.

Raw bytes matter

@app.post("/callbacks/{channel}")
async def notifier_callback(channel: str, request: Request) -> dict[str, Any]:
    raw_body = await request.body()          # ← bytes, not a parsed model
    decision, reply_context = notifier.parse_callback(raw_body, request.headers)

The HMAC is computed over the exact bytes Slack sent. Let FastAPI parse the
body into a Pydantic model first and you cannot reconstruct them: key order,
whitespace, and encoding all shift. Signature verification will then fail
mysteriously and you will lose an afternoon.

Parsing the payload

Slack sends application/x-www-form-urlencoded with a single payload field
containing JSON. Yes, really.

    form = urllib.parse.parse_qs(raw_body.decode("utf-8"))
    payload_values = form.get("payload")
    if not payload_values:
        raise ValueError("No payload found")

    try:
        payload = json.loads(payload_values[0])
    except json.JSONDecodeError as exc:
        raise ValueError("Payload is not valid JSON") from exc

    actions = payload.get("actions") or []
    if not actions:
        raise ValueError("No actions in payload")

    value = str(actions[0].get("value", ""))
    action: Literal["approve", "deny"]
    if value.startswith("approve_"):
        action, finding_id = "approve", value.removeprefix("approve_")
    elif value.startswith("deny_"):
        action, finding_id = "deny", value.removeprefix("deny_")
    else:
        raise ValueError(f"Unrecognized action value: {value!r}")

    actor = payload.get("user", {}).get("username") or payload.get("user", {}).get(
        "id", "unknown"
    )

    decision = Decision(
        finding_id=finding_id,
        actor=actor,
        action=action,
        decided_at=datetime.now(UTC),
        channel=self.channel_name,
    )
    reply_context = {
        "response_url": payload.get("response_url"),
        "original_blocks": payload.get("message", {}).get("blocks", []),
    }
    return decision, reply_context

The output is a domain object. Everything Slack-shaped (form encoding, the
payload wrapper, response_url) stops at this boundary. The domain service
receiving this Decision has no idea Slack exists.

The button value carries a system-generated finding ID, never model output
and never anything a user typed. The parser refuses anything not matching the
expected prefixes.

Mapping exceptions to status codes

    try:
        decision, reply_context = notifier.parse_callback(raw_body, request.headers)
    except PermissionError as exc:
        raise HTTPException(status_code=401, detail=str(exc)) from exc
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

The port defines the exception contract: PermissionError for auth failures,
ValueError for malformed input, so any notifier implementation maps cleanly
to HTTP without the route knowing which one is configured.

Part 4: What happens after the click

The route hands off to the domain immediately:

def _decide(finding_id: str, action: str, actor: str, channel: str) -> bool:
    repo = get_repository()
    if action == "approve":
        return approve_finding(
            finding_id,
            repo,
            # The resolver, not a gateway: the service picks the endpoint for
            # the finding's own region.
            get_cloud_gateway,
            actor=actor,
            channel=channel,
            dry_run=settings.dry_run,
        )
    return deny_finding(finding_id, repo, actor=actor, channel=channel)

Inside approve_finding, guardrails are re-checked at approval time, not
trusted from detection time:

    if resource.lifecycle == ResourceLifecycle.DELETED:
        _audit(repo, "approve_blocked_resource_gone", finding.id, {...})
        return False

    if finding.protected or rules.is_protected(resource.current_tags):
        _audit(repo, "approve_blocked_protected", finding.id, {...})
        return False

    if not rules.is_remediable(finding.rule):
        _audit(repo, "approve_blocked_notify_only", finding.id, {...})
        return False

    playbook = rules.PLAYBOOK_ALLOWLIST.get(resource.resource_type)
    if playbook is None:
        _audit(repo, "approve_blocked_no_playbook", finding.id, {...})
        return False

An alert can sit in Slack for hours. In that window someone may have tagged the
resource finops:protected=true, or deleted it out of band. The newer intent
wins, and each refusal gets its own audit event name so "why didn't it act?"
is answerable from the database.

The double-click problem

Slack buttons are trivially double-clickable, and Slack itself retries on
timeout. Without protection you get two remediations.

    if not repo.transition_finding(finding.id, finding.status, FindingStatus.APPROVED):
        return False  # lost the race, someone else already decided

Backed by a compare-and-swap:

UPDATE findings SET status = :new WHERE id = :id AND status = :expected

Returns True only when rowcount == 1. The second click finds the status
already APPROVED, matches nothing, updates nothing, returns False. At most
one remediation, structurally, not by luck of timing.

Editing the message

def confirm_decision(self, reply_context: dict[str, Any], text: str) -> None:
    response_url = reply_context.get("response_url")
    if not response_url:
        return

    blocks: list[dict[str, Any]] = []
    original_blocks = reply_context.get("original_blocks") or []
    if original_blocks:
        blocks.append(original_blocks[0])  # keep the alert text, drop the buttons
    blocks.append({"type": "section", "text": {"type": "mrkdwn", "text": text}})

    response = WebhookClient(response_url).send(
        text=text, blocks=blocks, replace_original=True
    )
    if response.status_code != 200:
        logger.error("Failed to update Slack message: %s", response.body)

Keeping block [0] and dropping the rest preserves the context (what the alert
was) while removing the buttons so a decided finding cannot be clicked again.
response_url is valid for 30 minutes and 5 uses, which is plenty for one edit.

The outcome text distinguishes every terminal state:

outcome = f"*Approved* by @{decision.actor}: remediation executed{where}."
outcome = f"*Approved* by @{decision.actor}: DRY RUN, no resources were changed."
outcome = f"*Denied* by @{decision.actor}: no action taken, finding closed."
outcome = f"*Remediation failed*{where} after approval by @{decision.actor}: ..."

"Approved" alone does not tell an operator whether anything actually changed.

Part 5: Testing without a workspace

The whole flow is testable with no Slack account at all, because the notifier is
a port:

class FakeNotifier(Notifier):
    """Records what was sent. Zero Slack, zero HTTP."""

    def __init__(self):
        self.alerts: list[tuple[str, str]] = []
        self.digests: list[tuple[str, list[str]]] = []

    @property
    def channel_name(self):
        return "fake"

    def send_finding_alert(self, finding, resource):
        self.alerts.append((finding.id, resource.resource_id))
        return f"msg-{len(self.alerts)}"

    def send_digest(self, title, sections):
        self.digests.append((title, sections))
        return f"digest-{len(self.digests)}"

The approve-and-remediate flow runs against FakeNotifier + a fake cloud
gateway + an in-memory repository. If that passes, swapping Slack for Telegram
cannot break the approval logic, because the approval logic never knew about
Slack.

For the adapter itself, mock the webhook client and assert on the blocks:

def send_and_capture(finding, resource):
    """Send an alert through a mocked webhook and return the blocks sent."""
    settings.slack_webhook_url = "https://hooks.slack.test/T/B/X"
    try:
        with patch("finops_sentinel.adapters.notifications.slack.WebhookClient") as client_cls:
            client_cls.return_value.send.return_value = MagicMock(status_code=200, body="ok")
            SlackAdapter().send_finding_alert(finding, resource)
            return client_cls.return_value.send.call_args.kwargs["blocks"]
    finally:
        settings.slack_webhook_url = None


def test_advisory_findings_get_no_buttons():
    blocks = send_and_capture(make_finding("ec2_idle"), resource)
    assert not any(b["type"] == "actions" for b in blocks)

That test encodes a product decision, not an implementation detail. It fails
if someone later adds buttons to advisory alerts, which is exactly when you want
to be interrupted.

Troubleshooting

Symptom	Cause
`401 Invalid Slack signature`	Secret mismatch, or the body was parsed before verification
Buttons do nothing	Request URL doesn't match your current tunnel; ngrok rotates on restart
`This content can't be displayed` on mobile	Missing `text=` fallback alongside `blocks=`
Timeouts	Slack wants 200 OK within 3 seconds; do slow work after responding
Message never edits	`response_url` expired (30 min / 5 uses)
Duplicate remediations	No compare-and-swap on the status transition

On the 3-second rule: this project's state transition is a single indexed
UPDATE, so it responds well inside the budget. If your remediation is slow,
acknowledge immediately and do the work in the background, otherwise Slack
retries, and now you're relying on idempotency you may not have.

What I would do differently

Fail closed on a missing signing secret. The bypass is convenient locally
and dangerous everywhere else.

Use a bot token instead of an incoming webhook. Webhooks return no message
timestamp, so editing depends on response_url and its 30-minute window. A bot
token gives you chat.update on any message, any time.

Add a confirmation dialog for high-impact actions. Block Kit supports a
native confirm object on buttons, one extra tap between a mis-tap and a
deleted database.

The takeaway

The Slack integration is about 260 lines. Roughly a third is Block Kit
formatting, a third is signature verification, and a third is turning payloads
into domain objects.

What makes it safe is not in the Slack code at all. It is that a button click
is a request, not a command: the domain re-checks every guardrail, uses an
atomic transition so a double-click cannot execute twice, and refuses outright
when the rule was never remediable.

Slack is the driving adapter. The safety lives inside.

Full source: github.com/boazleleina/finops-sentinel. See adapters/notifications/slack.py and SLACK_SETUP.md.