DEV Community: Boaz Leleina

Alembic in Practice: Structure, Revisions, and the Traps SQLite Sets

Boaz Leleina — Tue, 11 Aug 2026 22:14:03 +0000

What those hex filenames mean, why migrations silently applied to the wrong database for two months, and how to widen a CHECK constraint without quietly breaking it.

Alembic is the migration tool for SQLAlchemy. Most tutorials get you as far as
alembic revision --autogenerate and stop, which leaves you unprepared for the
two things that actually go wrong: SQLite's refusal to alter constraints, and
configuration that points migrations at a different database than your
application uses.

This post covers both, using real migrations from
FinOps Sentinel, an AWS cost
agent with four migrations, one of which took three attempts to get right.

The layout

alembic.ini                     # config; script location, logging
alembic/
├── env.py                      # runs on every alembic command
├── script.py.mako              # template for new revision files
└── versions/
    ├── c447645246ac_initial_schema.py
    ├── 08872b80af2e_notifications_and_remediations.py
    ├── a1c9f4d27b13_add_rds_and_s3_resource_types.py
    └── b7e2f81c4a90_add_spend_snapshots.py

Plus one table Alembic creates in your database:

CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL);

One row. One value. That is the entirety of Alembic's state: which database is
at which revision. Everything else is derived.

Why the filenames look like keyboard mashing

This is the most common first question, and the answer explains the design.

c447645246ac_initial_schema.py
b7e2f81c4a90_add_spend_snapshots.py

The hex prefix is the revision ID: a 12-character slice of a UUID, generated
by alembic revision. The filename is <revision_id>_<your message slugified>.py.

Inside each file:

revision: str = "b7e2f81c4a90"
down_revision: Union[str, None] = "a1c9f4d27b13"

Those two fields form a linked list:

c447645246ac  initial_schema                  (down_revision = None)  ← root
      ↓
08872b80af2e  notifications_and_remediations
      ↓
a1c9f4d27b13  add_rds_and_s3_resource_types
      ↓
b7e2f81c4a90  add_spend_snapshots                                     ← head

Order comes from the pointers, never from filenames. Look at the project's
list again: b7e2f81c4a90 sorts alphabetically before c447645246ac, yet
c447645246ac is the first migration and b7e2f81c4a90 is the newest. If
Alembic sorted by filename, everything would run backwards.

Why not 0001, 0002, 0003?

Because sequential numbers collide the moment two people work in parallel.

main
 ├── feature/billing   → 0004_add_invoices.py
 └── feature/auth      → 0004_add_sessions.py

Both are valid. Both are 0004. Merging produces a conflict where the numbering
lies about order, and worse, the conflict is in the filename, and git resolves it
by keeping both, and now your migration history has two fourth steps.

With hex IDs there is no collision. And if both branches set down_revision to
the same parent, Alembic detects a genuine branch:

$ alembic upgrade head
ERROR: Multiple head revisions are present

It refuses to guess, and you resolve it explicitly with alembic merge. The
scary-looking IDs are what make that detection possible.

If you want readable filenames

Add to alembic.ini:

file_template = %%(year)d%%(month).2d%%(day).2d_%%(rev)s_%%(slug)s

Giving 20260728_b7e2f81c4a90_add_spend_snapshots.py: date-sortable, revision
ID still present because Alembic needs it. Existing files keep their names; only
new ones pick it up.

The configuration trap that cost two months

This is the bug I most want to warn people about, because it produces no
error message at all.

The default env.py reads the database URL from alembic.ini. FinOps Sentinel
needed it configurable, so early on it did the obvious thing:

# alembic/env.py: the version with the bug
def get_url():
    db_path = os.getenv("SENTINEL_DB_PATH", ".sentinel.db")
    return f"sqlite:///{db_path}"

Reasonable. Except the application reads the same setting through
pydantic-settings:

class Settings(BaseSettings):
    sentinel_db_path: str = ".sentinel.db"
    model_config = SettingsConfigDict(env_file=".env")   # ← reads .env too

os.getenv does not read .env. pydantic-settings does.

The documented setup puts SENTINEL_DB_PATH=data/sentinel.db in .env, not in
the shell. So:

