From 347484fa50226ce4e9c599cd0f74d7629efc43fc Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 29 Jun 2026 01:54:56 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20P1=20hardening=20suite=20=E2=80=94?= =?UTF-8?q?=20approval=20gate,=20audit=20anchoring,=20egress,=20pre-flight?= =?UTF-8?q?,=20key-strength?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the remaining roadmap issues (#4 partial, #5, #6, #7, #8). #6 approval gate (approval.py): --sandbox local now requires a parameter-bound, fail-closed approval token (HMAC over mode + canonical-sorted-targets + expiry, keyed by the audit key). Mint with `aegis approve local [--ttl N]`, pass via --approval / ARGUS_APPROVAL_TOKEN. Cannot be replayed against other targets or after expiry; wildcard grants are impossible. The banner no longer "authorizes" — no token, no run. #5 audit anchoring (anchor.py): AuditLog mirrors the chain tip {seq, tip, ts} to an out-of-band anchor after every entry; `aegis audit` cross-checks it. Detects a full-log rewrite that re-verifies under a leaked key (HMAC-VALID but anchor MISMATCH). Enabled via audit.anchor_path (point at S3 Object Lock / WORM in prod). #7 egress control (egress.py): `aegis egress-rules` emits a default-drop nftables allow-list from the scope policy (denied ranges carved out first) for the disposable recon host — the missing network-layer second barrier for local mode. #8 pre-flight (preflight.py): warns on public-IP targets / over-broad scope / resolve_dns before any packet on the un-isolated path. #4 (short-term): AuditLog fails closed on a weak audit key (<32 chars) so a placeholder key can't sign a "tamper-evident" chain. Full out-of-band signer remains tracked. Config: Policy.audit_anchor_path; commented anchor_path example in scope-policy.yaml. Tests: +14 (test_p1_hardening.py); test fixtures bumped to 32-char keys. 95 passed, ruff + Bandit(medium+) clean. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PH7w1wBoUxFESq9gh6yjAQ --- aegis/aegis/anchor.py | 39 +++++++ aegis/aegis/approval.py | 69 +++++++++++++ aegis/aegis/cli.py | 60 ++++++++++- aegis/aegis/config.py | 8 ++ aegis/aegis/egress.py | 49 +++++++++ aegis/aegis/guardrail.py | 46 ++++++++- aegis/aegis/preflight.py | 60 +++++++++++ aegis/docs/NEXT_STEPS.md | 57 +++++++---- aegis/tests/test_agent.py | 2 +- aegis/tests/test_guardrail.py | 2 +- aegis/tests/test_integration_agentic.py | 2 +- aegis/tests/test_p1_hardening.py | 131 ++++++++++++++++++++++++ aegis/tests/test_recon_modules.py | 2 +- aegis/tests/test_webrecon.py | 2 +- targets/scope-policy.yaml | 5 + 15 files changed, 501 insertions(+), 33 deletions(-) create mode 100644 aegis/aegis/anchor.py create mode 100644 aegis/aegis/approval.py create mode 100644 aegis/aegis/egress.py create mode 100644 aegis/aegis/preflight.py create mode 100644 aegis/tests/test_p1_hardening.py diff --git a/aegis/aegis/anchor.py b/aegis/aegis/anchor.py new file mode 100644 index 0000000..82661e8 --- /dev/null +++ b/aegis/aegis/anchor.py @@ -0,0 +1,39 @@ +"""External anchoring of the audit-chain tip (#5). + +In-file HMAC chaining proves tamper-evidence *only while the signing key is secret*. +If the key leaks, an attacker can rewrite the whole log and recompute every HMAC so it +re-verifies. Anchoring defends against that: the latest chain tip ``{seq, tip, ts}`` is +written to a SEPARATE file that production points at a write-once store the tool runner +cannot rewrite (S3 Object Lock compliance mode, a KMS-signed object, or a WORM volume +owned by a different uid). ``aegis audit`` then cross-checks the live chain against the +anchor, so a full-log rewrite is still detectable unless the attacker also forges the +out-of-band anchor — which the runner has no path to. + +Best practice: "HMAC alone does not defend against a compromised app — layer append-only +storage and external anchoring on top" (Tracehold; CloudTrail log-file validation). + +This module owns only the small, infra-agnostic anchor file format. The production WORM +target is an operational choice (point ``audit.anchor_path`` at the mounted WORM path). +""" +from __future__ import annotations + +import json +from pathlib import Path + + +def write_anchor(path: Path, seq: int, tip: str, ts: float) -> None: + """Overwrite the anchor with the current chain tip. One small JSON object, no chaining + (the anchor's protection comes from living on a write-once medium, not from a MAC).""" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps({"seq": seq, "tip": tip, "ts": round(ts, 3)}, + separators=(",", ":")) + "\n") + + +def read_anchor(path: Path) -> dict | None: + """Return the anchor record, or None if absent/unreadable.""" + if not path.exists(): + return None + try: + return json.loads(path.read_text()) + except (json.JSONDecodeError, OSError): + return None diff --git a/aegis/aegis/approval.py b/aegis/aegis/approval.py new file mode 100644 index 0000000..5c5b6d9 --- /dev/null +++ b/aegis/aegis/approval.py @@ -0,0 +1,69 @@ +"""Parameter-bound, fail-closed approval tokens for high-risk run modes (#6). + +A token authorizes ONE mode (e.g. ``local``) against an EXACT set of targets until a +hard expiry. It is an HMAC over ``(mode, canonical-sorted-targets, expiry)`` keyed by +the audit signing key, so: + + * only an operator who holds the key can mint one (the LLM/agent never can), + * it cannot be replayed against different targets, and + * it stops working after it expires. + +Best practice: bind approval to the exact action and fail closed — telling the model +"stay in scope" is not enforcement (OWASP AI Agent Security Cheat Sheet; ROE Gate). +Wildcard / open-ended grants are intentionally impossible: every token names its +targets and carries a TTL. +""" +from __future__ import annotations + +import base64 +import binascii +import hashlib +import hmac +import time + +from .guardrail import canon_network + + +class ApprovalError(Exception): + """Raised when an approval token is missing, malformed, expired, or mismatched.""" + + +def _canon_targets(targets: list[str]) -> str: + """Normalize targets to canonical networks, deduped + sorted, so the token binds to + the same set regardless of ordering or obfuscation (decimal/hex/leading-zero).""" + nets = sorted({str(canon_network(t)) for t in targets}) + return ",".join(nets) + + +def _body(mode: str, targets: list[str], expiry: int) -> str: + return f"v1|{mode}|{_canon_targets(targets)}|{expiry}" + + +def mint(mode: str, targets: list[str], key: str, *, ttl: int = 3600, + now: float | None = None) -> str: + """Mint a base64 approval token authorizing `mode` against `targets` for `ttl` seconds.""" + if ttl <= 0: + raise ApprovalError("ttl must be positive") + expiry = int((time.time() if now is None else now) + ttl) + mac = hmac.new(key.encode(), _body(mode, targets, expiry).encode(), hashlib.sha256).hexdigest() + return base64.urlsafe_b64encode(f"{expiry}.{mac}".encode()).decode() + + +def verify(token: str, mode: str, targets: list[str], key: str, + *, now: float | None = None) -> None: + """Raise ApprovalError unless `token` authorizes exactly `mode`+`targets` and is unexpired.""" + if not token: + raise ApprovalError(f"{mode} mode requires an approval token (mint one with `aegis approve`)") + try: + raw = base64.urlsafe_b64decode(token.encode()).decode() + expiry_s, mac = raw.split(".", 1) + expiry = int(expiry_s) + except (binascii.Error, ValueError, UnicodeDecodeError) as exc: + raise ApprovalError("malformed approval token") from exc + now = time.time() if now is None else now + if now > expiry: + raise ApprovalError("approval token expired — mint a fresh one") + expect = hmac.new(key.encode(), _body(mode, targets, expiry).encode(), hashlib.sha256).hexdigest() + # compare_digest over the full token guards target/mode tampering AND a forged MAC. + if not hmac.compare_digest(mac, expect): + raise ApprovalError("approval token does not authorize this mode/target set") diff --git a/aegis/aegis/cli.py b/aegis/aegis/cli.py index 006e491..f2f89f0 100644 --- a/aegis/aegis/cli.py +++ b/aegis/aegis/cli.py @@ -9,9 +9,11 @@ from __future__ import annotations import argparse +import os import sys from pathlib import Path +from . import approval, egress, preflight from .config import DEFAULT_POLICY, Policy from .guardrail import Guardrail, GuardrailError from .orchestrator import Orchestrator, default_plan @@ -41,6 +43,16 @@ def cmd_scan(args) -> int: print("WARNING: --sandbox local runs tools directly on THIS host with NO " "network isolation. Use only for AUTHORIZED off-lab recon under a " "tight, written-authorized scope policy.", file=sys.stderr) + # Pre-flight: surface likely misconfiguration before any packet leaves (#8). + for w in preflight.check(args.targets, guard.policy): + print(f" preflight: {w}", file=sys.stderr) + # Fail-closed, parameter-bound approval gate — banners don't enforce (#6). + try: + approval.verify(args.approval, "local", args.targets, + os.environ.get(guard.policy.audit_key_env, "")) + except approval.ApprovalError as exc: + print(f"REFUSED --sandbox local: {exc}", file=sys.stderr) + return 2 try: sandbox = LocalSandbox(audit_key_env=guard.policy.audit_key_env) except RuntimeError as exc: @@ -180,9 +192,40 @@ def cmd_audit(args) -> int: guard = _guard(args) ok = guard.audit.verify() print("audit chain:", "VALID ✅" if ok else "TAMPERED ❌") + if guard.policy.audit_anchor_path is not None: + anchored, reason = guard.audit.cross_check_anchor() + print("anchor:", "OK ✅" if anchored else "MISMATCH ❌", "—", reason) + ok = ok and anchored return 0 if ok else 1 +def cmd_approve(args) -> int: + """Mint a parameter-bound approval token for a high-risk run mode (#6).""" + policy = Policy.load(args.policy) + key = os.environ.get(policy.audit_key_env) + if not key: + print(f"REFUSED: audit key env {policy.audit_key_env} unset", file=sys.stderr) + return 2 + # Scope-validate every target so an out-of-scope set can never be approved. + guard = _guard(args) + for t in args.targets: + d = guard.check_target(t) + if not d.allowed: + print(f"REFUSED target {t}: {d.reason}", file=sys.stderr) + return 2 + token = approval.mint(args.mode, args.targets, key, ttl=args.ttl) + print(token) + print(f"# authorizes `{args.mode}` against {args.targets} for {args.ttl}s", + file=sys.stderr) + return 0 + + +def cmd_egress(args) -> int: + """Print an nftables egress allow-list derived from the scope policy (#7).""" + sys.stdout.write(egress.nftables_ruleset(Policy.load(args.policy))) + return 0 + + def cmd_verify(args) -> int: """Static policy sanity check (no network).""" policy = Policy.load(args.policy) @@ -212,7 +255,9 @@ def build_parser() -> argparse.ArgumentParser: s.add_argument("--sandbox", choices=["docker", "local"], default="docker", help="docker = isolated lab container (default); " "local = run read-only tools on the host for AUTHORIZED off-lab " - "recon (requires a tight scope policy)") + "recon (requires a tight scope policy + approval token)") + s.add_argument("--approval", default=os.environ.get("ARGUS_APPROVAL_TOKEN"), + help="approval token (REQUIRED for --sandbox local); mint with `aegis approve`") s.set_defaults(func=cmd_scan) h = sub.add_parser("host", help="credentialed read-only host audit (Linux/SSH or Windows/WinRM)") @@ -273,9 +318,20 @@ def build_parser() -> argparse.ArgumentParser: ag.add_argument("--dry-run", action="store_true") ag.set_defaults(func=cmd_agent) - a = sub.add_parser("audit", help="verify the HMAC audit chain") + a = sub.add_parser("audit", help="verify the HMAC audit chain (+ anchor cross-check)") a.set_defaults(func=cmd_audit) + ap = sub.add_parser("approve", + help="mint a parameter-bound approval token for --sandbox local (#6)") + ap.add_argument("mode", choices=["local"]) + ap.add_argument("targets", nargs="+") + ap.add_argument("--ttl", type=int, default=3600, help="token lifetime in seconds (default 3600)") + ap.set_defaults(func=cmd_approve) + + eg = sub.add_parser("egress-rules", + help="print an nftables egress allow-list from the scope policy (#7)") + eg.set_defaults(func=cmd_egress) + v = sub.add_parser("verify", help="print loaded policy") v.set_defaults(func=cmd_verify) return p diff --git a/aegis/aegis/config.py b/aegis/aegis/config.py index 4f1f2f5..7598edb 100644 --- a/aegis/aegis/config.py +++ b/aegis/aegis/config.py @@ -35,6 +35,8 @@ class Policy: audit_path: Path audit_chained: bool redact_patterns: tuple[str, ...] = field(default=()) + # Optional out-of-band anchor for the chain tip (#5). Point at a WORM/append-only path. + audit_anchor_path: Path | None = None @staticmethod def load(path: Path | str = DEFAULT_POLICY) -> Policy: @@ -47,6 +49,11 @@ def load(path: Path | str = DEFAULT_POLICY) -> Policy: audit_path = Path(aud.get("path", "aegis/output/audit.ndjson")) if not audit_path.is_absolute(): audit_path = REPO_ROOT / audit_path + anchor_path = aud.get("anchor_path") + if anchor_path: + anchor_path = Path(anchor_path) + if not anchor_path.is_absolute(): + anchor_path = REPO_ROOT / anchor_path def nets(key: str) -> tuple[ipaddress.IPv4Network, ...]: return tuple( @@ -71,4 +78,5 @@ def nets(key: str) -> tuple[ipaddress.IPv4Network, ...]: audit_path=audit_path, audit_chained=bool(aud.get("chained", True)), redact_patterns=tuple(san.get("redact_patterns", [])), + audit_anchor_path=anchor_path or None, ) diff --git a/aegis/aegis/egress.py b/aegis/aegis/egress.py new file mode 100644 index 0000000..a2f644b --- /dev/null +++ b/aegis/aegis/egress.py @@ -0,0 +1,49 @@ +"""Generate an nftables egress allow-list from the scope policy (#7). + +``--sandbox local`` removes the container's network boundary, leaving the app-layer scope +guard as the only thing between the tool runner and the wider network. A kernel firewall +restores defense-in-depth: permit egress ONLY to the policy's allowed CIDRs (minus the +denied ones), drop everything else. Apply this on the disposable recon host BEFORE an +off-lab run — a guardrail bug or an HTTP redirect then still cannot reach an out-of-scope +host, because the packet never leaves the box. + +Best practice: enforce scope at the network layer, not just the app (Aikido); exclusions +beat authorizations (IntegSec agentic-pentest proxy). This emits the ruleset; the operator +applies it with ``nft -f`` (it does not touch the live firewall itself). +""" +from __future__ import annotations + +from .config import Policy + +_TABLE = "argus_egress" + + +def nftables_ruleset(policy: Policy) -> str: + """Render a deterministic nftables script: allow established + loopback + the policy's + in-scope networks (denied ones carved out first), default-drop new egress.""" + allowed = [n for n in policy.allowed_networks] + denied = [str(d) for d in policy.denied_networks] + lines = [ + "#!/usr/sbin/nft -f", + "# Argus egress allow-list — generated from the scope policy. Apply on the", + "# disposable recon host before `--sandbox local`. Fail-closed: default drop.", + f"add table inet {_TABLE}", + f"delete table inet {_TABLE}", + f"table inet {_TABLE} {{", + " chain output {", + " type filter hook output priority 0; policy drop;", + " ct state established,related accept", + " oifname \"lo\" accept", + ] + # Denied carve-outs first so a /32 deny inside an allowed /24 wins (firewall semantics). + for d in denied: + lines.append(f" ip daddr {d} drop") + if allowed: + joined = ", ".join(str(a) for a in allowed) + lines.append(f" ip daddr {{ {joined} }} accept") + lines += [ + " # everything else: dropped by chain policy", + " }", + "}", + ] + return "\n".join(lines) + "\n" diff --git a/aegis/aegis/guardrail.py b/aegis/aegis/guardrail.py index 0ff2032..068e056 100644 --- a/aegis/aegis/guardrail.py +++ b/aegis/aegis/guardrail.py @@ -94,6 +94,11 @@ def canon_network(token: str) -> ipaddress.IPv4Network: return ipaddress.ip_network(f"{'.'.join(octets)}/{int(prefix)}", strict=True) +# Minimum audit-key length (#4). The documented key is `openssl rand -hex 32` (64 chars); +# we fail closed below 32 so a weak/placeholder key can't sign a "tamper-evident" chain. +MIN_AUDIT_KEY_LEN = 32 + + class AuditLog: """HMAC-SHA256 chained, append-only, tamper-evident audit trail.""" @@ -103,23 +108,31 @@ def __init__(self, policy: Policy): raise GuardrailError( f"audit key env {policy.audit_key_env} unset — refusing to run unaudited" ) + if len(key) < MIN_AUDIT_KEY_LEN: + raise GuardrailError( + f"audit key too short ({len(key)}<{MIN_AUDIT_KEY_LEN} chars) — a weak key " + f"makes the chain forgeable; generate one with `openssl rand -hex 32`" + ) self._key = key.encode() self._path = policy.audit_path self._chained = policy.audit_chained + self._anchor_path = policy.audit_anchor_path self._path.parent.mkdir(parents=True, exist_ok=True) - self._prev = self._last_hmac() + self._prev, self._seq = self._last_state() - def _last_hmac(self) -> str: + def _last_state(self) -> tuple[str, int]: + """Return (last_hmac, entry_count) by replaying the existing log.""" if not self._path.exists(): - return "genesis" - last = "genesis" + return "genesis", 0 + last, seq = "genesis", 0 for line in self._path.read_text().splitlines(): if line.strip(): try: last = json.loads(line)["hmac"] + seq += 1 except (json.JSONDecodeError, KeyError): continue - return last + return last, seq def write(self, event: dict) -> str: entry = {"ts": round(time.time(), 3), @@ -131,8 +144,31 @@ def write(self, event: dict) -> str: with self._path.open("a") as fh: fh.write(json.dumps(entry, separators=(",", ":")) + "\n") self._prev = mac + self._seq += 1 + # Update the out-of-band anchor so the chain tip is mirrored to a WORM store (#5). + if self._anchor_path is not None: + from . import anchor + anchor.write_anchor(self._anchor_path, self._seq, mac, entry["ts"]) return mac + def cross_check_anchor(self) -> tuple[bool, str]: + """Compare the live chain tip against the out-of-band anchor (#5). + + Detects a full-log rewrite that the in-file HMAC alone cannot (an attacker with the + leaked key recomputes every MAC, but cannot also forge the WORM anchor). + """ + if self._anchor_path is None: + return True, "no anchor configured" + from . import anchor + rec = anchor.read_anchor(self._anchor_path) + if rec is None: + return False, "anchor missing — chain tip was never anchored or anchor was deleted" + tip, seq = self._last_state() + if rec.get("tip") != tip or rec.get("seq") != seq: + return (False, f"anchor mismatch — anchor=(seq={rec.get('seq')}, " + f"tip={str(rec.get('tip'))[:8]}…) chain=(seq={seq}, tip={tip[:8]}…)") + return True, f"anchor matches chain tip (seq={seq})" + def verify(self) -> bool: """Replay the chain; False if any entry was altered, reordered, or removed.""" prev = "genesis" diff --git a/aegis/aegis/preflight.py b/aegis/aegis/preflight.py new file mode 100644 index 0000000..658866b --- /dev/null +++ b/aegis/aegis/preflight.py @@ -0,0 +1,60 @@ +"""Pre-flight sanity checks before an un-isolated ``--sandbox local`` run (#8). + +Configuration mistakes are likelier than malice: the most common way to cause harm is to +point recon at the wrong network (production, a public host, an over-broad scope). These +checks run BEFORE any packet and surface warnings so a human catches the mistake at setup +time rather than relying on runtime controls to undo it. + +Best practice: "catch human error before execution starts, rather than relying on runtime +controls to fix avoidable setup mistakes" (Aikido pre-flight checks). +""" +from __future__ import annotations + +import ipaddress + +from .config import Policy +from .guardrail import canon_network + + +def check(targets: list[str], policy: Policy) -> list[str]: + """Return human-readable warnings for an un-isolated run. Empty list = looks fine. + + Never raises on a bad target — the scope guard already rejects those; here we only + advise. The caller decides whether to surface or hard-stop on the warnings. + """ + warnings: list[str] = [] + + for t in targets: + try: + net = canon_network(t) + except ValueError: + continue # scope guard will reject; not our job to re-validate + addr = net.network_address + if addr.is_global: + warnings.append( + f"target {t} is a PUBLIC IP — on the un-isolated local path this sends " + f"live packets across the internet; confirm written authorization") + if not net.is_private and not addr.is_global and not addr.is_loopback: + warnings.append(f"target {t} is a special-use address ({net}) — double-check intent") + + # An over-broad allow-list is dangerous specifically when there is no container boundary. + total = sum(n.num_addresses for n in policy.allowed_networks) + if total > 256: + warnings.append( + f"scope allow-list spans {total} addresses (> a /24) — tighten to the exact " + f"hosts for off-lab recon (ideally /32 entries)") + + if policy.resolve_dns: + warnings.append("resolve_dns is enabled — a name could resolve off-scope; prefer literal IPs") + + return warnings + + +def is_lab_only(policy: Policy) -> bool: + """True if the loaded policy still only allows RFC-1918 space (i.e. likely the lab + default, not a tightened off-lab scope) — used to flag a probable misconfiguration.""" + return bool(policy.allowed_networks) and all( + n.subnet_of(ipaddress.ip_network("10.0.0.0/8")) + or n.subnet_of(ipaddress.ip_network("172.16.0.0/12")) + or n.subnet_of(ipaddress.ip_network("192.168.0.0/16")) + for n in policy.allowed_networks) diff --git a/aegis/docs/NEXT_STEPS.md b/aegis/docs/NEXT_STEPS.md index 0e5758e..155007d 100644 --- a/aegis/docs/NEXT_STEPS.md +++ b/aegis/docs/NEXT_STEPS.md @@ -52,35 +52,42 @@ own guidance in code — refuse to start if the audit key path is owned by the s runner or has loose perms (fail-to-start, don't bypass). Longer term: a tiny out-of-band signer the orchestrator talks to but never holds the key for. +**✅ Short-term landed:** `AuditLog` now **fails closed on a weak audit key** (`MIN_AUDIT_KEY_LEN += 32`) — a placeholder/short key can no longer sign a "tamper-evident" chain (`guardrail.py`). +**🔭 Remaining:** the full out-of-band signer process (orchestrator never holds the key). + > *Best practice:* "the agent never touches the signing keys" (ROE Gate, reference-monitor / > Anderson 1972); "the key sits outside the log volume so an attacker who can write the log > can't read or rotate the key" (`bernstein` audit-log operations doc). -### 🔭 5. External anchoring + WORM for the audit chain -HMAC chaining alone does not survive a key compromise. On a timer, write the latest chain -tip to a destination the runner can't rewrite (S3 Object Lock compliance mode, or a -KMS-signed artifact) and have the verifier cross-check it. +### ✅ 5. External anchoring + WORM for the audit chain +`AuditLog` mirrors the chain tip `{seq, tip, ts}` to an out-of-band **anchor file** after every +entry (`anchor.py`); `aegis audit` **cross-checks** the live chain against it, so a full-log +rewrite (even with a leaked key) is detectable. Enabled by setting `audit.anchor_path` in the +policy. **🔭 Operational:** point that path at a real write-once store (S3 Object Lock compliance +mode / KMS-signed object / WORM volume) owned by a different uid — the file format is ready. > *Best practice:* "HMAC alone does not defend against a compromised app — layer append-only > storage and external anchoring on top." — Tracehold; SystemsHardening audit-logging > architecture (CloudTrail-style log-file validation). -### 🔭 6. Risk-tiered approval gate for `--sandbox local` / `--arm` -Replace the current warning banner with a real, parameter-bound acknowledgment: bind -approval to the exact action (actor + tool + normalized target + expiry), fail closed on -missing approval, and forbid wildcard grants for the un-isolated local path. +### ✅ 6. Risk-tiered approval gate for `--sandbox local` +`--sandbox local` now requires a **parameter-bound, fail-closed approval token** (`approval.py`): +an HMAC over `(mode, canonical-sorted-targets, expiry)` keyed by the audit key. Mint with +`aegis approve local [--ttl N]`; pass via `--approval`/`ARGUS_APPROVAL_TOKEN`. A token +cannot be replayed against different targets or after expiry, and wildcard grants are impossible +(every token names its targets). The banner alone is gone — no token, no run. > *Best practice:* "system-prompt guardrails don't guard anything… agents take risky actions > 23.9% of the time even with explicit safety instructions" (ROE Gate); "bind approval to the > exact action… fail closed when approval validation fails" (OWASP AI Agent Security Cheat > Sheet); avoid wildcard trust grants (AWS Well-Architected, agentic-AI lens). -### 🔭 7. Network-layer egress control for the local path -`--sandbox local` now relies *solely* on the app-layer scope guard (the container boundary is -gone). Best-of-breed pentest agents enforce scope at the **network/DNS layer** too, so a -guardrail bug or HTTP redirect can't reach an out-of-scope host. Even a documented -`nftables` egress-allowlist helper (matching the project's own two-node disposable-host -guidance) adds the missing second layer. +### ✅ 7. Network-layer egress control for the local path +`aegis egress-rules` generates a deterministic **nftables egress allow-list** from the scope +policy (`egress.py`): default-drop, allow only the policy's in-scope CIDRs (denied ranges carved +out first). Apply it with `nft -f` on the disposable recon host before `--sandbox local`, so a +guardrail bug or HTTP redirect can't reach an out-of-scope host — the packet never leaves the box. > *Best practice:* "we don't rely on prompts or humans for scope enforcement — we enforce at > the network layer, intercepting HTTP and DNS" (Aikido); "exclusions beat authorizations; @@ -101,15 +108,23 @@ new tests in `tests/test_sandbox.py`. Added `aegis/pyproject.toml` (Bandit / Ruff / pytest config) and `.pre-commit-config.yaml` mirroring the CI gates for shift-left feedback. -### 🔭 10. Pre-flight "is this production?" checks -Before a `--sandbox local` run, validate reachability and surface "this resembles production" -warnings early — "configuration mistakes are more likely than malicious behaviour." — Aikido -pre-flight checks. +### ✅ 10. Pre-flight "is this production?" checks +Before a `--sandbox local` run, `preflight.check()` (`preflight.py`) surfaces warnings to stderr +— public-IP targets on the un-isolated path, an over-broad allow-list (> a /24), `resolve_dns` +enabled — so a human catches the misconfiguration at setup time rather than after packets fly. + +> *Best practice:* "catch human error before execution starts, rather than relying on runtime +> controls to fix avoidable setup mistakes." — Aikido pre-flight checks. --- -## Suggested sequencing -`P0 (CI + pinning)` → `P1.6 approval gate + P2.8 127-fix (done)` → `P1.4/5 key isolation + -anchoring` → `P1.7 network egress` → `P2 polish`. +## Status +**Done:** P0 (CI + supply-chain), P1.5/6/7, P2.8/9/10, and the P1.4 short-term key-strength gate. +**Remaining:** P1.4 full out-of-band signer process, and the P1.5 production WORM target +(the in-repo anchor format + cross-check are ready; pointing it at S3 Object Lock / KMS is an +operational step). + +## Suggested sequencing (remaining) +`P1.4 out-of-band signer` → wire `audit.anchor_path` to a real WORM store in production. _Tracking issues are linked from each 🔭 item once opened._ diff --git a/aegis/tests/test_agent.py b/aegis/tests/test_agent.py index ef0a076..44fb999 100644 --- a/aegis/tests/test_agent.py +++ b/aegis/tests/test_agent.py @@ -3,7 +3,7 @@ import pytest -os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "test") +os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "k" * 32) from aegis.agent import chains, planner from aegis.agent.poc_runner import PoCRefused, gate_check, verify diff --git a/aegis/tests/test_guardrail.py b/aegis/tests/test_guardrail.py index 49cb289..238824d 100644 --- a/aegis/tests/test_guardrail.py +++ b/aegis/tests/test_guardrail.py @@ -6,7 +6,7 @@ import pytest -os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "test-key-0123456789") +os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "k" * 32) from aegis.config import Policy # noqa: E402 from aegis.guardrail import Guardrail, GuardrailError, canon_network # noqa: E402 diff --git a/aegis/tests/test_integration_agentic.py b/aegis/tests/test_integration_agentic.py index 9c89969..b6597a7 100644 --- a/aegis/tests/test_integration_agentic.py +++ b/aegis/tests/test_integration_agentic.py @@ -1,7 +1,7 @@ """End-to-end: a scan auto-enriches with shadow-AI, segmentation, cred-exposure, chains.""" import os -os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "test") +os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "k" * 32) from dataclasses import dataclass diff --git a/aegis/tests/test_p1_hardening.py b/aegis/tests/test_p1_hardening.py new file mode 100644 index 0000000..39c43ed --- /dev/null +++ b/aegis/tests/test_p1_hardening.py @@ -0,0 +1,131 @@ +"""P1 hardening suite: approval gate (#6), anchoring (#5), egress (#7), +pre-flight (#8), and audit-key strength (#4).""" +import os + +os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "k" * 32) + +import pytest + +from aegis import anchor, approval, egress, preflight +from aegis.config import DEFAULT_POLICY, Policy +from aegis.guardrail import AuditLog, GuardrailError + +KEY = "k" * 32 + + +# ---- #6 approval gate ------------------------------------------------------- +def test_approval_roundtrip_ok(): + tok = approval.mint("local", ["172.30.0.10", "172.30.0.11"], KEY, ttl=60) + # order/representation independence: same set, different order/form verifies. + approval.verify(tok, "local", ["172.30.0.11", "172.30.0.10"], KEY) + + +def test_approval_rejects_different_targets(): + tok = approval.mint("local", ["172.30.0.10"], KEY, ttl=60) + with pytest.raises(approval.ApprovalError): + approval.verify(tok, "local", ["172.30.0.99"], KEY) + + +def test_approval_rejects_extra_target(): + tok = approval.mint("local", ["172.30.0.10"], KEY, ttl=60) + with pytest.raises(approval.ApprovalError): + approval.verify(tok, "local", ["172.30.0.10", "172.30.0.11"], KEY) + + +def test_approval_rejects_wrong_key(): + tok = approval.mint("local", ["172.30.0.10"], KEY, ttl=60) + with pytest.raises(approval.ApprovalError): + approval.verify(tok, "local", ["172.30.0.10"], "z" * 32) + + +def test_approval_expired(): + tok = approval.mint("local", ["172.30.0.10"], KEY, ttl=10, now=1000) + with pytest.raises(approval.ApprovalError, match="expired"): + approval.verify(tok, "local", ["172.30.0.10"], KEY, now=2000) + + +def test_approval_missing_and_malformed(): + with pytest.raises(approval.ApprovalError): + approval.verify("", "local", ["172.30.0.10"], KEY) + with pytest.raises(approval.ApprovalError): + approval.verify("not-a-token", "local", ["172.30.0.10"], KEY) + + +# ---- #4 audit-key strength -------------------------------------------------- +def test_short_audit_key_refused(tmp_path, monkeypatch): + monkeypatch.setenv("PENTEST_AUDIT_HMAC_KEY", "tooshort") + pol = Policy.load(DEFAULT_POLICY) + object.__setattr__(pol, "audit_path", tmp_path / "a.ndjson") + with pytest.raises(GuardrailError, match="too short"): + AuditLog(pol) + + +# ---- #5 anchoring ----------------------------------------------------------- +def _audit(tmp_path, monkeypatch): + monkeypatch.setenv("PENTEST_AUDIT_HMAC_KEY", KEY) + pol = Policy.load(DEFAULT_POLICY) + object.__setattr__(pol, "audit_path", tmp_path / "audit.ndjson") + object.__setattr__(pol, "audit_anchor_path", tmp_path / "anchor.json") + return AuditLog(pol), pol + + +def test_anchor_tracks_chain_tip(tmp_path, monkeypatch): + log, _ = _audit(tmp_path, monkeypatch) + log.write({"event": "a"}) + log.write({"event": "b"}) + ok, reason = log.cross_check_anchor() + assert ok, reason + rec = anchor.read_anchor(log._anchor_path) + assert rec["seq"] == 2 and rec["tip"] == log._prev + + +def test_anchor_detects_log_rewrite(tmp_path, monkeypatch): + log, pol = _audit(tmp_path, monkeypatch) + log.write({"event": "a"}) + log.write({"event": "b"}) + # Attacker truncates the log to drop the last entry; anchor still records seq=2. + lines = pol.audit_path.read_text().splitlines() + pol.audit_path.write_text(lines[0] + "\n") + ok, reason = log.cross_check_anchor() + assert not ok and "mismatch" in reason.lower() + + +def test_anchor_detects_missing_anchor(tmp_path, monkeypatch): + log, _ = _audit(tmp_path, monkeypatch) + log.write({"event": "a"}) + log._anchor_path.unlink() + ok, reason = log.cross_check_anchor() + assert not ok and "missing" in reason.lower() + + +# ---- #7 egress rules -------------------------------------------------------- +def test_egress_ruleset_from_policy(): + pol = Policy.load(DEFAULT_POLICY) + rs = egress.nftables_ruleset(pol) + assert "policy drop" in rs # fail-closed default + assert "172.30.0.0/24" in rs # the lab allow-list CIDR + assert 'oifname "lo" accept' in rs + # a denied range is dropped explicitly + assert "10.0.0.0/8 drop" in rs + + +# ---- #8 pre-flight ---------------------------------------------------------- +def test_preflight_flags_public_ip(): + pol = Policy.load(DEFAULT_POLICY) + warns = preflight.check(["8.8.8.8"], pol) + assert any("PUBLIC IP" in w for w in warns) + + +def test_preflight_clean_for_inscope_lab_ip(): + pol = Policy.load(DEFAULT_POLICY) + # in-scope private lab IP with the tight default /24 policy → no target-level warning + warns = preflight.check(["172.30.0.10"], pol) + assert not any("PUBLIC IP" in w for w in warns) + + +def test_preflight_flags_broad_scope(tmp_path, monkeypatch): + import ipaddress + pol = Policy.load(DEFAULT_POLICY) + object.__setattr__(pol, "allowed_networks", (ipaddress.ip_network("172.30.0.0/16"),)) + warns = preflight.check(["172.30.0.10"], pol) + assert any("allow-list spans" in w for w in warns) diff --git a/aegis/tests/test_recon_modules.py b/aegis/tests/test_recon_modules.py index 2ee372c..d0640bd 100644 --- a/aegis/tests/test_recon_modules.py +++ b/aegis/tests/test_recon_modules.py @@ -1,7 +1,7 @@ """Modules 4 & 5 — shadow-AI discovery and segmentation validator (offline).""" import os -os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "test") +os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "k" * 32) from aegis.recon import segmentation, shadow_ai from aegis.tools import Observation diff --git a/aegis/tests/test_webrecon.py b/aegis/tests/test_webrecon.py index 4de2a0a..0264222 100644 --- a/aegis/tests/test_webrecon.py +++ b/aegis/tests/test_webrecon.py @@ -3,7 +3,7 @@ import pytest -os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "test") +os.environ.setdefault("PENTEST_AUDIT_HMAC_KEY", "k" * 32) from aegis.recon.web import (FixtureTransport, HttpResult, WebReconOrchestrator, probe) diff --git a/targets/scope-policy.yaml b/targets/scope-policy.yaml index 3941efb..2ea2bd5 100644 --- a/targets/scope-policy.yaml +++ b/targets/scope-policy.yaml @@ -59,6 +59,11 @@ audit: # Layer 6 — tamper-evident # Local-dev default (project-relative). For production, point at an OS-append-only # (chattr +a / WORM) path owned by a different uid than the tool runner. path: aegis/output/audit.ndjson + # Optional out-of-band anchor (#5): the chain tip {seq, tip, ts} is mirrored here after + # every entry. Point at a write-once store (S3 Object Lock / KMS object / WORM volume) + # the tool runner cannot rewrite; `aegis audit` cross-checks the live chain against it, + # so a full-log rewrite is detectable even if the HMAC key leaks. Off by default. + # anchor_path: /mnt/worm/argus/audit-anchor.json output_sanitizer: # Layer 7 redact_patterns: From 812950ebe8b73c7fb3b22d9c66cc319b69993bf4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 29 Jun 2026 01:59:59 +0000 Subject: [PATCH 2/2] feat: extend approval gate to --arm (completes #6) Per review feedback: #6 scoped the gate to --sandbox local AND --arm, but the first pass covered only local. Generalize the approval token to authorize a SET of modes ({local, arm}); cmd_scan now requires: - a `local`-scoped token for --sandbox local (un-isolated host exec), - an `arm`-scoped token for --arm (exploit-capable tools) in any sandbox, - a token authorizing BOTH when used together. Tokens are exact-match on the canonical mode set (a {local,arm} token does not satisfy a {local}-only requirement), fail closed, and dry-runs need none. `aegis approve [--mode local] [--mode arm] [--ttl N]` mints them. Adds combined-mode + arm-mode tests. 97 passed; ruff + Bandit clean. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01PH7w1wBoUxFESq9gh6yjAQ --- aegis/aegis/approval.py | 34 ++++++++++++++++--------- aegis/aegis/cli.py | 43 +++++++++++++++++++++----------- aegis/docs/NEXT_STEPS.md | 15 ++++++----- aegis/tests/test_p1_hardening.py | 16 ++++++++++++ 4 files changed, 75 insertions(+), 33 deletions(-) diff --git a/aegis/aegis/approval.py b/aegis/aegis/approval.py index 5c5b6d9..2940443 100644 --- a/aegis/aegis/approval.py +++ b/aegis/aegis/approval.py @@ -1,8 +1,8 @@ """Parameter-bound, fail-closed approval tokens for high-risk run modes (#6). -A token authorizes ONE mode (e.g. ``local``) against an EXACT set of targets until a -hard expiry. It is an HMAC over ``(mode, canonical-sorted-targets, expiry)`` keyed by -the audit signing key, so: +A token authorizes a specific SET of modes (e.g. ``local`` and/or ``arm``) against an +EXACT set of targets until a hard expiry. It is an HMAC over +``(canonical-modes, canonical-sorted-targets, expiry)`` keyed by the audit signing key, so: * only an operator who holds the key can mint one (the LLM/agent never can), * it cannot be replayed against different targets, and @@ -28,6 +28,15 @@ class ApprovalError(Exception): """Raised when an approval token is missing, malformed, expired, or mismatched.""" +def _norm_modes(modes: str | list[str]) -> list[str]: + return [modes] if isinstance(modes, str) else list(modes) + + +def _canon_modes(modes: str | list[str]) -> str: + """Canonical mode string: deduped + sorted so {arm,local} == {local,arm}.""" + return "+".join(sorted(set(_norm_modes(modes)))) + + def _canon_targets(targets: list[str]) -> str: """Normalize targets to canonical networks, deduped + sorted, so the token binds to the same set regardless of ordering or obfuscation (decimal/hex/leading-zero).""" @@ -35,25 +44,26 @@ def _canon_targets(targets: list[str]) -> str: return ",".join(nets) -def _body(mode: str, targets: list[str], expiry: int) -> str: - return f"v1|{mode}|{_canon_targets(targets)}|{expiry}" +def _body(modes: str | list[str], targets: list[str], expiry: int) -> str: + return f"v1|{_canon_modes(modes)}|{_canon_targets(targets)}|{expiry}" -def mint(mode: str, targets: list[str], key: str, *, ttl: int = 3600, +def mint(modes: str | list[str], targets: list[str], key: str, *, ttl: int = 3600, now: float | None = None) -> str: - """Mint a base64 approval token authorizing `mode` against `targets` for `ttl` seconds.""" + """Mint a base64 token authorizing `modes` against `targets` for `ttl` seconds.""" if ttl <= 0: raise ApprovalError("ttl must be positive") expiry = int((time.time() if now is None else now) + ttl) - mac = hmac.new(key.encode(), _body(mode, targets, expiry).encode(), hashlib.sha256).hexdigest() + mac = hmac.new(key.encode(), _body(modes, targets, expiry).encode(), hashlib.sha256).hexdigest() return base64.urlsafe_b64encode(f"{expiry}.{mac}".encode()).decode() -def verify(token: str, mode: str, targets: list[str], key: str, +def verify(token: str, modes: str | list[str], targets: list[str], key: str, *, now: float | None = None) -> None: - """Raise ApprovalError unless `token` authorizes exactly `mode`+`targets` and is unexpired.""" + """Raise ApprovalError unless `token` authorizes exactly `modes`+`targets` and is unexpired.""" if not token: - raise ApprovalError(f"{mode} mode requires an approval token (mint one with `aegis approve`)") + raise ApprovalError( + f"{_canon_modes(modes)} mode requires an approval token (mint one with `aegis approve`)") try: raw = base64.urlsafe_b64decode(token.encode()).decode() expiry_s, mac = raw.split(".", 1) @@ -63,7 +73,7 @@ def verify(token: str, mode: str, targets: list[str], key: str, now = time.time() if now is None else now if now > expiry: raise ApprovalError("approval token expired — mint a fresh one") - expect = hmac.new(key.encode(), _body(mode, targets, expiry).encode(), hashlib.sha256).hexdigest() + expect = hmac.new(key.encode(), _body(modes, targets, expiry).encode(), hashlib.sha256).hexdigest() # compare_digest over the full token guards target/mode tampering AND a forged MAC. if not hmac.compare_digest(mac, expect): raise ApprovalError("approval token does not authorize this mode/target set") diff --git a/aegis/aegis/cli.py b/aegis/aegis/cli.py index f2f89f0..de4b69c 100644 --- a/aegis/aegis/cli.py +++ b/aegis/aegis/cli.py @@ -37,22 +37,33 @@ def cmd_scan(args) -> int: if not d.allowed: print(f"REFUSED target {t}: {d.reason}", file=sys.stderr) return 2 + # Fail-closed, parameter-bound approval gate for high-risk modes — banners don't + # enforce (#6). `local` = un-isolated host exec; `arm` = exploit-capable tools. + # Dry-run executes nothing, so it needs no approval. + required = [] + if not args.dry_run: + if args.sandbox == "local": + required.append("local") + if getattr(args, "arm", None): + required.append("arm") + if required: + if "local" in required: + # Pre-flight: surface likely misconfiguration before any packet leaves (#8). + for w in preflight.check(args.targets, guard.policy): + print(f" preflight: {w}", file=sys.stderr) + try: + approval.verify(args.approval, required, args.targets, + os.environ.get(guard.policy.audit_key_env, "")) + except approval.ApprovalError as exc: + print(f"REFUSED [{'+'.join(sorted(set(required)))}]: {exc}", file=sys.stderr) + return 2 + if args.dry_run: sandbox = DryRunSandbox() elif args.sandbox == "local": print("WARNING: --sandbox local runs tools directly on THIS host with NO " "network isolation. Use only for AUTHORIZED off-lab recon under a " "tight, written-authorized scope policy.", file=sys.stderr) - # Pre-flight: surface likely misconfiguration before any packet leaves (#8). - for w in preflight.check(args.targets, guard.policy): - print(f" preflight: {w}", file=sys.stderr) - # Fail-closed, parameter-bound approval gate — banners don't enforce (#6). - try: - approval.verify(args.approval, "local", args.targets, - os.environ.get(guard.policy.audit_key_env, "")) - except approval.ApprovalError as exc: - print(f"REFUSED --sandbox local: {exc}", file=sys.stderr) - return 2 try: sandbox = LocalSandbox(audit_key_env=guard.policy.audit_key_env) except RuntimeError as exc: @@ -213,10 +224,11 @@ def cmd_approve(args) -> int: if not d.allowed: print(f"REFUSED target {t}: {d.reason}", file=sys.stderr) return 2 - token = approval.mint(args.mode, args.targets, key, ttl=args.ttl) + modes = args.mode or ["local"] + token = approval.mint(modes, args.targets, key, ttl=args.ttl) print(token) - print(f"# authorizes `{args.mode}` against {args.targets} for {args.ttl}s", - file=sys.stderr) + print(f"# authorizes [{'+'.join(sorted(set(modes)))}] against {args.targets} " + f"for {args.ttl}s", file=sys.stderr) return 0 @@ -322,9 +334,10 @@ def build_parser() -> argparse.ArgumentParser: a.set_defaults(func=cmd_audit) ap = sub.add_parser("approve", - help="mint a parameter-bound approval token for --sandbox local (#6)") - ap.add_argument("mode", choices=["local"]) + help="mint a parameter-bound approval token for --sandbox local / --arm (#6)") ap.add_argument("targets", nargs="+") + ap.add_argument("--mode", action="append", choices=["local", "arm"], + help="mode(s) this token authorizes (repeatable; default: local)") ap.add_argument("--ttl", type=int, default=3600, help="token lifetime in seconds (default 3600)") ap.set_defaults(func=cmd_approve) diff --git a/aegis/docs/NEXT_STEPS.md b/aegis/docs/NEXT_STEPS.md index 155007d..7524351 100644 --- a/aegis/docs/NEXT_STEPS.md +++ b/aegis/docs/NEXT_STEPS.md @@ -71,12 +71,15 @@ mode / KMS-signed object / WORM volume) owned by a different uid — the file fo > storage and external anchoring on top." — Tracehold; SystemsHardening audit-logging > architecture (CloudTrail-style log-file validation). -### ✅ 6. Risk-tiered approval gate for `--sandbox local` -`--sandbox local` now requires a **parameter-bound, fail-closed approval token** (`approval.py`): -an HMAC over `(mode, canonical-sorted-targets, expiry)` keyed by the audit key. Mint with -`aegis approve local [--ttl N]`; pass via `--approval`/`ARGUS_APPROVAL_TOKEN`. A token -cannot be replayed against different targets or after expiry, and wildcard grants are impossible -(every token names its targets). The banner alone is gone — no token, no run. +### ✅ 6. Risk-tiered approval gate for `--sandbox local` / `--arm` +Both high-risk modes now require a **parameter-bound, fail-closed approval token** (`approval.py`): +an HMAC over `(canonical-modes, canonical-sorted-targets, expiry)` keyed by the audit key. +`--sandbox local` requires a `local`-scoped token; `--arm` (exploit-capable tools) requires an +`arm`-scoped token in any sandbox; using both requires a token authorizing both. Mint with +`aegis approve [--mode local] [--mode arm] [--ttl N]`; pass via +`--approval`/`ARGUS_APPROVAL_TOKEN`. A token cannot be replayed against different targets/modes +or after expiry, and wildcard grants are impossible (every token names its exact targets + +modes). Dry-runs need no token (they execute nothing). The banner alone is gone — no token, no run. > *Best practice:* "system-prompt guardrails don't guard anything… agents take risky actions > 23.9% of the time even with explicit safety instructions" (ROE Gate); "bind approval to the diff --git a/aegis/tests/test_p1_hardening.py b/aegis/tests/test_p1_hardening.py index 39c43ed..e3c50c8 100644 --- a/aegis/tests/test_p1_hardening.py +++ b/aegis/tests/test_p1_hardening.py @@ -38,6 +38,22 @@ def test_approval_rejects_wrong_key(): approval.verify(tok, "local", ["172.30.0.10"], "z" * 32) +def test_approval_combined_modes_exact_match(): + tok = approval.mint(["local", "arm"], ["172.30.0.10"], KEY, ttl=60) + # order-independent within the set + approval.verify(tok, ["arm", "local"], ["172.30.0.10"], KEY) + # a {local,arm} token must NOT satisfy a {local}-only requirement (exact match, fail closed) + with pytest.raises(approval.ApprovalError): + approval.verify(tok, ["local"], ["172.30.0.10"], KEY) + + +def test_approval_arm_mode_distinct_from_local(): + tok = approval.mint("arm", ["172.30.0.10"], KEY, ttl=60) + approval.verify(tok, "arm", ["172.30.0.10"], KEY) + with pytest.raises(approval.ApprovalError): + approval.verify(tok, "local", ["172.30.0.10"], KEY) + + def test_approval_expired(): tok = approval.mint("local", ["172.30.0.10"], KEY, ttl=10, now=1000) with pytest.raises(approval.ApprovalError, match="expired"):