feat(gateway): enforce image_digest bans at cage run create + claim (#1678)

The agent_bans store has accepted image_digest targets since migration
0029 but never consulted them. A banned sandbox image digest now blocks
fail-closed (403) at both moments that matter:

- POST /v1/agent-cage/runs — when cage_spec.image_digest is present,
  registration is denied before the run record or start command exist;
- POST /v1/agent-cage/runs/:id/claim — re-checked against the digest
  stored on the run record, so a ban created after registration still
  blocks the sandbox from ever starting.

Storage errors during the check are 500s (unreadable enforcement state
blocks). Tests written RED-first: banned digest denied at create,
late ban denied at claim, unrelated digest ban does not block.

Honesty ledger: fingerprint has no carrier field at any choke point
today (mTLS yields CN only — needs a protobuf-first contract change)
and prompt_hash appears only at telemetry ingest, where blocking would
destroy evidence (enforcement belongs to SOC auto-response). Both stay
stored-but-unenforced, now documented as such with the reason.

Author: lavkushry

.claude/PRPs/tasks/task.md

@@ -69,8 +69,10 @@ Blocks multi-replica and enterprise-ops claims.
 - [x] **Ban/quarantine enforcement** at all choke points (authorize, broker,
       cage start/claim, egress) + agent-ban propagation to live runs as signed
       `kill_run` commands — see `feat/ban-enforcement-choke-points`.
-      Remaining: `fingerprint`/`image_digest`/`prompt_hash` ban target types
-      are stored but not yet consulted anywhere.
+      `image_digest` bans now enforced at cage run create + claim
+      (`feat/ban-image-digest-enforcement`). Remaining: `fingerprint` (no
+      carrier field at any choke point yet) and `prompt_hash` (telemetry-only;
+      enforcement = SOC auto-response design) stay stored-but-unenforced.
 - [ ] **Standalone MCP proxy binary** (optional product line; Lite remains prod).
 - [ ] **Branch hygiene** — merge-or-close pass on ~80 remote branches (no bulk
       delete without one-by-one review).

docs/Implementation_Status.md

@@ -64,7 +64,7 @@ and is not production-authoritative.
 | Agent runs registry (cage control-plane API) | Partial | `lib/storage/src/db/agent_runs.rs`, migration `0026`; `routes/runtime.rs` (`/v1/agent-cage/runs` + controls) | claim/heartbeat/lease APIs + executor not on main (WIP branch) | #1681 | route + storage | beta | cage execution PR |
 | Runtime events ingest + timeline | Partial | `POST /v1/ingest/runtime-events`, `list/query` ASE APIs | rich producers (process/fs/net) on sensor | #1681 | storage + route | beta | sensor collectors |
 | Control commands (signed kill/pause/quarantine) | Partial | store + protocol + gateway issue routes; sensor poll/verify + **host ProcessEnforcer** (SIGTERM/STOP/CONT for registered PIDs); cage-runner Docker kill path | auto PID discovery/collectors; grace_period from command payload | Phased PR plan §5 | storage + sensor unit (real child kill) | beta | collectors |
-| Ban system (first-class store) | Implemented | `lib/storage/src/db/agent_bans.rs`, migration `0029`; enforced at every choke point: `POST /v1/authorize` (agent + tool bans → durable deterministic deny via `write_decision_and_audit`), `POST /v1/broker/execute` (tool ban → 403 before approval consumption), `POST /v1/agent-cage/runs` create + claim (agent ban → 403; claim re-checks so a ban created after registration still blocks start), `POST /v1/egress/check` (destination ban, pre-existing); sensor prop: `POST /v1/bans` with `target_type=agent` issues signed `kill_run` control commands for the agent's active runs (`list_active_agent_runs_for_agent`) | `fingerprint`/`image_digest`/`prompt_hash` target types stored but not yet consulted at a choke point | #1678 | storage + route (authorize/broker/runtime/control/egress) | beta | remaining target types |
+| Ban system (first-class store) | Implemented | `lib/storage/src/db/agent_bans.rs`, migration `0029`; enforced at every choke point: `POST /v1/authorize` (agent + tool bans → durable deterministic deny via `write_decision_and_audit`), `POST /v1/broker/execute` (tool ban → 403 before approval consumption), `POST /v1/agent-cage/runs` create + claim (agent ban → 403; claim re-checks so a ban created after registration still blocks start), `POST /v1/egress/check` (destination ban, pre-existing); sensor prop: `POST /v1/bans` with `target_type=agent` issues signed `kill_run` control commands for the agent's active runs (`list_active_agent_runs_for_agent`) | `image_digest` bans enforced at cage run create + claim (`image_digest_ban_denial`, re-checked at claim so a ban created after registration still blocks start); `fingerprint`/`prompt_hash` target types stored but not consulted — `fingerprint` has no carrier at any choke point today (mTLS yields CN only; needs a wire field, i.e. a protobuf-first contract change), `prompt_hash` only appears at telemetry ingest where blocking would destroy evidence (enforcement belongs to SOC auto-response, a separate design) | #1678 | storage + route (authorize/broker/runtime/control/egress) | beta | remaining target types |
 | Quarantine records | Partial | `lib/storage/src/db/quarantine.rs`, migration `0030`; agent-status quarantine; `quarantine_records` enforced at `POST /v1/authorize` (agent), `POST /v1/broker/execute` (tool), cage run create + claim (agent + run), egress (run/sandbox/agent, pre-existing) | workspace/sandbox evidence-freeze semantics (needs cage) | #1679 | storage + route | beta | cage integration |
 | Node sensor | Implemented | `bins/aegis-node-sensor` (main, spool, shipper, command_receiver); real `process`/`net`/`fs`/`secret` collectors (`AEGIS_RUN_ID`-tagged process discovery -> `ProcessEnforcer` registration + `network_connection`/filesystem/secret-signal runtime events), all polled from the main loop; Dockerfile, Helm, compose; `tests/real_host_integration.rs` proves `scan_host_aegis_processes`/`ProcessCollector`/`NetCollector`/`FsCollector`/`SecretCollector` against a real Linux host's actual `/proc`, a real signal-killed child, a real established TCP socket, a real open file descriptor, and a real secret-shaped env var (name reported, value never leaves the host) -- not the synthetic `/proc` tempdir fixtures the module unit tests use; `scripts/sensor-soak.sh` + CI `sensor-soak.yml` (nightly 15-min + dispatchable multi-hour + PR smoke) soak the real sensor on the runner's real `/proc` under continuous collector workload, signed-kill round-trips, and a mid-soak gateway outage, asserting bounded RSS/fds/spool disk, zero panics, full spool drain, and unattended recovery; steady-state spool compaction (`compact_if_reclaimable`, wired into the ship tick) keeps lane files from growing without bound on long-lived hosts | extended-duration soak on a production-grade deployment (the nightly CI soak runs on a CI runner, not a production host under production load) | Phased PR plan §5 | unit (89 tests) + 5 real-host integration tests + nightly CI soak | beta | production-deployment soak |
 | Agent cage runner | Partial | binary + DockerRuntime + Dockerfile + compose `cage` + Helm; claim lifecycle smoke; host-Docker review + hardened create; `scripts/cage-docker-e2e.sh` + CI job (quick finish + signed kill against real Docker) | sensor↔runner IPC; forced egress netns; product “untrusted→incident” narrative e2e | Phased PR plan §6 | unit (lib) + claim lifecycle + cage-docker-e2e + helm lint | beta (local/k8s) | forced egress + sensor enforce |

src/src/routes/runtime.rs

@@ -323,6 +323,31 @@ async fn agent_ban_denial(
     }
 }
 