Component	Reads from	Resolves to
The app	env var → `.env` → default	`data/sentinel.db`
Alembic	env var → default	`.sentinel.db`

Every migration ever run landed on .sentinel.db. The application used
data/sentinel.db. Two files, silently, for two months.

Why nothing broke for so long

The application also calls Base.metadata.create_all() in tests, and the first
three migrations only created tables that SQLAlchemy would have created anyway.
The schemas happened to agree.

Then came a migration whose table nothing else creates: spend_snapshots. The
next scan died:

alembic upgrade head
INFO  [alembic.runtime.migration] Running upgrade a1c9f4d27b13 -> b7e2f81c4a90

sentinel scan
OperationalError: (sqlite3.OperationalError) no such table: spend_snapshots

Alembic reported success. The app said the table didn't exist. Both were telling
the truth about different files.

The fix: one owner for the URL

# config.py: the single source of truth
def database_url() -> str:
    """The one place the findings database URL is built.

    Both the application (bootstrap.get_repository) and Alembic (alembic/env.py)
    call this. They used to build it independently, and drifted.
    """
    return f"sqlite:///{settings.sentinel_db_path}"

# alembic/env.py
from finops_sentinel.config import database_url

def get_url():
    return database_url()

# bootstrap.py
def get_repository() -> FindingsRepository:
    return SqlAlchemyRepository(db_url=database_url())

And alembic.ini is deliberately emptied:

# Deliberately empty. env.py overrides this with the application's own
# Settings.sentinel_db_path. A URL here would be dead config that reads as
# authoritative, which is exactly how this drifted before.
sqlalchemy.url =

Takeaway: if your app and your migrations resolve the database path
separately, they will eventually disagree, and nothing will tell you. Give the
URL exactly one owner and have both call it.

The project now pins this with tests:

def test_the_app_and_alembic_build_the_url_the_same_way():
    assert get_repository().db_url == database_url()

def test_alembic_ini_does_not_hardcode_a_database():
    parser = configparser.ConfigParser()
    parser.read(REPO_ROOT / "alembic.ini")
    assert parser.get("alembic", "sqlalchemy.url", fallback="").strip() == ""

def test_alembic_env_does_not_read_the_environment_itself():
    env_py = (REPO_ROOT / "alembic" / "env.py").read_text()
    assert "database_url" in env_py
    assert "os.getenv(" not in env_py

Autogenerate, and where to stop trusting it

alembic revision --autogenerate -m "add spend snapshots"

Alembic compares your SQLAlchemy models against the live database and writes the
diff. It needs target_metadata wired up in env.py:

from finops_sentinel.adapters.persistence.sqlalchemy_repo import Base
target_metadata = Base.metadata

It detects new/dropped tables and columns, nullability changes, most index
and unique-constraint changes, and some type changes.

It misses table and column renames (it sees a drop plus an add, and will
happily destroy your data), CHECK constraint changes, server defaults in many
cases, and anything requiring data transformation.

The generated file is a draft. The comment Alembic itself writes says so:

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###

Always read it. For anything beyond adding a column, expect to hand-write.

Trap: SQLite cannot ALTER a constraint

Adding two values to an enum required widening a CHECK constraint:

class ResourceType(StrEnum):
    EBS_VOLUME = "ebs_volume"
    ELASTIC_IP = "elastic_ip"
    EC2_INSTANCE = "ec2_instance"
    EBS_SNAPSHOT = "ebs_snapshot"
    RDS_INSTANCE = "rds_instance"    # new
    S3_BUCKET = "s3_bucket"          # new

On Postgres: ALTER TABLE ... DROP CONSTRAINT ... ADD CONSTRAINT .... Done.

On SQLite: there is no ALTER for constraints. The only route is create a new
table, copy the data, drop the old, rename the new. Alembic wraps this in
batch_alter_table:

with op.batch_alter_table("resources", recreate="always") as batch_op:
    ...

Attempt 1: silently kept the old constraint

The first version passed table_args and ran without error:

with op.batch_alter_table(
    "resources",
    recreate="always",
    table_args=(
        sa.CheckConstraint(_resource_type_check(NEW_TYPES), name="check_resource_type"),
    ),
):
    pass

alembic upgrade head printed success. Inserting an rds_instance row still
failed.

