diff --git a/CHANGELOG.md b/CHANGELOG.md index dd8cbca6..f002e66e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ OpenShield uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### Added +- Add the remediation approval, idempotency, audit and rescan gate (`api/models/remediation.py`, `docs/remediation-gate.md`) for issue #266: approved-only, allowlisted, at-most-once execution grants with an append-only audit trail and rescan-based verification. Proposal-only; nothing executes a playbook. - Add all ten evidence-rich enterprise network and perimeter controls `AZ-NET-018` through `AZ-NET-027` for issue #253, preserving API failures and incomplete data as indeterminate. - Branch protection declared as versioned GitHub rulesets for `dev` and `main`, with a scheduled drift audit that retains evidence (#298) - CI and CodeQL post-merge runs on `dev` and `main` (#298) diff --git a/alembic/versions/b6d2f8a4c1e7_remediation_action_gate.py b/alembic/versions/b6d2f8a4c1e7_remediation_action_gate.py new file mode 100644 index 00000000..1cc1f3ad --- /dev/null +++ b/alembic/versions/b6d2f8a4c1e7_remediation_action_gate.py @@ -0,0 +1,164 @@ +"""Add the approval, idempotency, audit and rescan gate for remediation actions. + +Revision ID: b6d2f8a4c1e7 +Revises: 3a76ff935bf6 +Create Date: 2026-09-30 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision: str = "b6d2f8a4c1e7" +down_revision: Union[str, Sequence[str], None] = "3a76ff935bf6" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +_STATUS_CONSTRAINT = "ck_remediation_actions_status_v1" +_OPEN_INDEX = "uq_remediation_actions_one_open_per_finding" +_APPEND_ONLY_TRIGGER = "trg_remediation_audit_log_append_only" +_NO_TRUNCATE_TRIGGER = "trg_remediation_audit_log_no_truncate" +_APPEND_ONLY_FUNCTION = "remediation_audit_log_append_only" + +_STATUSES = ( + "PROPOSED", + "APPROVED", + "EXECUTING", + "EXECUTION_FAILED", + "PENDING_VERIFICATION", + "VERIFICATION_FAILED", + "VERIFIED", + "REJECTED", +) +# A finding may have at most one action that is still alive. EXECUTION_FAILED +# and VERIFICATION_FAILED are deliberately *not* terminal: they need a human +# decision, so a second proposal for the same finding cannot be created behind +# their back. Only VERIFIED and REJECTED release the finding. +_OPEN_STATUSES = ( + "PROPOSED", + "APPROVED", + "EXECUTING", + "EXECUTION_FAILED", + "PENDING_VERIFICATION", + "VERIFICATION_FAILED", +) + + +def _in_list(values: Sequence[str]) -> str: + return ", ".join(f"'{value}'" for value in values) + + +def upgrade() -> None: + op.create_table( + "remediation_actions", + sa.Column("action_id", postgresql.UUID(), nullable=False), + sa.Column("finding_id", sa.Integer(), nullable=False), + sa.Column("playbook", sa.Text(), nullable=False), + # What the approver is actually approving: the script's bytes at proposal + # time and the exact target. begin_execution recomputes the hash and + # refuses to run if the script changed after approval. + sa.Column("playbook_sha256", sa.Text(), nullable=False), + sa.Column("resource_id", sa.Text(), nullable=True), + sa.Column("subscription_id", sa.Text(), nullable=False), + sa.Column("status", sa.Text(), nullable=False, server_default=sa.text("'PROPOSED'")), + sa.Column("proposed_by", sa.Text(), nullable=False), + sa.Column("approved_by", sa.Text(), nullable=True), + sa.Column("executed_by", sa.Text(), nullable=True), + sa.Column("verification_scan_id", postgresql.UUID(), nullable=True), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP") + ), + sa.Column( + "updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP") + ), + sa.Column("approved_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("executed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("verified_at", sa.DateTime(timezone=True), nullable=True), + sa.ForeignKeyConstraint(["finding_id"], ["findings.id"], name="remediation_actions_finding_id_fkey"), + sa.ForeignKeyConstraint( + ["verification_scan_id"], ["scans.scan_id"], name="remediation_actions_verification_scan_id_fkey" + ), + sa.PrimaryKeyConstraint("action_id", name="remediation_actions_pkey"), + ) + op.create_check_constraint( + _STATUS_CONSTRAINT, + "remediation_actions", + f"status IN ({_in_list(_STATUSES)})", + ) + op.create_index("idx_remediation_actions_finding_id", "remediation_actions", ["finding_id"], unique=False) + op.create_index("idx_remediation_actions_status", "remediation_actions", ["status"], unique=False) + # The database, not the caller, guarantees one live action per finding, so + # two concurrent proposals cannot both win. + # The table was created empty two statements ago, so a plain (blocking) + # index build costs nothing and keeps this revision transactional. + op.execute( + f""" + CREATE UNIQUE INDEX {_OPEN_INDEX} + ON remediation_actions (finding_id) + WHERE status IN ({_in_list(_OPEN_STATUSES)}) + """ + ) + + op.create_table( + "remediation_audit_log", + sa.Column("audit_id", sa.BigInteger(), autoincrement=True, nullable=False), + sa.Column("action_id", postgresql.UUID(), nullable=False), + sa.Column("event", sa.Text(), nullable=False), + sa.Column("from_status", sa.Text(), nullable=True), + sa.Column("to_status", sa.Text(), nullable=True), + sa.Column("actor", sa.Text(), nullable=False), + sa.Column("detail", postgresql.JSONB(), nullable=False, server_default=sa.text("'{}'::jsonb")), + sa.Column( + "occurred_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("CURRENT_TIMESTAMP") + ), + sa.ForeignKeyConstraint( + ["action_id"], ["remediation_actions.action_id"], name="remediation_audit_log_action_id_fkey" + ), + sa.PrimaryKeyConstraint("audit_id", name="remediation_audit_log_pkey"), + ) + op.create_index( + "idx_remediation_audit_log_action_id", "remediation_audit_log", ["action_id", "audit_id"], unique=False + ) + + # The audit trail is only evidence if it cannot be rewritten afterwards. + op.execute( + f""" + CREATE FUNCTION {_APPEND_ONLY_FUNCTION}() RETURNS trigger AS $$ + BEGIN + RAISE EXCEPTION 'remediation_audit_log is append-only'; + END; + $$ LANGUAGE plpgsql + """ + ) + op.execute( + f""" + CREATE TRIGGER {_APPEND_ONLY_TRIGGER} + BEFORE UPDATE OR DELETE ON remediation_audit_log + FOR EACH ROW EXECUTE FUNCTION {_APPEND_ONLY_FUNCTION}() + """ + ) + # Row-level triggers never fire for TRUNCATE, which would otherwise empty the + # trail in one statement, so it needs its own statement-level trigger. + op.execute( + f""" + CREATE TRIGGER {_NO_TRUNCATE_TRIGGER} + BEFORE TRUNCATE ON remediation_audit_log + FOR EACH STATEMENT EXECUTE FUNCTION {_APPEND_ONLY_FUNCTION}() + """ + ) + + +def downgrade() -> None: + op.execute(f"DROP TRIGGER IF EXISTS {_NO_TRUNCATE_TRIGGER} ON remediation_audit_log") + op.execute(f"DROP TRIGGER IF EXISTS {_APPEND_ONLY_TRIGGER} ON remediation_audit_log") + op.execute(f"DROP FUNCTION IF EXISTS {_APPEND_ONLY_FUNCTION}()") + op.drop_index("idx_remediation_audit_log_action_id", table_name="remediation_audit_log") + op.drop_table("remediation_audit_log") + op.drop_index(_OPEN_INDEX, table_name="remediation_actions") + op.drop_index("idx_remediation_actions_status", table_name="remediation_actions") + op.drop_index("idx_remediation_actions_finding_id", table_name="remediation_actions") + op.drop_constraint(_STATUS_CONSTRAINT, "remediation_actions", type_="check") + op.drop_table("remediation_actions") diff --git a/api/models/remediation.py b/api/models/remediation.py new file mode 100644 index 00000000..e03eda77 --- /dev/null +++ b/api/models/remediation.py @@ -0,0 +1,557 @@ +"""Approval, idempotency, audit and rescan gate for agent-driven remediation (issue #266). + +Running a ``playbooks/cli`` script against a real subscription has real blast +radius: a bad proposal or a retry can change customer infrastructure. This +module is the gate every execution-adjacent path must go through. It is +deliberately narrow: it records proposals, enforces who may advance them, and +decides whether a remediation actually worked. It never runs a playbook. + +Lifecycle:: + + PROPOSED -> APPROVED -> EXECUTING -> PENDING_VERIFICATION -> VERIFIED + | | | | + | | v v + | | EXECUTION_FAILED VERIFICATION_FAILED + v v + REJECTED REJECTED + +Guarantees, each enforced by a single conditional ``UPDATE`` or a database +constraint rather than a check-then-write in Python: + +* **Approval is mandatory.** ``begin_execution`` only matches ``APPROVED`` rows. +* **Nothing runs twice.** ``APPROVED -> EXECUTING`` can be won by exactly one + caller. A crash mid-execution leaves the row ``EXECUTING`` and it is never + retried automatically: at-most-once, because an unknown outcome on customer + infrastructure needs a human, not a second attempt. +* **Nothing runs unless allowlisted.** ``EXECUTION_ALLOWLIST`` starts empty. +* **Approval covers exactly what will run.** ``propose`` stores the script's + SHA-256 and the target; the ``approved`` audit row records both, and + ``begin_execution`` recomputes the hash and refuses if the script changed. +* **Approval is race-free.** Two concurrent approvals of one ``PROPOSED`` row + cannot both transition it; the loser is told the row already moved on. +* **One live action per finding.** A partial unique index backs ``propose``. +* **Done means rescanned.** ``VERIFIED`` requires a later completed scan whose + rule evaluation for the same resource is ``PASS``. A missing, ``UNKNOWN`` or + ``ERROR`` result is inconclusive and leaves the action open, and a ``FAIL`` + moves it to ``VERIFICATION_FAILED``. Neither closes the finding. +* **Every step is audited** in an append-only table, in the same transaction + as the state change it describes. +""" + +import hashlib +import json +import logging +import uuid +from pathlib import Path +from typing import Any, Callable, Dict, FrozenSet, Iterable, List, Optional, Tuple, Union + +import psycopg2.extras + +from api.models.finding import DatabaseManager +from scanner.evaluation import EvaluationStatus, subscription_scope_id + +logger = logging.getLogger(__name__) + +PROPOSED = "PROPOSED" +APPROVED = "APPROVED" +EXECUTING = "EXECUTING" +EXECUTION_FAILED = "EXECUTION_FAILED" +PENDING_VERIFICATION = "PENDING_VERIFICATION" +VERIFICATION_FAILED = "VERIFICATION_FAILED" +VERIFIED = "VERIFIED" +REJECTED = "REJECTED" + +# Statuses that keep a finding locked to its action. Mirrors the partial unique +# index in migration b6d2f8a4c1e7. +OPEN_STATUSES: FrozenSet[str] = frozenset( + {PROPOSED, APPROVED, EXECUTING, EXECUTION_FAILED, PENDING_VERIFICATION, VERIFICATION_FAILED} +) +_REJECTABLE = (PROPOSED, APPROVED, EXECUTION_FAILED, VERIFICATION_FAILED) +_VERIFIABLE = (PENDING_VERIFICATION, VERIFICATION_FAILED) + +# Playbooks eligible for any automated execution. Empty on purpose: phase one +# ships proposal generation only. Adding an entry is a reviewed code change, not +# configuration, so the set of things an agent may ever run stays auditable. +EXECUTION_ALLOWLIST: FrozenSet[str] = frozenset() + +# Repository root that playbook paths (``playbooks/cli/fix_.sh``) are relative to. +_REPO_ROOT = Path(__file__).resolve().parents[2] + + +class RemediationError(RuntimeError): + """Base class for gate refusals.""" + + +class RemediationNotFound(RemediationError): + """The finding or action does not exist.""" + + +class InvalidTransition(RemediationError): + """The action is not in a state that permits the requested step.""" + + def __init__(self, action_id: str, requested: str, current: str) -> None: + super().__init__(f"Action {action_id} cannot {requested} from status {current}") + self.action_id = action_id + self.requested = requested + self.current = current + + +class NotAllowlisted(RemediationError): + """The playbook is not eligible for automated execution.""" + + +class PlaybookUnavailable(RemediationError): + """The playbook file is missing, not a regular file, or outside ``playbooks/``.""" + + +class PlaybookChanged(RemediationError): + """The playbook's contents differ from what was approved.""" + + +def _require_actor(actor: str, role: str) -> str: + actor = (actor or "").strip() + if not actor: + raise ValueError(f"{role} identity is required") + return actor + + +def _action_detail(action: Dict[str, Any]) -> Dict[str, Any]: + """What an approver is approving, recorded verbatim in the audit trail.""" + return { + "finding_id": action["finding_id"], + "playbook": action["playbook"], + "playbook_sha256": action["playbook_sha256"], + "resource_id": action["resource_id"], + "subscription_id": action["subscription_id"], + } + + +def playbook_sha256(playbook: str, root: Path = _REPO_ROOT) -> str: + """Return the SHA-256 of a playbook's bytes. + + The stored playbook value comes from the database, so it is never trusted + as a path: it must resolve to a regular file under ``/playbooks``. + """ + playbooks_dir = (root / "playbooks").resolve() + candidate = (root / playbook).resolve() + if not candidate.is_relative_to(playbooks_dir) or not candidate.is_file(): + raise PlaybookUnavailable(f"Playbook {playbook!r} is not a file under playbooks/") + return hashlib.sha256(candidate.read_bytes()).hexdigest() + + +class RemediationGate: + """Durable state machine for remediation actions, backed by PostgreSQL.""" + + def __init__( + self, + db: DatabaseManager, + allowlist: Optional[Iterable[str]] = None, + playbook_root: Optional[Path] = None, + ) -> None: + self._db = db + self._playbook_root = _REPO_ROOT if playbook_root is None else Path(playbook_root) + self._allowlist: FrozenSet[str] = EXECUTION_ALLOWLIST if allowlist is None else frozenset(allowlist) + + def is_allowlisted(self, playbook: str) -> bool: + return playbook in self._allowlist + + # ------------------------------------------------------------------ # + # Internals # + # ------------------------------------------------------------------ # + + @staticmethod + def _audit( + cur: Any, + action_id: str, + event: str, + actor: str, + from_status: Optional[str], + to_status: Optional[str], + detail: Optional[Dict[str, Any]] = None, + ) -> None: + cur.execute( + """ + INSERT INTO remediation_audit_log (action_id, event, from_status, to_status, actor, detail) + VALUES (%s, %s, %s, %s, %s, %s::jsonb) + """, + (action_id, event, from_status, to_status, actor, json.dumps(detail or {}, default=str)), + ) + + @staticmethod + def _current_status(cur: Any, action_id: str) -> str: + cur.execute("SELECT status FROM remediation_actions WHERE action_id = %s", (action_id,)) + row = cur.fetchone() + if row is None: + raise RemediationNotFound(f"Remediation action {action_id} not found") + return row["status"] + + def _transition( + self, + action_id: str, + *, + event: str, + actor: str, + from_statuses: Tuple[str, ...], + to_status: str, + assignments: str = "", + params: Tuple[Any, ...] = (), + extra_where: str = "", + where_params: Tuple[Any, ...] = (), + detail: Union[Dict[str, Any], Callable[[Dict[str, Any]], Dict[str, Any]], None] = None, + ) -> Dict[str, Any]: + """Move one action between statuses atomically and audit it. + + The ``WHERE status = ANY(...)`` clause is the entire concurrency + control: of any number of simultaneous callers exactly one matches the + row, and the rest see zero rows updated. + """ + conn = self._db._get_conn() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + # FOR UPDATE in the CTE captures the status this caller is + # replacing, so the audit row records the real predecessor when + # several source statuses are allowed. + cur.execute( + f""" + WITH old AS ( + SELECT action_id, status FROM remediation_actions WHERE action_id = %s FOR UPDATE + ) + UPDATE remediation_actions AS a + SET status = %s, updated_at = CURRENT_TIMESTAMP{assignments} + FROM old + WHERE a.action_id = old.action_id AND a.status = ANY(%s){extra_where} + RETURNING a.*, old.status AS previous_status + """, # nosec B608 - assignments/extra_where are fixed fragments chosen by this module's own steps + (action_id, to_status, *params, list(from_statuses), *where_params), + ) + row = cur.fetchone() + if row is None: + current = self._current_status(cur, action_id) + conn.rollback() + raise InvalidTransition(action_id, event, current) + row = dict(row) + previous = row.pop("previous_status") + self._audit( + cur, action_id, event, actor, previous, to_status, detail(row) if callable(detail) else detail + ) + conn.commit() + return row + except Exception: + self._db.rollback(conn) + raise + + # ------------------------------------------------------------------ # + # Steps # + # ------------------------------------------------------------------ # + + def propose(self, finding_id: int, proposed_by: str) -> Tuple[Dict[str, Any], str]: + """Record a proposal for a finding's playbook. + + Returns ``(action, outcome)`` where outcome is ``created`` or + ``existing``. A finding has at most one live action, so a retried or + concurrent proposal converges on the same row instead of duplicating it. + """ + proposed_by = _require_actor(proposed_by, "Proposer") + conn = self._db._get_conn() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + """ + SELECT f.id, f.playbook, f.resource_id, s.subscription_id + FROM findings f JOIN scans s ON s.scan_id = f.scan_id + WHERE f.id = %s + """, + (finding_id,), + ) + finding = cur.fetchone() + if finding is None: + raise RemediationNotFound(f"Finding {finding_id} not found") + playbook = (finding["playbook"] or "").strip() + if not playbook: + raise RemediationError(f"Finding {finding_id} has no playbook to propose") + # Hashed here, before any row exists, so an unreadable or + # out-of-tree script can never be proposed, let alone approved. + digest = playbook_sha256(playbook, self._playbook_root) + + open_statuses = sorted(OPEN_STATUSES) + cur.execute( + """ + INSERT INTO remediation_actions + (action_id, finding_id, playbook, playbook_sha256, resource_id, subscription_id, + status, proposed_by) + VALUES (%s, %s, %s, %s, %s, %s, 'PROPOSED', %s) + ON CONFLICT (finding_id) WHERE status = ANY(%s) DO NOTHING + RETURNING * + """, + ( + str(uuid.uuid4()), + finding_id, + playbook, + digest, + finding["resource_id"], + finding["subscription_id"], + proposed_by, + open_statuses, + ), + ) + created = cur.fetchone() + if created is not None: + self._audit( + cur, + str(created["action_id"]), + "proposed", + proposed_by, + None, + PROPOSED, + _action_detail(created), + ) + conn.commit() + return dict(created), "created" + + cur.execute( + "SELECT * FROM remediation_actions WHERE finding_id = %s AND status = ANY(%s)", + (finding_id, open_statuses), + ) + existing = cur.fetchone() + if existing is None: + raise RuntimeError("remediation proposal conflict did not return an existing action") + conn.commit() + return dict(existing), "existing" + except Exception: + self._db.rollback(conn) + raise + + def approve(self, action_id: str, approved_by: str) -> Dict[str, Any]: + """Approve a proposal. ``PROPOSED -> APPROVED``, won by exactly one caller.""" + approved_by = _require_actor(approved_by, "Approver") + return self._transition( + action_id, + event="approved", + actor=approved_by, + from_statuses=(PROPOSED,), + to_status=APPROVED, + assignments=", approved_by = %s, approved_at = CURRENT_TIMESTAMP", + params=(approved_by,), + detail=_action_detail, + ) + + def reject(self, action_id: str, rejected_by: str, reason: str = "") -> Dict[str, Any]: + """Close an action without executing it, or after a failed attempt.""" + rejected_by = _require_actor(rejected_by, "Rejecter") + return self._transition( + action_id, + event="rejected", + actor=rejected_by, + from_statuses=_REJECTABLE, + to_status=REJECTED, + detail={"reason": reason}, + ) + + def begin_execution(self, action_id: str, executor: str) -> Dict[str, Any]: + """Grant permission to execute an approved, allowlisted action, once. + + The returned row is the execution grant. A second call for the same + action, concurrent or retried, raises :class:`InvalidTransition` and + never returns a grant. + """ + executor = _require_actor(executor, "Executor") + conn = self._db._get_conn() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + "SELECT playbook, playbook_sha256, status FROM remediation_actions WHERE action_id = %s", + (action_id,), + ) + row = cur.fetchone() + if row is None: + raise RemediationNotFound(f"Remediation action {action_id} not found") + if not self.is_allowlisted(row["playbook"]): + self._audit( + cur, + action_id, + "execution_refused", + executor, + row["status"], + row["status"], + {"reason": "playbook_not_allowlisted", "playbook": row["playbook"]}, + ) + conn.commit() + raise NotAllowlisted(f"Playbook {row['playbook']} is not eligible for automated execution") + + # The approval covers the script the approver's audit row shows. + # If it changed since, or vanished, nothing is granted. + try: + current = playbook_sha256(row["playbook"], self._playbook_root) + except PlaybookUnavailable: + current = None + if current != row["playbook_sha256"]: + self._audit( + cur, + action_id, + "execution_refused", + executor, + row["status"], + row["status"], + { + "reason": "playbook_changed", + "playbook": row["playbook"], + "approved_sha256": row["playbook_sha256"], + "current_sha256": current, + }, + ) + conn.commit() + raise PlaybookChanged(f"Playbook {row['playbook']} changed after it was proposed") + conn.rollback() + except Exception: + self._db.rollback(conn) + raise + + return self._transition( + action_id, + event="execution_started", + actor=executor, + from_statuses=(APPROVED,), + to_status=EXECUTING, + assignments=", executed_by = %s, executed_at = CURRENT_TIMESTAMP", + params=(executor,), + ) + + def complete_execution( + self, action_id: str, executor: str, succeeded: bool, detail: Optional[Dict[str, Any]] = None + ) -> Dict[str, Any]: + """Record the outcome. Only the executor that holds the grant may do so. + + Success moves to ``PENDING_VERIFICATION``, never to ``VERIFIED``: + a script exiting cleanly is not evidence the finding is gone. + """ + executor = _require_actor(executor, "Executor") + return self._transition( + action_id, + event="execution_succeeded" if succeeded else "execution_failed", + actor=executor, + from_statuses=(EXECUTING,), + to_status=PENDING_VERIFICATION if succeeded else EXECUTION_FAILED, + extra_where=" AND a.executed_by = %s", + where_params=(executor,), + detail=detail, + ) + + def verify(self, action_id: str, scan_id: str, actor: str = "system") -> Tuple[Dict[str, Any], str]: + """Decide whether a remediation worked, using a later completed scan. + + Returns ``(action, outcome)``: ``verified``, ``failed`` or + ``inconclusive``. Only ``verified`` closes the action. The rescan must + be completed, cover the same subscription and have finished after the + execution, and its rule evaluation for the exact resource must be + ``PASS``. A missing, ``UNKNOWN`` or ``ERROR`` evaluation proves nothing + and is recorded as inconclusive rather than treated as cleared. + """ + actor = _require_actor(actor, "Verifier") + conn = self._db._get_conn() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT * FROM remediation_actions WHERE action_id = %s FOR UPDATE", (action_id,)) + action = cur.fetchone() + if action is None: + raise RemediationNotFound(f"Remediation action {action_id} not found") + if action["status"] not in _VERIFIABLE: + raise InvalidTransition(action_id, "verified", action["status"]) + + cur.execute( + """ + SELECT f.rule_id, f.resource_id, s.subscription_id + FROM findings f JOIN scans s ON s.scan_id = f.scan_id + WHERE f.id = %s + """, + (action["finding_id"],), + ) + finding = cur.fetchone() + if finding is None: + raise RemediationNotFound(f"Finding {action['finding_id']} not found") + + cur.execute( + "SELECT status, subscription_id, completed_at, completed_at > %s AS after_execution " + "FROM scans WHERE scan_id = %s", + (action["executed_at"], scan_id), + ) + scan = cur.fetchone() + if scan is None: + raise RemediationNotFound(f"Scan {scan_id} not found") + if ( + scan["status"] != "completed" + or scan["subscription_id"] != finding["subscription_id"] + or not scan["after_execution"] + ): + raise RemediationError( + f"Scan {scan_id} is not a completed rescan of this subscription after the execution" + ) + + resource_id = finding["resource_id"] or subscription_scope_id(finding["subscription_id"]) + cur.execute( + "SELECT status FROM rule_evaluations WHERE scan_id = %s AND rule_id = %s AND resource_id = %s", + (scan_id, finding["rule_id"], resource_id), + ) + evaluation = cur.fetchone() + observed = evaluation["status"] if evaluation else None + detail = {"scan_id": scan_id, "rule_id": finding["rule_id"], "observed": observed} + + if observed == EvaluationStatus.PASS: + outcome, to_status, event = "verified", VERIFIED, "verification_passed" + elif observed == EvaluationStatus.FAIL: + outcome, to_status, event = "failed", VERIFICATION_FAILED, "verification_failed" + else: + self._audit( + cur, action_id, "verification_inconclusive", actor, action["status"], action["status"], detail + ) + conn.commit() + return dict(action), "inconclusive" + + cur.execute( + """ + UPDATE remediation_actions + SET status = %s, updated_at = CURRENT_TIMESTAMP, verification_scan_id = %s, + verified_at = CASE WHEN %s = 'VERIFIED' THEN CURRENT_TIMESTAMP ELSE verified_at END + WHERE action_id = %s + RETURNING * + """, + (to_status, scan_id, to_status, action_id), + ) + updated = dict(cur.fetchone()) + self._audit(cur, action_id, event, actor, action["status"], to_status, detail) + conn.commit() + return updated, outcome + except Exception: + self._db.rollback(conn) + raise + + # ------------------------------------------------------------------ # + # Reads # + # ------------------------------------------------------------------ # + + def get_action(self, action_id: str) -> Optional[Dict[str, Any]]: + conn = self._db._get_conn() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute("SELECT * FROM remediation_actions WHERE action_id = %s", (action_id,)) + row = cur.fetchone() + conn.commit() + return dict(row) if row else None + except Exception: + self._db.rollback(conn) + raise + + def get_audit_trail(self, action_id: str) -> List[Dict[str, Any]]: + """Return every recorded step for an action, oldest first.""" + conn = self._db._get_conn() + try: + with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute( + "SELECT * FROM remediation_audit_log WHERE action_id = %s ORDER BY audit_id", + (action_id,), + ) + rows = [dict(r) for r in cur.fetchall()] + conn.commit() + return rows + except Exception: + self._db.rollback(conn) + raise diff --git a/docs/remediation-gate.md b/docs/remediation-gate.md new file mode 100644 index 00000000..393644d3 --- /dev/null +++ b/docs/remediation-gate.md @@ -0,0 +1,83 @@ +# Remediation Gate + +Running a `playbooks/cli` script against a real subscription has real blast +radius. A bad proposal or a retry can change customer infrastructure. The +remediation gate (`api/models/remediation.py`, issue #266) is the code every +execution-adjacent path must go through before any agent-driven remediation +exists. + +**Phase one is proposal-only.** The gate records proposals, enforces who may +advance them, and decides whether a remediation worked. It never runs a +playbook, and nothing in this repository calls `begin_execution`. No execution +path may be merged that bypasses it. + +## Lifecycle + +``` +PROPOSED -> APPROVED -> EXECUTING -> PENDING_VERIFICATION -> VERIFIED + | | | | + | | v v + | | EXECUTION_FAILED VERIFICATION_FAILED + v v + REJECTED REJECTED +``` + +`EXECUTION_FAILED` and `VERIFICATION_FAILED` can also be moved to `REJECTED` by +a human. Every transition is one conditional `UPDATE ... WHERE status = ANY(...)` +plus an audit row in the same transaction. + +## Guarantees + +| Requirement | How it is enforced | +| --- | --- | +| Mandatory human approval | `begin_execution` only matches `APPROVED` rows; `approve` records who approved and when. | +| Typed allowlist, starts empty | `EXECUTION_ALLOWLIST` in code. Adding a playbook is a reviewed change, not configuration. A refused attempt is audited. | +| A retry never executes twice | `APPROVED -> EXECUTING` can be won by exactly one caller; every other call raises `InvalidTransition` and gets no grant. | +| Race-free approval | Two concurrent approvals of one `PROPOSED` row cannot both succeed. | +| One live action per finding | Partial unique index `uq_remediation_actions_one_open_per_finding`. Only `VERIFIED` and `REJECTED` release a finding. | +| Full audit log | `remediation_audit_log` records proposal, approval, execution and verification. Triggers reject `UPDATE`, `DELETE` and `TRUNCATE`. | +| Approval covers exactly what runs | `propose` stores the playbook's SHA-256 and the target (`resource_id`, `subscription_id`). The `approved` audit row records all of it. `begin_execution` recomputes the hash and refuses, with an audited `playbook_changed` refusal, if the script changed or disappeared. | +| Done means rescanned | `complete_execution(succeeded=True)` moves to `PENDING_VERIFICATION`, never `VERIFIED`. See below. | + +## What the audit trail does and does not prove + +Triggers stop the application, and anyone using normal SQL, from editing, +deleting or truncating audit rows. They do not stop the table owner or a +superuser, who can drop the trigger. If the trail must also be evidence against +a database operator, it needs a hash chain whose head is published outside the +database, and the application's database role should not hold +`UPDATE`/`DELETE`/`TRUNCATE` on the table. Both depend on how OpenShield is +deployed and are left out of phase one. + +Playbook paths come from the database, so they are never trusted as paths: +`propose` only accepts a regular file that resolves under `playbooks/`. + +## Execution is at-most-once + +If a runner dies after `begin_execution`, the action stays `EXECUTING` and is +never retried automatically. The outcome on customer infrastructure is unknown, +so recovery is a human decision (`reject`, then a new proposal). + +## Verification + +`verify(action_id, scan_id)` accepts a later scan only if it is `completed`, +covers the same subscription, and finished after the execution. It then reads +the rule evaluation (#263) for the same rule and resource: + +| Rescan evaluation | Result | +| --- | --- | +| `PASS` | `VERIFIED`: the action closes and the finding is released | +| `FAIL` | `VERIFICATION_FAILED`: stays open, a new proposal is refused | +| missing, `UNKNOWN`, `ERROR`, `NOT_APPLICABLE` | `inconclusive`: stays `PENDING_VERIFICATION` | + +Absence of a finding is not evidence of a fix, so a rule that did not evaluate +the resource can never close an action. A `VERIFICATION_FAILED` action can still +be verified by a later rescan. + +## Not yet in scope + +- HTTP routes and UI for proposing and approving. +- Generating proposals from findings automatically. +- Separation of duties (requiring approver and proposer to differ). +- Authentication of the `actor` strings; callers pass an identity the API layer + has already authenticated. diff --git a/tests/test_remediation_gate_postgres.py b/tests/test_remediation_gate_postgres.py new file mode 100644 index 00000000..9f4698c5 --- /dev/null +++ b/tests/test_remediation_gate_postgres.py @@ -0,0 +1,622 @@ +"""PostgreSQL-backed tests for the remediation approval/idempotency/audit/rescan gate (#266).""" + +import hashlib +import os +import threading +import uuid +from datetime import datetime, timedelta, timezone + +import psycopg2 +import pytest + +from api.models.finding import DatabaseManager +from api.models.remediation import ( + APPROVED, + EXECUTION_ALLOWLIST, + EXECUTION_FAILED, + EXECUTING, + OPEN_STATUSES, + PENDING_VERIFICATION, + PROPOSED, + REJECTED, + VERIFICATION_FAILED, + VERIFIED, + InvalidTransition, + NotAllowlisted, + PlaybookChanged, + PlaybookUnavailable, + RemediationError, + RemediationGate, + RemediationNotFound, +) + + +pytestmark = pytest.mark.skipif( + not os.environ.get("DATABASE_URL"), reason="DATABASE_URL is required for PostgreSQL tests" +) + +PLAYBOOK = "playbooks/cli/fix_az_test_001.sh" +SCRIPT_BODY = b"#!/usr/bin/env bash\nset -euo pipefail\naz storage account update --help\n" +RULE_ID = "AZ-TEST-001" + + +@pytest.fixture() +def dsn() -> str: + return os.environ["DATABASE_URL"] + + +@pytest.fixture() +def playbook_root(tmp_path): + """A throwaway repo root holding the one playbook the tests propose.""" + script = tmp_path / PLAYBOOK + script.parent.mkdir(parents=True) + script.write_bytes(SCRIPT_BODY) + return tmp_path + + +@pytest.fixture() +def world(dsn): + """Create scans/findings for one test and remove every trace afterwards.""" + created = {"scans": [], "findings": []} + subscription_id = str(uuid.uuid4()) + + def scan(completed_at=None, status="completed", subscription=None) -> str: + scan_id = str(uuid.uuid4()) + now = datetime.now(timezone.utc) + with psycopg2.connect(dsn) as conn, conn.cursor() as cur: + cur.execute( + "INSERT INTO scans (scan_id, subscription_id, started_at, completed_at, status) " + "VALUES (%s, %s, %s, %s, %s)", + (scan_id, subscription or subscription_id, now, completed_at or now, status), + ) + created["scans"].append(scan_id) + return scan_id + + def finding(scan_id: str, playbook=PLAYBOOK, resource_id="/subscriptions/x/resource/1") -> int: + with psycopg2.connect(dsn) as conn, conn.cursor() as cur: + cur.execute( + "INSERT INTO findings (scan_id, rule_id, rule_name, severity, resource_id, playbook, " + "detected_at, finding_key) VALUES (%s, %s, 'Test rule', 'HIGH', %s, %s, CURRENT_TIMESTAMP, %s) " + "RETURNING id", + (scan_id, RULE_ID, resource_id, playbook, uuid.uuid4().hex), + ) + finding_id = cur.fetchone()[0] + created["findings"].append(finding_id) + return finding_id + + def evaluation(scan_id: str, status: str, resource_id="/subscriptions/x/resource/1") -> None: + reason_code = None if status in ("PASS", "FAIL") else "TEST_REASON" + with psycopg2.connect(dsn) as conn, conn.cursor() as cur: + cur.execute( + "INSERT INTO rule_evaluations (scan_id, rule_id, resource_id, status, reason_code, evaluated_at) " + "VALUES (%s, %s, %s, %s, %s, CURRENT_TIMESTAMP)", + (scan_id, RULE_ID, resource_id, status, reason_code), + ) + + class World: + pass + + w = World() + w.subscription_id = subscription_id + w.scan, w.finding, w.evaluation = scan, finding, evaluation + yield w + + with psycopg2.connect(dsn) as conn, conn.cursor() as cur: + # The audit log rejects UPDATE, DELETE and TRUNCATE on purpose. Only this + # teardown switches its triggers off, inside one transaction, to tidy up. + cur.execute("ALTER TABLE remediation_audit_log DISABLE TRIGGER USER") + cur.execute("DELETE FROM remediation_audit_log") + cur.execute("ALTER TABLE remediation_audit_log ENABLE TRIGGER USER") + cur.execute("DELETE FROM remediation_actions") + cur.execute("DELETE FROM rule_evaluations WHERE scan_id = ANY(%s::uuid[])", (created["scans"],)) + cur.execute("DELETE FROM findings WHERE id = ANY(%s)", (created["findings"],)) + cur.execute("DELETE FROM scans WHERE scan_id = ANY(%s::uuid[])", (created["scans"],)) + + +@pytest.fixture() +def gate(dsn, playbook_root): + db = DatabaseManager(dsn) + yield RemediationGate(db, allowlist={PLAYBOOK}, playbook_root=playbook_root) + db.close() + + +def _run_concurrently(dsn: str, n: int, fn, playbook_root): + """Run fn(gate, index) on n threads, each with its own connection, released together.""" + barrier = threading.Barrier(n) + results: list = [None] * n + + def worker(i: int) -> None: + db = DatabaseManager(dsn) + try: + barrier.wait(timeout=10) + results[i] = ("ok", fn(RemediationGate(db, allowlist={PLAYBOOK}, playbook_root=playbook_root), i)) + except Exception as exc: # noqa: BLE001 - the exception type is the result under test + results[i] = ("err", exc) + finally: + db.close() + + threads = [threading.Thread(target=worker, args=(i,)) for i in range(n)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + return results + + +def _approved_action(gate, world): + finding_id = world.finding(world.scan()) + action, _ = gate.propose(finding_id, "agent") + return gate.approve(str(action["action_id"]), "alice") + + +def _executed_action(gate, world): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + gate.begin_execution(action_id, "runner-1") + return gate.complete_execution(action_id, "runner-1", succeeded=True) + + +# --------------------------------------------------------------------------- # +# Proposal # +# --------------------------------------------------------------------------- # + + +def test_allowlist_ships_empty(): + assert EXECUTION_ALLOWLIST == frozenset() + + +def test_propose_records_action_and_audit(gate, world): + finding_id = world.finding(world.scan()) + + action, outcome = gate.propose(finding_id, "agent") + + assert outcome == "created" + assert action["status"] == PROPOSED + assert action["playbook"] == PLAYBOOK + trail = gate.get_audit_trail(str(action["action_id"])) + assert [(e["event"], e["from_status"], e["to_status"], e["actor"]) for e in trail] == [ + ("proposed", None, PROPOSED, "agent") + ] + + +def test_repeated_propose_returns_the_same_action(gate, world): + finding_id = world.finding(world.scan()) + + first, first_outcome = gate.propose(finding_id, "agent") + second, second_outcome = gate.propose(finding_id, "agent") + + assert (first_outcome, second_outcome) == ("created", "existing") + assert first["action_id"] == second["action_id"] + assert len(gate.get_audit_trail(str(first["action_id"]))) == 1 + + +def test_concurrent_proposals_create_exactly_one_action(dsn, gate, world, playbook_root): + finding_id = world.finding(world.scan()) + + results = _run_concurrently(dsn, 8, lambda g, i: g.propose(finding_id, f"agent-{i}"), playbook_root) + + assert all(kind == "ok" for kind, _ in results), results + assert sorted(outcome for _, (_, outcome) in results) == ["created"] + ["existing"] * 7 + assert len({str(action["action_id"]) for _, (action, _) in results}) == 1 + + +def test_propose_rejects_unknown_finding_missing_playbook_and_blank_actor(gate, world): + with pytest.raises(RemediationNotFound): + gate.propose(2_000_000_000, "agent") + with pytest.raises(RemediationError, match="no playbook"): + gate.propose(world.finding(world.scan(), playbook=""), "agent") + with pytest.raises(ValueError): + gate.propose(world.finding(world.scan()), " ") + + +def test_finding_is_released_only_by_verified_or_rejected(gate, world): + finding_id = world.finding(world.scan()) + action, _ = gate.propose(finding_id, "agent") + gate.reject(str(action["action_id"]), "alice", "not wanted") + + replacement, outcome = gate.propose(finding_id, "agent") + + assert outcome == "created" + assert replacement["action_id"] != action["action_id"] + assert REJECTED not in OPEN_STATUSES and VERIFIED not in OPEN_STATUSES + + +# --------------------------------------------------------------------------- # +# Approval # +# --------------------------------------------------------------------------- # + + +def test_execution_is_refused_without_approval(gate, world): + action, _ = gate.propose(world.finding(world.scan()), "agent") + + with pytest.raises(InvalidTransition) as excinfo: + gate.begin_execution(str(action["action_id"]), "runner-1") + + assert excinfo.value.current == PROPOSED + assert gate.get_action(str(action["action_id"]))["status"] == PROPOSED + + +def test_concurrent_approvals_have_exactly_one_winner(dsn, gate, world, playbook_root): + action, _ = gate.propose(world.finding(world.scan()), "agent") + action_id = str(action["action_id"]) + + results = _run_concurrently(dsn, 8, lambda g, i: g.approve(action_id, f"approver-{i}"), playbook_root) + + winners = [r for kind, r in results if kind == "ok"] + losers = [r for kind, r in results if kind == "err"] + assert len(winners) == 1 + assert len(losers) == 7 and all(isinstance(e, InvalidTransition) for e in losers) + stored = gate.get_action(action_id) + assert stored["status"] == APPROVED + assert stored["approved_by"] == winners[0]["approved_by"] + assert [e["event"] for e in gate.get_audit_trail(action_id)].count("approved") == 1 + + +def test_approve_requires_an_identified_approver(gate, world): + action, _ = gate.propose(world.finding(world.scan()), "agent") + with pytest.raises(ValueError): + gate.approve(str(action["action_id"]), "") + + +def test_reject_records_the_real_predecessor_status(gate, world): + action = _approved_action(gate, world) + + gate.reject(str(action["action_id"]), "alice", "changed my mind") + + last = gate.get_audit_trail(str(action["action_id"]))[-1] + assert (last["event"], last["from_status"], last["to_status"]) == ("rejected", APPROVED, REJECTED) + assert last["detail"] == {"reason": "changed my mind"} + + +# --------------------------------------------------------------------------- # +# Execution # +# --------------------------------------------------------------------------- # + + +def test_default_gate_refuses_every_playbook_and_audits_the_refusal(dsn, world, playbook_root): + db = DatabaseManager(dsn) + try: + strict = RemediationGate(db, playbook_root=playbook_root) + action, _ = strict.propose(world.finding(world.scan()), "agent") + strict.approve(str(action["action_id"]), "alice") + + with pytest.raises(NotAllowlisted): + strict.begin_execution(str(action["action_id"]), "runner-1") + + assert strict.get_action(str(action["action_id"]))["status"] == APPROVED + events = [e["event"] for e in strict.get_audit_trail(str(action["action_id"]))] + assert events[-1] == "execution_refused" + finally: + db.close() + + +def test_a_retried_execution_never_gets_a_second_grant(gate, world): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + + grant = gate.begin_execution(action_id, "runner-1") + with pytest.raises(InvalidTransition) as excinfo: + gate.begin_execution(action_id, "runner-1") + + assert grant["status"] == EXECUTING + assert excinfo.value.current == EXECUTING + assert [e["event"] for e in gate.get_audit_trail(action_id)].count("execution_started") == 1 + + +def test_concurrent_execution_requests_yield_exactly_one_grant(dsn, gate, world, playbook_root): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + + results = _run_concurrently(dsn, 8, lambda g, i: g.begin_execution(action_id, f"runner-{i}"), playbook_root) + + grants = [r for kind, r in results if kind == "ok"] + refusals = [r for kind, r in results if kind == "err"] + assert len(grants) == 1 + assert len(refusals) == 7 and all(isinstance(e, InvalidTransition) for e in refusals) + assert gate.get_action(action_id)["executed_by"] == grants[0]["executed_by"] + assert [e["event"] for e in gate.get_audit_trail(action_id)].count("execution_started") == 1 + + +def test_a_crashed_execution_is_left_executing_and_cannot_be_retried(gate, world): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + gate.begin_execution(action_id, "runner-1") + + # runner-1 dies here: no complete_execution call. + with pytest.raises(InvalidTransition): + gate.begin_execution(action_id, "runner-2") + + assert gate.get_action(action_id)["status"] == EXECUTING + + +def test_only_the_grant_holder_can_report_the_outcome(gate, world): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + gate.begin_execution(action_id, "runner-1") + + with pytest.raises(InvalidTransition): + gate.complete_execution(action_id, "runner-2", succeeded=True) + + assert gate.get_action(action_id)["status"] == EXECUTING + + +def test_successful_execution_awaits_verification_and_failure_stays_open(gate, world): + done = _executed_action(gate, world) + assert done["status"] == PENDING_VERIFICATION + + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + gate.begin_execution(action_id, "runner-1") + failed = gate.complete_execution(action_id, "runner-1", succeeded=False, detail={"exit_code": 2}) + assert failed["status"] == EXECUTION_FAILED + assert EXECUTION_FAILED in OPEN_STATUSES + with pytest.raises(InvalidTransition): + gate.begin_execution(action_id, "runner-1") + + +# --------------------------------------------------------------------------- # +# Rescan verification # +# --------------------------------------------------------------------------- # + + +def _rescan(world, offset=timedelta(seconds=5), **kwargs) -> str: + return world.scan(completed_at=datetime.now(timezone.utc) + offset, **kwargs) + + +def test_pass_on_the_rescan_verifies_and_releases_the_finding(gate, world): + action = _executed_action(gate, world) + action_id = str(action["action_id"]) + rescan = _rescan(world) + world.evaluation(rescan, "PASS") + + updated, outcome = gate.verify(action_id, rescan) + + assert outcome == "verified" + assert updated["status"] == VERIFIED and updated["verification_scan_id"] is not None + assert updated["verified_at"] is not None + last = gate.get_audit_trail(action_id)[-1] + assert (last["event"], last["detail"]["observed"]) == ("verification_passed", "PASS") + + +def test_a_still_failing_rescan_stays_open_as_verification_failed(gate, world): + action = _executed_action(gate, world) + action_id = str(action["action_id"]) + rescan = _rescan(world) + world.evaluation(rescan, "FAIL") + + updated, outcome = gate.verify(action_id, rescan) + + assert outcome == "failed" + assert updated["status"] == VERIFICATION_FAILED + assert updated["verified_at"] is None + with pytest.raises(InvalidTransition): + gate.begin_execution(action_id, "runner-1") + replacement, proposal_outcome = gate.propose(updated["finding_id"], "agent") + assert proposal_outcome == "existing" and replacement["action_id"] == updated["action_id"] + + +@pytest.mark.parametrize("observed", [None, "UNKNOWN", "ERROR", "NOT_APPLICABLE"]) +def test_missing_or_non_conclusive_evidence_never_closes_the_action(gate, world, observed): + action = _executed_action(gate, world) + action_id = str(action["action_id"]) + rescan = _rescan(world) + if observed: + world.evaluation(rescan, observed) + + updated, outcome = gate.verify(action_id, rescan) + + assert outcome == "inconclusive" + assert updated["status"] == PENDING_VERIFICATION + assert gate.get_audit_trail(action_id)[-1]["event"] == "verification_inconclusive" + + +def test_a_failed_verification_can_later_be_verified_by_a_fresh_rescan(gate, world): + action = _executed_action(gate, world) + action_id = str(action["action_id"]) + first = _rescan(world) + world.evaluation(first, "FAIL") + gate.verify(action_id, first) + second = _rescan(world, timedelta(seconds=10)) + world.evaluation(second, "PASS") + + _, outcome = gate.verify(action_id, second) + + assert outcome == "verified" + assert gate.get_action(action_id)["status"] == VERIFIED + + +def test_rescan_must_be_completed_later_and_for_the_same_subscription(gate, world): + action = _executed_action(gate, world) + action_id = str(action["action_id"]) + candidates = { + "older than the execution": world.scan(completed_at=datetime.now(timezone.utc) - timedelta(days=1)), + "not completed": _rescan(world, status="running"), + "another subscription": _rescan(world, subscription=str(uuid.uuid4())), + } + for scan_id in candidates.values(): + world.evaluation(scan_id, "PASS") + + for label, scan_id in candidates.items(): + with pytest.raises(RemediationError, match="completed rescan"): + gate.verify(action_id, scan_id) + assert gate.get_action(action_id)["status"] == PENDING_VERIFICATION, label + + +def test_verification_requires_a_pending_action(gate, world): + action = _approved_action(gate, world) + rescan = _rescan(world) + world.evaluation(rescan, "PASS") + + with pytest.raises(InvalidTransition) as excinfo: + gate.verify(str(action["action_id"]), rescan) + + assert excinfo.value.current == APPROVED + + +def test_concurrent_verification_of_one_action_settles_once(dsn, gate, world, playbook_root): + action = _executed_action(gate, world) + action_id = str(action["action_id"]) + rescan = _rescan(world) + world.evaluation(rescan, "PASS") + + results = _run_concurrently(dsn, 6, lambda g, i: g.verify(action_id, rescan), playbook_root) + + outcomes = [r[1] for kind, r in results if kind == "ok"] + assert outcomes.count("verified") == 1 + assert all(isinstance(r, InvalidTransition) for kind, r in results if kind == "err") + assert [e["event"] for e in gate.get_audit_trail(action_id)].count("verification_passed") == 1 + + +# --------------------------------------------------------------------------- # +# Audit trail # +# --------------------------------------------------------------------------- # + + +def test_full_lifecycle_is_audited_in_order(gate, world): + action = _executed_action(gate, world) + action_id = str(action["action_id"]) + rescan = _rescan(world) + world.evaluation(rescan, "PASS") + gate.verify(action_id, rescan) + + trail = gate.get_audit_trail(action_id) + + assert [e["event"] for e in trail] == [ + "proposed", + "approved", + "execution_started", + "execution_succeeded", + "verification_passed", + ] + assert [e["actor"] for e in trail] == ["agent", "alice", "runner-1", "runner-1", "system"] + assert [e["to_status"] for e in trail] == [PROPOSED, APPROVED, EXECUTING, PENDING_VERIFICATION, VERIFIED] + + +def test_audit_log_is_append_only(dsn, gate, world): + action, _ = gate.propose(world.finding(world.scan()), "agent") + + with psycopg2.connect(dsn) as conn, conn.cursor() as cur: + with pytest.raises(psycopg2.Error, match="append-only"): + cur.execute( + "UPDATE remediation_audit_log SET actor = 'mallory' WHERE action_id = %s", (str(action["action_id"]),) + ) + with psycopg2.connect(dsn) as conn, conn.cursor() as cur: + with pytest.raises(psycopg2.Error, match="append-only"): + cur.execute("DELETE FROM remediation_audit_log WHERE action_id = %s", (str(action["action_id"]),)) + + assert gate.get_audit_trail(str(action["action_id"]))[0]["actor"] == "agent" + + +def test_unknown_action_is_reported_as_not_found(gate): + missing = str(uuid.uuid4()) + with pytest.raises(RemediationNotFound): + gate.approve(missing, "alice") + assert gate.get_action(missing) is None + + +# --------------------------------------------------------------------------- # +# Audit trail cannot be wiped # +# --------------------------------------------------------------------------- # + + +def test_audit_log_cannot_be_truncated(dsn, gate, world): + action, _ = gate.propose(world.finding(world.scan()), "agent") + action_id = str(action["action_id"]) + + with psycopg2.connect(dsn) as conn, conn.cursor() as cur: + with pytest.raises(psycopg2.Error, match="append-only"): + cur.execute("TRUNCATE remediation_audit_log") + + assert [e["event"] for e in gate.get_audit_trail(action_id)] == ["proposed"] + + +# --------------------------------------------------------------------------- # +# Approval covers exactly what will run # +# --------------------------------------------------------------------------- # + + +def test_proposal_pins_the_script_hash_and_the_target(gate, world): + resource = "/subscriptions/x/resourceGroups/rg/providers/Microsoft.Storage/storageAccounts/acct" + finding_id = world.finding(world.scan(), resource_id=resource) + + action, _ = gate.propose(finding_id, "agent") + + assert action["playbook_sha256"] == hashlib.sha256(SCRIPT_BODY).hexdigest() + assert action["resource_id"] == resource + assert action["subscription_id"] == world.subscription_id + + +def test_approval_audit_row_records_what_the_approver_approved(gate, world): + action = _approved_action(gate, world) + + approved = [e for e in gate.get_audit_trail(str(action["action_id"])) if e["event"] == "approved"] + + assert len(approved) == 1 + detail = approved[0]["detail"] + assert detail["playbook"] == PLAYBOOK + assert detail["playbook_sha256"] == hashlib.sha256(SCRIPT_BODY).hexdigest() + assert detail["resource_id"] == action["resource_id"] + assert detail["subscription_id"] == world.subscription_id + + +def test_a_script_edited_after_approval_is_never_granted(gate, world, playbook_root): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + (playbook_root / PLAYBOOK).write_bytes(SCRIPT_BODY + b"az group delete --name production --yes\n") + + with pytest.raises(PlaybookChanged): + gate.begin_execution(action_id, "runner-1") + + assert gate.get_action(action_id)["status"] == APPROVED + refusal = gate.get_audit_trail(action_id)[-1] + assert refusal["event"] == "execution_refused" + assert refusal["detail"]["reason"] == "playbook_changed" + assert refusal["detail"]["approved_sha256"] == hashlib.sha256(SCRIPT_BODY).hexdigest() + assert refusal["detail"]["current_sha256"] != refusal["detail"]["approved_sha256"] + + +def test_restoring_the_approved_script_makes_the_action_grantable_again(gate, world, playbook_root): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + script = playbook_root / PLAYBOOK + script.write_bytes(SCRIPT_BODY + b"# tampered\n") + with pytest.raises(PlaybookChanged): + gate.begin_execution(action_id, "runner-1") + + script.write_bytes(SCRIPT_BODY) + + assert gate.begin_execution(action_id, "runner-1")["status"] == EXECUTING + + +def test_a_script_removed_after_approval_is_never_granted(gate, world, playbook_root): + action = _approved_action(gate, world) + action_id = str(action["action_id"]) + (playbook_root / PLAYBOOK).unlink() + + with pytest.raises(PlaybookChanged): + gate.begin_execution(action_id, "runner-1") + + assert gate.get_action(action_id)["status"] == APPROVED + assert gate.get_audit_trail(action_id)[-1]["detail"]["current_sha256"] is None + + +@pytest.mark.parametrize( + "playbook", + [ + "playbooks/cli/does_not_exist.sh", + "../outside.sh", + "playbooks/../../outside.sh", + "/etc/passwd", + "playbooks/cli", + ], +) +def test_propose_refuses_a_playbook_that_is_missing_or_outside_playbooks(gate, world, playbook_root, playbook): + (playbook_root.parent / "outside.sh").write_text("echo outside") + finding_id = world.finding(world.scan(), playbook=playbook) + + with pytest.raises(PlaybookUnavailable): + gate.propose(finding_id, "agent") + + with psycopg2.connect(os.environ["DATABASE_URL"]) as conn, conn.cursor() as cur: + cur.execute("SELECT count(*) FROM remediation_actions WHERE finding_id = %s", (finding_id,)) + assert cur.fetchone()[0] == 0