+/// Image-digest ban enforcement shared by run registration and claim
+/// (#1678: ban target types beyond agent/tool). A banned sandbox image is
+/// blocked before it ever starts; a storage error is a 500 (fail-closed —
+/// unreadable enforcement state blocks).
+async fn image_digest_ban_denial(
+    state: &AppState,
+    tenant_id: &str,
+    image_digest: &str,
+) -> Result<Option<StatusError>, StatusError> {
+    match state
+        .storage
+        .is_banned(tenant_id, "image_digest", image_digest, Utc::now())
+        .await
+    {
+        Err(e) => {
+            error!("Failed to check image-digest ban state: {:?}", e);
+            Err(StatusError::internal("Database error"))
+        }
+        Ok(true) => Ok(Some(StatusError::forbidden(
+            "sandbox image digest is banned; runs are denied (fail-closed)",
+        ))),
+        Ok(false) => Ok(None),
+    }
+}
+
 pub async fn create_agent_run(
     State(state): State<Arc<AppState>>,
     TenantId(tenant_id): TenantId,
@@ -342,6 +367,17 @@ pub async fn create_agent_run(
             Err(e) => return e.into_response(),
         }
     }
+    if let Some(digest) = req
+        .cage_spec
+        .as_ref()
+        .and_then(|spec| spec.image_digest.as_deref())
+    {
+        match image_digest_ban_denial(&state, &tenant_id, digest).await {
+            Ok(Some(denial)) => return denial.into_response(),
+            Ok(None) => {}
+            Err(e) => return e.into_response(),
+        }
+    }
 
     let mut image_ref = None;
     let mut image_digest = None;