Why: table_args adds to a reflected table definition rather than
replacing it. Alembic reflected the live table, including the old four-value
CHECK, then appended the new six-value one. The rebuilt table had both, and
the old one still rejected every new row.

Only dumping the schema revealed it:

sqlite3 data/sentinel.db ".schema resources"
# CONSTRAINT check_resource_type CHECK (resource_type IN ('ebs_volume', ...))   ← old
# CONSTRAINT check_resource_type CHECK (resource_type IN ('ebs_volume', ..., 's3_bucket'))

Attempt 2: `copy_from` to suppress reflection

The fix is copy_from, an explicit table definition that tells Alembic not
to reflect:

def _columns_only_resources() -> sa.Table:
    """The `resources` table as columns and primary key, with NO constraints.

    Handed to batch_alter_table as `copy_from`, which suppresses reflection of
    the live table. That suppression is the point: `table_args` adds to a
    reflected definition rather than replacing it.
    """
    return sa.Table(
        "resources",
        sa.MetaData(),
        sa.Column("id", sa.String(), primary_key=True, nullable=False),
        sa.Column("resource_id", sa.String(), nullable=False),
        sa.Column("resource_type", sa.String(), nullable=False),
        sa.Column("resource_arn", sa.String(), nullable=False),
        sa.Column("region", sa.String(), nullable=False),
        sa.Column("current_tags", sa.String(), nullable=False),
        sa.Column("lifecycle", sa.String(), nullable=False),
        sa.Column("first_seen_at", sa.DateTime(timezone=True), nullable=False),
        sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=False),
    )

Attempt 3: the indexes were gone

copy_from suppresses index reflection too. The rebuilt table lost its
unique index on resource_id, which the upsert logic depends on for
correctness. So it gets recreated by hand:

def _rebuild_resources(resource_types: tuple[str, ...]) -> None:
    with op.batch_alter_table(
        "resources",
        recreate="always",
        copy_from=_columns_only_resources(),
        table_args=(
            sa.CheckConstraint("lifecycle IN ('active', 'deleted')",
                               name="check_resource_lifecycle"),
            sa.CheckConstraint(_resource_type_check(resource_types),
                               name="check_resource_type"),
        ),
    ):
        pass  # the rebuild itself is the migration

    # copy_from also suppresses index reflection, so re-create by hand.
    bind = op.get_bind()
    existing = {index["name"] for index in sa.inspect(bind).get_indexes("resources")}
    if "ix_resources_resource_id" not in existing:
        op.create_index(op.f("ix_resources_resource_id"), "resources",
                        ["resource_id"], unique=True)

Rule for SQLite batch migrations: whatever the batch block omits is silently
absent from the rebuilt table. Every column, every constraint, every index has
to be re-declared. And "it ran without error" proves nothing. Dump the schema.

Downgrades that refuse to run

Most downgrade() functions are written on autopilot as the inverse of
upgrade(). Sometimes the honest inverse is refusing:

def downgrade() -> None:
    bind = op.get_bind()
    stranded = bind.execute(
        sa.text("SELECT COUNT(*) FROM resources "
                "WHERE resource_type IN ('rds_instance', 's3_bucket')")
    ).scalar_one()
    if stranded:
        raise RuntimeError(
            f"Cannot downgrade: {stranded} resource(s) use rds_instance or s3_bucket, "
            "which the previous CHECK constraint forbids. Delete those resources and "
            "the findings referencing them first if this downgrade is intended."
        )
    _rebuild_resources(OLD_RESOURCE_TYPES)

Narrowing the constraint with those rows present means deleting them, which
also orphans every finding referencing them. A downgrade that silently destroys
data to succeed is worse than one that stops and explains.

The docstring says so up front, so nobody is surprised at 2am:

downgrade() restores the four-value constraint and will FAIL if any
rds_instance or s3_bucket row exists by then, deliberately.

A simple migration, for contrast

Not every migration is a fight. Adding a table is exactly what you'd hope:

