feat: P1 hardening suite — approval gate, audit anchoring, egress, pre-flight, key-strength

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

aegis/aegis/approval.py

@@ -0,0 +1,79 @@
+"""Parameter-bound, fail-closed approval tokens for high-risk run modes (#6).
+
+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
+  * 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 _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)."""
+    nets = sorted({str(canon_network(t)) for t in targets})
+    return ",".join(nets)
+
+
+def _body(modes: str | list[str], targets: list[str], expiry: int) -> str:
+    return f"v1|{_canon_modes(modes)}|{_canon_targets(targets)}|{expiry}"
+
+
+def mint(modes: str | list[str], targets: list[str], key: str, *, ttl: int = 3600,
+         now: float | None = None) -> str:
+    """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(modes, targets, expiry).encode(), hashlib.sha256).hexdigest()
+    return base64.urlsafe_b64encode(f"{expiry}.{mac}".encode()).decode()
+
+
+def verify(token: str, modes: str | list[str], targets: list[str], key: str,
+           *, now: float | None = None) -> None:
+    """Raise ApprovalError unless `token` authorizes exactly `modes`+`targets` and is unexpired."""
+    if not token:
+        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)
+        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(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")

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
@@ -35,6 +37,27 @@ 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":
@@ -180,9 +203,41 @@ 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
+    modes = args.mode or ["local"]
+    token = approval.mint(modes, args.targets, key, ttl=args.ttl)
+    print(token)
+    print(f"# authorizes [{'+'.join(sorted(set(modes)))}] against {args.targets} "
+          f"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 +267,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 +330,21 @@ 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 / --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)
+
+    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

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,
         )

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"

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"