@@ -490,6 +526,13 @@ pub async fn claim_run(
             Err(e) => return e.into_response(),
         }
     }
+    if let Some(digest) = run.image_digest.as_deref() {
+        match image_digest_ban_denial(&state, &tenant_id, digest).await {
+            Ok(Some(denial)) => return denial.into_response(),
+            Ok(None) => {}
+            Err(e) => return e.into_response(),
+        }
+    }
     match state
         .storage
         .is_quarantined(&tenant_id, "run", &run_id)
@@ -1822,14 +1865,19 @@ mod tests {
 
     // ── Ban / quarantine enforcement at cage start (Phase 2.4/2.5) ──────────
 
-    async fn ban_agent(state: &Arc<AppState>, tenant_id: &str, agent_id: &str) {
+    async fn ban_target(
+        state: &Arc<AppState>,
+        tenant_id: &str,
+        target_type: &str,
+        target_value: &str,
+    ) {
         state
             .storage
             .insert_ban(&AgentBanRecord {
                 id: Uuid::new_v4().to_string(),
                 tenant_id: tenant_id.to_string(),
-                target_type: "agent".to_string(),
-                target_value: agent_id.to_string(),
+                target_type: target_type.to_string(),
+                target_value: target_value.to_string(),
                 scope: "tenant".to_string(),
                 reason: Some("test ban".to_string()),
                 actor: "test-operator".to_string(),
@@ -1843,6 +1891,10 @@ mod tests {
             .unwrap();
     }
 
+    async fn ban_agent(state: &Arc<AppState>, tenant_id: &str, agent_id: &str) {
+        ban_target(state, tenant_id, "agent", agent_id).await;
+    }
+
     async fn quarantine_target(
         state: &Arc<AppState>,
         tenant_id: &str,
@@ -1918,6 +1970,125 @@ mod tests {
         assert_eq!(response.status(), StatusCode::FORBIDDEN);
     }
 
+    /// Registering a cage run whose `image_digest` is banned is a fail-closed
+    /// 403 — the `agent_bans` store promises image-digest enforcement
+    /// "before sandbox start" (#1678: target types beyond agent/tool).
+    #[tokio::test]
+    async fn create_agent_run_denies_banned_image_digest() {
+        let (state, tenant_id, _agent_token) = setup_state_with_command_signing_key(
+            "cage_create_banned_image",
+            TEST_COMMAND_SIGNING_SECRET_HEX,
+        )
+        .await;
+        let digest = "sha256:feedfacefeedfacefeedfacefeedfacefeedfacefeedfacefeedfacefeedface";
+        ban_target(&state, &tenant_id, "image_digest", digest).await;
+
+        let mut spec = sample_cage_spec();
+        spec.image_digest = Some(digest.to_string());
+        let response = create_agent_run(
+            State(state.clone()),
+            TenantId(tenant_id.clone()),
+            Json(CreateAgentRunRequest {
+                run_key: "run-banned-image".to_string(),
+                agent_id: None,
+                source_component: "cage-runner".to_string(),
+                mode: None,
+                root_trace_id: None,
+                root_trust_level: None,
+                cage_spec: Some(spec),
+            }),
+        )
+        .await
+        .into_response();
+        assert_eq!(response.status(), StatusCode::FORBIDDEN);
+    }
+
+    /// A ban on a different image digest must not block an unrelated run.
+    #[tokio::test]
+    async fn create_agent_run_allows_unrelated_image_digest_ban() {
+        let (state, tenant_id, _agent_token) = setup_state_with_command_signing_key(
+            "cage_create_other_image_ban",
+            TEST_COMMAND_SIGNING_SECRET_HEX,
+        )
+        .await;
+        ban_target(
+            &state,
+            &tenant_id,
+            "image_digest",
+            "sha256:0000000000000000000000000000000000000000000000000000000000000000",
+        )
+        .await;
+
+        let mut spec = sample_cage_spec();
+        spec.image_digest = Some(
+            "sha256:1111111111111111111111111111111111111111111111111111111111111111".to_string(),
+        );
+        let response = create_agent_run(
+            State(state.clone()),
+            TenantId(tenant_id.clone()),
+            Json(CreateAgentRunRequest {
+                run_key: "run-other-image".to_string(),
+                agent_id: None,
+                source_component: "cage-runner".to_string(),
+                mode: None,
+                root_trace_id: None,
+                root_trust_level: None,
+                cage_spec: Some(spec),
+            }),
+        )
+        .await
+        .into_response();
+        assert_eq!(response.status(), StatusCode::CREATED);
+    }
+
+    /// An image-digest ban created AFTER the run was registered still blocks
+    /// the claim — claiming is the moment execution actually starts, so it
+    /// re-checks against the digest stored on the run record.
+    #[tokio::test]
+    async fn claim_run_denies_image_digest_banned_after_registration() {
+        let (state, tenant_id, _agent_token) = setup_state_with_command_signing_key(
+            "cage_claim_banned_image",
+            TEST_COMMAND_SIGNING_SECRET_HEX,
+        )
+        .await;
+        let digest = "sha256:deadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef";
+
+        let mut spec = sample_cage_spec();
+        spec.image_digest = Some(digest.to_string());
+        let response = create_agent_run(
+            State(state.clone()),
+            TenantId(tenant_id.clone()),
+            Json(CreateAgentRunRequest {
+                run_key: "run-claim-image-ban".to_string(),
+                agent_id: None,
+                source_component: "cage-runner".to_string(),
+                mode: None,
+                root_trace_id: None,
+                root_trust_level: None,
+                cage_spec: Some(spec),
+            }),
+        )
+        .await
+        .into_response();
+        assert_eq!(response.status(), StatusCode::CREATED);
+        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        let run: AgentRunRecord = serde_json::from_slice(&body).unwrap();
+
+        ban_target(&state, &tenant_id, "image_digest", digest).await;
+
+        let response = claim_run(
+            State(state.clone()),
+            TenantId(tenant_id.clone()),
+            Path(run.id.clone()),
+            Json(RunnerIdRequest {
+                runner_id: "runner-1".to_string(),
+            }),
+        )
+        .await
+        .into_response();
+        assert_eq!(response.status(), StatusCode::FORBIDDEN);
+    }
+
     /// A ban created AFTER the run was registered still blocks the claim —
     /// claiming is the moment execution actually starts, so it re-checks.
     #[tokio::test]