def upgrade() -> None:
    op.create_table(
        "spend_snapshots",
        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True, nullable=False),
        sa.Column("snapshot_date", sa.Date(), nullable=False),
        # Stored as a string, matching SafeNumeric: SQLite has no Decimal, and
        # a float column would quietly round money.
        sa.Column("total_estimated_monthly_usd", sa.String(), nullable=False),
        sa.Column("open_findings", sa.Integer(), nullable=False),
        sa.Column("active_resources", sa.Integer(), nullable=False),
        sa.Column("captured_at", sa.DateTime(timezone=True), nullable=False),
    )
    op.create_index(
        op.f("ix_spend_snapshots_snapshot_date"),
        "spend_snapshots", ["snapshot_date"], unique=True,
    )

Note the docstring convention this project uses: the why, not the what:

snapshot_date is UNIQUE on purpose. The repository upserts by date, so two
scans on the same day collapse to one row. Otherwise a day that happened to
be scanned five times would weigh five times as much in the trailing mean, and
the z-score would measure scan cadence rather than spend.

Six months later, that paragraph is worth more than the code.

Keeping constraints from drifting

CHECK constraints duplicating enum values is a drift hazard. This project
generates them:

def _in_clause(column: str, values: type[StrEnum]) -> str:
    """Render `column IN ('a', 'b', ...)` straight from a StrEnum.

    Generated rather than hand-written so the constraint cannot drift from the
    enum: adding a ResourceType now updates this automatically, and only the
    Alembic migration has to be written by hand.
    """
    members = ", ".join(f"'{member.value}'" for member in values)
    return f"{column} IN ({members})"


class ResourceModel(Base):
    __tablename__ = "resources"
    ...
    __table_args__ = (
        CheckConstraint(_in_clause("lifecycle", ResourceLifecycle),
                        name="check_resource_lifecycle"),
        CheckConstraint(_in_clause("resource_type", ResourceType),
                        name="check_resource_type"),
    )

The model side can never drift from the enum. Only the migration is
hand-written, which is unavoidable, since a migration must describe a transition between
two states, and only one of them exists in your code.

Testing migrations

A green alembic upgrade head proves almost nothing (see attempt 1). Two things
worth doing:

1. Round-trip against a copy of a real database.

cp data/sentinel.db /tmp/mig-test.db
SENTINEL_DB_PATH=/tmp/mig-test.db alembic upgrade head
SENTINEL_DB_PATH=/tmp/mig-test.db alembic downgrade -1
SENTINEL_DB_PATH=/tmp/mig-test.db alembic upgrade head

sqlite3 /tmp/mig-test.db "select count(*) from resources; select count(*) from findings;"
# 274 / 187, same as before

Real data, real row counts, and it catches data loss a fresh database never
would.

2. Assert the error message is useful.

An un-migrated database should not produce a raw SQL traceback:

def upsert_resource(self, resource: Resource) -> None:
    try:
        self._upsert_resource(resource)
    except IntegrityError as exc:
        if "check_resource_type" not in str(exc):
            raise
        raise RuntimeError(
            f"The findings database does not allow resource type "
            f"'{resource.resource_type}'. It was created by an older "
            f"version of FinOps Sentinel. Run `alembic upgrade head` "
            f"against {self.db_url} and scan again."
        ) from exc

Failing hard is right: a half-written inventory is worse than a stopped run.
But the traceback told nobody what to do. Now it names the type, the database,
and the command.

Command reference

alembic revision -m "message"                  # empty migration, hand-written
alembic revision --autogenerate -m "message"   # diff models vs database

alembic upgrade head          # apply everything outstanding
alembic upgrade +1            # one step forward
alembic downgrade -1          # one step back
alembic downgrade base        # all the way down

alembic current               # what revision is this database at?
alembic history --verbose     # the full chain
alembic heads                 # >1 means you have a branch to merge
alembic merge -m "merge" a1b2 c3d4

alembic upgrade head --sql    # print SQL instead of executing (offline mode)

alembic current is the one to reach for when something is confusing. It tells
you where Alembic thinks the database is, and if that disagrees with the
database you expect, you have the bug from earlier in this post.

Checklist

Never edit a migration that has been applied anywhere but your laptop. Write a new one. The old revision ID is a foreign key in every database that ran it.
Read autogenerated migrations before committing. Especially for renames, which autogenerate turns into drop-plus-add.
Write the "why" in the docstring. The SQL says what; only you can say why.
Test on a copy of real data. Fresh databases hide data-loss bugs.
Dump the schema after non-trivial migrations. "It ran" is not "it worked."
Give the database URL one owner that both the app and Alembic call.
Make downgrades honest, including refusing when the honest inverse is data loss.

Migrations are the one part of a codebase where a mistake is not just wrong but
permanently wrong, replicated to every environment that ran it. Worth the
extra ten minutes.

Every example here is from github.com/boazleleina/finops-sentinel

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 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.

It also has a second half that took me a round of review to see. Proving the
request came from Slack is not the same as proving the person may delete
infrastructure, and neither one is AWS agreeing that they may. Three different
questions, three different places to answer them: the adapter, the domain, and
IAM.

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; 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

And one attack none of them stops, which took me an embarrassingly long time to
see: a valid signature from someone who has no business deleting anything.
More on that in Part 4, where it turns out to be the more interesting hole.

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.

The signature proves the app, not the install

A signature says the request went through this app. It does not say which
install of it. Add the app to a second workspace, or widen the channel, and
well-signed approvals start arriving from a population nobody enumerated,
every one of them passing all three checks above.

That is Slack-shaped, so it belongs in the Slack adapter, right next to the
signature check:

def _verify_provenance(self, payload: dict[str, Any]) -> None:
    expected_team = settings.slack_team_id
    if expected_team:
        team_id = (payload.get("team") or {}).get("id")
        if team_id != expected_team:
            raise PermissionError(f"Callback from unexpected Slack workspace: {team_id!r}")

    allowed_channels = settings.allowed_slack_channels
    if allowed_channels:
        channel_id = (payload.get("channel") or {}).get("id")
        if channel_id not in allowed_channels:
            raise PermissionError(f"Callback from unexpected channel: {channel_id!r}")

team.id means nothing to a Telegram install (its equivalent is a chat id),
so this stays on the adapter side of the port. The question of whether the
person may delete infrastructure is different, and it goes somewhere else
entirely.

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,
            # A factory, not a gateway: the credentials depend on the approval
            # — its region, and whose role runs it.
            get_approval_gateway_factory(),
            actor=actor,
            channel=channel,
            dry_run=settings.dry_run,
            # Authority, checked in the domain: the notifier already proved the
            # request came through the app, which is a different question.
            authorizer=get_authorizer(),
        )
    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 authorizer is not None and not authorizer.can_approve(actor):
        _audit(repo, "approve_blocked_unauthorized", finding.id, {"actor": actor})
        return False

    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.

Authentication is not authorization

That first check is the one I originally didn't have, and its absence was the
most serious thing in this system.

Signature verification proves the request came through the app. The signing
secret is app-level, so that proof is shared by everyone who can see the
message. I captured actor, stamped it into the Decision, and printed it in
"Approved by @boaz", so the audit trail could answer who clicked, while the
system had never formed an opinion on whether that person was permitted to.

The check lives in the domain, not the adapter. approve_finding already
receives actor; putting the decision there means it survives the swap to
Telegram that the whole port arrangement exists to make possible. It gets its
own refusal path and its own audit event, because a denied attempt has to be as
visible in the log as an accepted one; otherwise the trail records only the
approvals that happened to be permitted.

The list is a config table (SENTINEL_APPROVERS), keyed on the Slack user
id rather than the username: display names are user-controlled, and this
string decides who gets to delete infrastructure.

The double-click problem, and the one I actually had

Slack buttons are trivially double-clickable, and Slack itself retries on
timeout. Without protection you get two remediations. Here is what I originally
wrote:

    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. That protects the simultaneous case:
two requests that both read NOTIFIED, one of which loses.

It does not protect a replay, and a double-click is a replay. The second
click arrives after the first one committed, so it loads APPROVED, passes
expected=APPROVED, and runs:

UPDATE findings SET status='APPROVED' WHERE id=:id AND status='APPROVED'

That matches. rowcount == 1. Remediation runs again.

What saved me was a check one gate above: the transition table has no
APPROVED → APPROVED edge, so a replay is refused before it reaches the CAS.
But that means the guarantee I was advertising came from a different mechanism
than the one I was pointing at, and it holds only for as long as nobody adds a
self-loop to that table. The fix is to name the pre-decision state literally:

    if not repo.transition_finding(finding.id, FindingStatus.NOTIFIED, FindingStatus.APPROVED):
        return False  # already decided — a concurrent click, or a replayed one

Now APPROVED → APPROVED updates zero rows, which is the behaviour I claimed in
the first place.

A concurrency test cannot catch this; a sequential one catches it immediately.
Approve once, let it commit, replay the identical payload, assert the fake cloud
gateway saw exactly one delete:

def test_replayed_approval_executes_once(repository):
    seed(repository)
    gateway = FakeCloudGateway()
    payload = {"actor": "boaz", "channel": "slack", "dry_run": False}

    assert approve_finding("f-mock", repository, resolver(gateway), **payload) is True
    assert approve_finding("f-mock", repository, resolver(gateway), **payload) is False

    assert gateway.executed == [("snapshot_then_delete_volume", "vol-123", False)]

Three seconds versus a snapshot

The version of this article that shipped first put the remediation inline in the
callback route, and buried the consequence in a troubleshooting note. The
consequence deserves better than a note.

The EBS playbook snapshots the volume and waits for the snapshot before
deleting (up to ten minutes). Slack gives up at three seconds and retries the
interaction itself. Across that entire interval the message still carried its
buttons, because the code that removes them ran after the playbook finished. So
the acknowledgement window, the button removal, and the CAS all stopped covering
the same interval, and the interval was as long as the remediation.

The fix is to split deciding from executing. Deciding is repository reads and
one write:

plan = commit_approval(finding_id, repo, actor=actor, channel=channel,
                       authorizer=get_authorizer())
if plan is None:
    return _rejected(...)          # a guardrail refused, or it was already decided

notifier.confirm_decision(reply_context, f"⏳ *Approved* by @{actor} — running "
                                         f"`{plan.playbook}`{where}.")
background_tasks.add_task(_run_remediation, plan, notifier, reply_context, actor, where)
return {"message": "accepted"}     # well inside the 3-second budget

The ordering is the point: the finding is APPROVED in the database before
the acknowledgement goes out. The state change that makes a replay a no-op and
the edit that removes the buttons land together, rather than a remediation
apart. The playbook then runs in the background and edits the message a second
time with the outcome.

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 @{actor} — running `{plan.playbook}`{where}."   # the ack
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.

The guardrail I couldn't write in Python

Everything above is Sentinel deciding whether Sentinel should act. None of it
is AWS deciding whether a person may delete a volume.

That distinction matters because the deletion runs with the app's own
credentials. An approver list is a string comparison; the DeleteVolume call
behind it is made by a principal that holds that permission permanently, for
every approval, whoever clicked. Someone on the list with zero IAM access still
deletes the volume. Someone with account admin but not on the list is refused.
The two facts are unrelated: a textbook confused deputy, where the click is
only a trigger.

Making AWS the enforcement point means the approval has to run as the
approver. SENTINEL_ASSUME_ROLE=true maps each actor to a role, and every
approval calls sts:AssumeRole with three things attached:

RoleSessionName=sentinel-U024BE7LH: CloudTrail records a session naming the human, not a shared service principal.
An ExternalId on the trust policy, so the role cannot be assumed by anything that doesn't hold the secret even if its ARN leaks.
An inline session policy scoped to this one approval. Sentinel already knows the playbook, the resource and the region, so the session it runs under can do ec2:DeleteVolume on exactly one volume ARN for fifteen minutes. STS intersects it with the role's own permissions, so it can only narrow: the approver's role may be broad, and the session that runs this deletion is not.

Sentinel's own role then drops to read-only plus sts:AssumeRole. A leaked
Sentinel credential can inventory the account and nothing else.

The limit is worth stating rather than glossing: the actor-to-role mapping is
Sentinel asserting an identity from a Slack payload. AWS enforces what the
resulting session may do; it never sees the Slack user. The chain is as strong
as the Slack account and the signature check in front of it. That is a real
improvement over a name list, and it is not the same thing as the approver
authenticating to AWS; "IAM enforces it" implies more than is true.

One testing note, because it is a trap: LocalStack Community evaluates no IAM
policies. It issues a session for any role ARN and then permits whatever that
session asks. A "denied" assertion there passes the deletion: a green test
for an enforcement that never ran. Test the wiring against LocalStack, test the
refusals against a fake STS, and verify enforcement once by hand in a sandbox
account.

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	CAS passing the status you just read instead of the literal pre-decision one
Well-signed approvals from strangers	No `team.id` / channel check, and no approver list

On the 3-second rule: the state transition is a single indexed UPDATE and
finishes well inside the budget, but the remediation does not, and that is
the part that matters. Commit the decision, acknowledge, remove the buttons,
then do the slow work in the background. Acknowledging late means Slack retries
into a message whose buttons are still live, and now you are 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.

Put the remediation on a durable queue. Backgrounding it inside the web
process closed the acknowledgement window, but a restart mid-remediation strands
the finding in APPROVED with no message update. Fine for one process; wrong
for anything that redeploys often.

Check authority on deny, too. Denying is the safe direction, so I gated
approve first, but denial is terminal, which means anyone who can see the
message can quietly close a finding nobody ever acts on.

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, asks
whether this actor may approve at all, uses a compare-and-swap against the
literal pre-decision status so neither a race nor a replay can execute twice,
and refuses outright when the rule was never remediable.

And the guarantee that matters most isn't in the code either. It is in IAM: the
session that runs the deletion belongs to the approver, is scoped to one
resource, and expires in fifteen minutes.

Slack is the driving adapter. The safety lives inside and, for the last mile,
in the account.

Corrections and thanks

This article has been revised. The original claimed the compare-and-swap made
double-clicks safe, and it did not: passing finding.status as the expected
value makes the update a tautology on a replay, and a double-click is a
replay, not a race. The original also treated signature verification as if it
answered a question about the person clicking, which it never did.

Both were pointed out by
anp2network in the comments: precisely, with
the failing sequence spelled out and the test that would have caught it. That
feedback prompted the fixes described above: the literal CAS precondition and
its sequential replay test, the domain-side authority check with its own audit
event, the workspace and channel provenance checks in the adapter, moving the
remediation off the request path, and ultimately the assume-role work that moved
enforcement from a name list into IAM.

The best kind of comment is the one that costs you a weekend. Thank you.

Full source: github.com/boazleleina/finops-sentinel; see adapters/notifications/slack.py, docs/iam-policies.md and SLACK_SETUP.md.

DEV Community: Boaz Leleina

Alembic in Practice: Structure, Revisions, and the Traps SQLite Sets

The layout

Why the filenames look like keyboard mashing

Why not 0001, 0002, 0003?

If you want readable filenames

The configuration trap that cost two months

Why nothing broke for so long

The fix: one owner for the URL

Autogenerate, and where to stop trusting it

Trap: SQLite cannot ALTER a constraint

Attempt 1: silently kept the old constraint

Attempt 2: copy_from to suppress reflection

Attempt 3: the indexes were gone

Downgrades that refuse to run

A simple migration, for contrast

Keeping constraints from drifting

Testing migrations

Command reference

Checklist

FinOps Sentinel: AWS Cost Protection Agent

Demo

The problem

What it does

The nine rules

Design: ports and adapters

How a scan works

The state machine

The safety model

Where the LLM fits (and where it does not)

Two rules worth explaining

Cost estimation, honestly

Testing

Three bugs that only appeared when I ran it

Running it

What I would build next

The takeaway

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

System Architecture

Building a Slack Approval Workflow That Deletes Cloud Infrastructure

The shape of the problem

Part 1: Setting up the Slack app

Create the app and get a webhook

Enable interactivity

Get the signing secret

Part 2: Sending a message worth acting on

The LLM summary goes in its own block

Buttons are conditional, and that is the whole design

Notification preview text

Part 3: Receiving the click safely

Verify before you parse

The signature proves the app, not the install

Raw bytes matter

Parsing the payload

Mapping exceptions to status codes

Part 4: What happens after the click

Authentication is not authorization

The double-click problem, and the one I actually had

Three seconds versus a snapshot

Editing the message

The guardrail I couldn't write in Python

Part 5: Testing without a workspace

Troubleshooting

What I would do differently

The takeaway

Corrections and thanks

Attempt 2: `copy_from` to suppress reflection