diff --git a/CLAUDE.md b/CLAUDE.md index b563925b..4c94dd94 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -38,7 +38,7 @@ A run executes in one of **three execution modes** behind the one `LLMProvider` engine is identical across all three. See [ADR-0012](docs/decisions/0012-managed-inference-dual-mode.md) to [ADR-0015](docs/decisions/0015-managed-mode-data-handling-and-compliance.md) and [docs/architecture/managed-inference.md](docs/architecture/managed-inference.md). -**Status:** Phase 1 in progress — milestone M1 (LLM seam proven) reached (PR #9, 2026-06-07); the `FallbackChain` runner (1.K) landed, completing 1.m2 with the cost tracker (PR #13, 2026-06-11); the run loop (1.N — `WorkflowEngine` + `RunEventBus`) landed (PR #17, 2026-06-13) **completing 1.m3** (parse → DAG → run loop emits the canonical event stream), with the built-in `ToolRegistry` (1.T, a 1.m4 component) landing alongside it as the other `AgentRunner` (1.O) join prerequisite; the **`AgentRunner` (1.O) — per-node LLM execution behind the seam — landed (PR #18, 2026-06-14)**; and the **node-type handlers (1.P) — the six non-agent `NodeExecutor` arms (condition / transform / fan_out / fan_in / input / output) behind a dispatching executor — landed (PR #20, 2026-06-14)**; and **checkpoint/resume (1.R) + the human gate (1.Q) landed (PR #22, 2026-06-15)** — the derived `Checkpointer` + cross-process `resumeFromCheckpoint`, and the `human_in_the_loop` gate with the one-shot timeout port; and **node retry (1.S) — the above-chain whole-node retry budget ([ADR-0040](docs/decisions/0040-node-retry-budget-above-the-chain.md), amending ADR-0038) — landed (PR #24, 2026-06-15)**, re-dispatching a whole node on a retryable failure up to `retry.max` attempts (with `node:retrying`, abort-aware backoff, and `retry_on` filtering), with retry-from-node (ADR-0040 Part B) deferred to Phase-2; and the **pre-egress budget governor (1.AC, [ADR-0028](docs/decisions/0028-workflow-resource-governance.md)) + the `AgentSession` agent-first entry point (1.V, [ADR-0024](docs/decisions/0024-agent-first-entry-point-agentsession.md)) landed together (PR #26, 2026-06-16)** — 1.AC was the last 1.m4 component, so **1.m4 is complete** (the full engine stack: node handlers, gate, checkpoint/resume, retry, tools, sandbox, budget governor), and 1.V opens the Lane-C agent-first sub-spine (1.m5); then the **end-to-end Node harness (1.U) landed (PR #27, 2026-06-16), reaching 🎯 M2** — the engine runs end-to-end (live streaming + per-node-boundary checkpointing + cross-process resume + node retry + provider failover, gap-free), **completing the Phase-1 engine critical path**. The remaining Phase-1 work is additive and off the critical path (Lane C: the **`session:*` namespace (1.W) landed (PR #28, 2026-06-17)** — the `SessionEventSink`→`RunEventBus` adapter + per-session `sequenceNumber`, the `SessionHandle`, and the combined `RunOrSessionEventSchema` gate — and **session persistence (1.X) landed (PR #29, 2026-06-17)** — the `agent_sessions`/`session_messages` tables + migration, `SessionMessageSchema`/`AgentSessionSchema`, and the `SessionStore` + domain↔row mappers (data-layer only); then **session checkpoint/resume (1.Y) + export-to-workflow (1.Z) landed (PR #30, 2026-06-17)** — `reconstructSessionState`/`AgentSession.resume` (reload-not-replay; preload the text-only transcript, re-seed turnCount/cost, no `session:started` re-emit) + the `serializeWorkflow`/`sessionToWorkflow` pair (one agent node per completed turn, transcript in `metadata`, secret/signature exclusion structural) — leaving only the **1.AA** chat-regression harness ‖ the 1.m6 multimodal sub-spine, whose first step — **media-input adapters + the shared SSRF policy primitive (1.AE) — landed (PR #32, 2026-06-18)** (base64 image/audio across the three adapters behind per-modality `assertMediaCapabilities`, and the one `@relavium/shared` SSRF policy primitive reused by the provider-`baseURL` + `http_request` callers; the SSRF *mechanism* half + per-modality FallbackChain gating moved into **1.AF**). **1.AF (engine media plumbing) is now in progress — PR #33** lands its P1+P2: the `MediaStore`/`deInlineMedia` choke point (the active I3 enforcement), per-modality capability gating (`mediaSupportReason`/`requestSupportReason` — replacing coarse vision), the `media_objects`/`media_references` tables, and ADR-0042/0043/0044; P3/P4 (egress + access/cost) follow. **Phase 2 (CLI, M3) is unblocked**. +**Status:** Phase 1 in progress — milestone M1 (LLM seam proven) reached (PR #9, 2026-06-07); the `FallbackChain` runner (1.K) landed, completing 1.m2 with the cost tracker (PR #13, 2026-06-11); the run loop (1.N — `WorkflowEngine` + `RunEventBus`) landed (PR #17, 2026-06-13) **completing 1.m3** (parse → DAG → run loop emits the canonical event stream), with the built-in `ToolRegistry` (1.T, a 1.m4 component) landing alongside it as the other `AgentRunner` (1.O) join prerequisite; the **`AgentRunner` (1.O) — per-node LLM execution behind the seam — landed (PR #18, 2026-06-14)**; and the **node-type handlers (1.P) — the six non-agent `NodeExecutor` arms (condition / transform / fan_out / fan_in / input / output) behind a dispatching executor — landed (PR #20, 2026-06-14)**; and **checkpoint/resume (1.R) + the human gate (1.Q) landed (PR #22, 2026-06-15)** — the derived `Checkpointer` + cross-process `resumeFromCheckpoint`, and the `human_in_the_loop` gate with the one-shot timeout port; and **node retry (1.S) — the above-chain whole-node retry budget ([ADR-0040](docs/decisions/0040-node-retry-budget-above-the-chain.md), amending ADR-0038) — landed (PR #24, 2026-06-15)**, re-dispatching a whole node on a retryable failure up to `retry.max` attempts (with `node:retrying`, abort-aware backoff, and `retry_on` filtering), with retry-from-node (ADR-0040 Part B) deferred to Phase-2; and the **pre-egress budget governor (1.AC, [ADR-0028](docs/decisions/0028-workflow-resource-governance.md)) + the `AgentSession` agent-first entry point (1.V, [ADR-0024](docs/decisions/0024-agent-first-entry-point-agentsession.md)) landed together (PR #26, 2026-06-16)** — 1.AC was the last 1.m4 component, so **1.m4 is complete** (the full engine stack: node handlers, gate, checkpoint/resume, retry, tools, sandbox, budget governor), and 1.V opens the Lane-C agent-first sub-spine (1.m5); then the **end-to-end Node harness (1.U) landed (PR #27, 2026-06-16), reaching 🎯 M2** — the engine runs end-to-end (live streaming + per-node-boundary checkpointing + cross-process resume + node retry + provider failover, gap-free), **completing the Phase-1 engine critical path**. The remaining Phase-1 work is additive and off the critical path (Lane C: the **`session:*` namespace (1.W) landed (PR #28, 2026-06-17)** — the `SessionEventSink`→`RunEventBus` adapter + per-session `sequenceNumber`, the `SessionHandle`, and the combined `RunOrSessionEventSchema` gate — and **session persistence (1.X) landed (PR #29, 2026-06-17)** — the `agent_sessions`/`session_messages` tables + migration, `SessionMessageSchema`/`AgentSessionSchema`, and the `SessionStore` + domain↔row mappers (data-layer only); then **session checkpoint/resume (1.Y) + export-to-workflow (1.Z) landed (PR #30, 2026-06-17)** — `reconstructSessionState`/`AgentSession.resume` (reload-not-replay; preload the text-only transcript, re-seed turnCount/cost, no `session:started` re-emit) + the `serializeWorkflow`/`sessionToWorkflow` pair (one agent node per completed turn, transcript in `metadata`, secret/signature exclusion structural) — leaving only the **1.AA** chat-regression harness ‖ the 1.m6 multimodal sub-spine, whose first step — **media-input adapters + the shared SSRF policy primitive (1.AE) — landed (PR #32, 2026-06-18)** (base64 image/audio across the three adapters behind per-modality `assertMediaCapabilities`, and the one `@relavium/shared` SSRF policy primitive reused by the provider-`baseURL` + `http_request` callers; the SSRF *mechanism* half + per-modality FallbackChain gating moved into **1.AF**). **1.AF (engine media plumbing) is in progress: P1+P2 merged (PR #33)** — the `MediaStore`/`deInlineMedia` choke point (the active I3 enforcement), per-modality capability gating (`mediaSupportReason`/`requestSupportReason` — replacing coarse vision), the `media_objects`/`media_references` tables, and ADR-0042/0043/0044 — and **P3 + P4/D13 have landed on `development` in a follow-on PR (pending merge):** the binary media-egress + the **SSRF mechanism half** (D9 — the `MediaUrlFetch` re-host hook + the `fetchMediaBytes` SSRF-validated host reference wired at the choke point), the `FallbackChain` resolve-before-egress + the **byte-free re-materialization sidecar** (D8/D7), and the byte-delivery **`Range` gate** (D13 — `MediaStore.readRange` + `validateByteRange`), with the dedicated **P3 egress/SSRF security-review clean** (0 blockers/highs). The rest of **P4** follows: `read_media` + the scope-set authz (D12), the terminal sweep + GC (D11), `save_to` (D16), the `output_modalities` load-check (D15), the per-modality media cost governor (D17), the byte-delivery security-review + the keychain IPC test. **Phase 2 (CLI, M3) is unblocked**. Phase 0 (M0, 2026-06-04) landed the monorepo, strict toolchain + CI, `@relavium/shared` (the full Zod contract set), the no-vendor-type seam fence, and `@relavium/db`. Phase 1 has since landed `@relavium/llm` — the `LLMProvider` seam + all three adapters (Anthropic, OpenAI/DeepSeek, @@ -64,7 +64,7 @@ The pre-egress budget governor (1.AC) + the agent-first `AgentSession` (1.V) lan **completing 1.m4**; then the end-to-end Node harness (1.U) landed (PR #27, 2026-06-16) **reaching M2** — the Phase-1 engine critical path is complete. The additive Lane-C agent-first sub-spine is now **complete** (session events **1.W ✅ (PR #28)** + persistence **1.X ✅ (PR #29)** + checkpoint/resume **1.Y** & export **1.Z ✅ (PR #30, 2026-06-17)** + the **1.AA** chat-regression harness ✅ (2026-06-17), closing **1.m5**); -on the 1.m6 multimodal sub-spine, **media-input adapters + the shared SSRF policy primitive (1.AE) landed (PR #32, 2026-06-18)** — after a multi-round + final 8-dimension adversarial review (no SSRF bypass found); the SSRF *mechanism* half + per-modality gating moved into 1.AF. **1.AF (engine media plumbing) is now in progress — PR #33** lands P1+P2 (the `MediaStore`/`deInlineMedia` choke point, per-modality gating, the media tables + ADR-0042/0043/0044); P3/P4 (egress + access/cost) follow. The remaining Phase-1 work is 1.AF (in progress) + 1.AG/1.AH; Phase 2 (CLI) is unblocked. See +on the 1.m6 multimodal sub-spine, **media-input adapters + the shared SSRF policy primitive (1.AE) landed (PR #32, 2026-06-18)** — after a multi-round + final 8-dimension adversarial review (no SSRF bypass found); the SSRF *mechanism* half + per-modality gating moved into 1.AF. **1.AF (engine media plumbing) is in progress: P1+P2 merged (PR #33)**; **P3 + P4/D13 landed on `development` (pending merge)** — the media-egress + SSRF mechanism half (D9), the `FallbackChain` resolve-before-egress + byte-free re-materialization sidecar (D8/D7), the byte-delivery `Range` gate (D13), and a clean P3 egress/SSRF security-review — with the rest of P4 (D12 `read_media` + authz, D11 GC sweep, D16 `save_to`, D15 load-check, D17 cost governor) to follow. The remaining Phase-1 work is 1.AF (in progress) + 1.AG/1.AH; Phase 2 (CLI) is unblocked. See [docs/roadmap/current.md](docs/roadmap/current.md). See [README.md](README.md) for the public overview. ## Non-negotiable rules for AI agents diff --git a/docs/roadmap/current.md b/docs/roadmap/current.md index f250b933..9fe90fba 100644 --- a/docs/roadmap/current.md +++ b/docs/roadmap/current.md @@ -217,18 +217,26 @@ harness is now ✅ Done (2026-06-17), completing **1.m5**; cost-event persistenc > **media-input adapters + the shared SSRF policy primitive (1.AE) are ✅ Done (PR #32, 2026-06-18)** — > after a multi-round + final 8-dimension adversarial review (no SSRF bypass found); the SSRF *mechanism* > half (host DNS-resolve + connect-by-validated-IP), per-modality FallbackChain gating, `mediaUnits`, and -> handle/url media resolution are deferred to **1.AF**. **1.AF (engine media plumbing) is 🔨 in progress on -> `development`** — its three design ADRs (0042/0043/0044) are Accepted, and **P1 + P2 have landed (NOT yet -> merged):** the `MediaStore` contract impls + `media_objects`/`media_references` tables (migration 0002), -> per-modality capability gating + `FallbackChain` skip, `deInlineMedia`, the strict `output_modalities`/ -> `save_to` node fields, OpenAI `mediaUnits`, and **the I3 keystone — `deInlineMedia` at the one -> `#emitDurable` choke point** (all green, Leakwatch-clean). **Remaining (P3 + P4, some security-critical):** -> the binary media-egress + SSRF mechanism half (D9), `read_media` + the byte-delivery gate (D12/D13), -> the failover sidecar (D7), adapter handle/url resolution (D8), the terminal sweep (D11), the -> `output_modalities` load-check (D15), the `save_to` write port (D16), and the per-modality media cost -> governor (D17) — D9 and D12/D13 want a dedicated security-review pass. The remaining Phase-1 work is -> **additive and off the critical path**: only **1.AF–1.AH** (engine media plumbing, output, surfaces) remain -> before Phase 1 closes. **Phase 2 (CLI, milestone M3) is unblocked.** +> handle/url media resolution are deferred to **1.AF**. **1.AF (engine media plumbing) is 🔨 in progress: +> P1 + P2 merged (PR #33)**, and **P3 + P4/D13 have landed on `development` in a follow-on PR (NOT yet +> merged)** — its three design ADRs (0042/0043/0044) are Accepted. +> **Merged (P1+P2, PR #33):** the `MediaStore` contract impls + `media_objects`/`media_references` tables +> (migration 0002), per-modality capability gating + `FallbackChain` skip, `deInlineMedia`, the strict +> `output_modalities`/`save_to` node fields, OpenAI `mediaUnits`, and **the I3 keystone — `deInlineMedia` +> at the one `#emitDurable` choke point**. +> **Landed on `development` (pending merge):** the binary media-egress + **SSRF mechanism half** (D9 — the +> `MediaUrlFetch` re-host hook + `fetchMediaBytes` SSRF-validated host reference: DNS-resolve + +> connect-by-validated-IP + per-hop redirect re-validation + streamed size-bound, wired at the choke point); +> the `FallbackChain` resolve-before-egress + **byte-free re-materialization sidecar** (D8 + D7); the +> **byte-delivery `Range` gate** (D13 — `MediaStore.readRange` + the engine-pure `validateByteRange`); and the +> dedicated **P3 egress/SSRF security-review** (independent adversarial — 0 blockers/highs). All green + +> Leakwatch-clean. +> **Remaining (P4):** `read_media` + the scope-set authz (D12), the terminal sweep + GC (D11), the +> `save_to` write port (D16), the `output_modalities` load-check (D15), and the per-modality media cost +> governor (D17) — D12/D13 share a dedicated byte-delivery security-review pass + the keychain no-raw-key +> IPC test; plus the still-pending canonical-home docs (built-in-tools, config-spec, security-review). The +> remaining Phase-1 work is **additive and off the critical path**: only **1.AF–1.AH** (engine media plumbing, +> output, surfaces) remain before Phase 1 closes. **Phase 2 (CLI, milestone M3) is unblocked.** Carry-over hardening is tracked in [deferred-tasks.md](deferred-tasks.md) — pick items up as Phase 1 first touches each file. diff --git a/docs/roadmap/phases/phase-1-engine-and-llm.md b/docs/roadmap/phases/phase-1-engine-and-llm.md index 6b5b13bb..3bb451bf 100644 --- a/docs/roadmap/phases/phase-1-engine-and-llm.md +++ b/docs/roadmap/phases/phase-1-engine-and-llm.md @@ -972,8 +972,9 @@ phases (2–6). Each phase below maps to the design doc's Phase A–E. **never returns a raw key from an IPC command** (direct test); and a run reaching a **terminal event (`run:completed|failed|cancelled`) deterministically reclaims its media refs** so a FAILED/CANCELLED run leaves no orphaned partial media (a terminal-state sweep, not refcount-GC alone). - - 🔨 *In progress on `development` (NOT merged — see [ADR-0042](../../decisions/0042-engine-media-storage-substrate-mediastore-deinline-retention.md)/[0043](../../decisions/0043-media-egress-failover-rematerialization-ssrf.md)/[0044](../../decisions/0044-media-access-governance-read-media-save-to-cost.md), the three accepted 1.AF design ADRs).* - **Landed (P1 + P2):** D5/D6 per-modality `requiredCapabilities()` + `FallbackChain` provider-skip + - 🔨 *P1+P2 **merged** (PR #33); P3 + P4/D13 landed on `development` in a follow-on PR (**pending merge**) — + see [ADR-0042](../../decisions/0042-engine-media-storage-substrate-mediastore-deinline-retention.md)/[0043](../../decisions/0043-media-egress-failover-rematerialization-ssrf.md)/[0044](../../decisions/0044-media-access-governance-read-media-save-to-cost.md), the three accepted 1.AF design ADRs.* + **Merged (P1 + P2, PR #33):** D5/D6 per-modality `requiredCapabilities()` + `FallbackChain` provider-skip (one shared `mediaSupportReason` predicate); D3 `deInlineMedia` (cycle-safe flight→durable transform + a pure base64 decoder); D1 `MediaStore` contract impls (`FilesystemMediaStore` CAS + `InMemoryMediaStore`); D10 the `media_objects` + `media_references` tables + migration 0002; D14 strict `output_modalities`/`save_to` @@ -982,16 +983,28 @@ phases (2–6). Each phase below maps to the design doc's Phase A–E. persist-before-deliver preserved; missing-store → loud `run:failed`, terminal stays emit-safe); plus the canonical-home doc updates that landed with the code — database-schema.md §Media tables, workflow-yaml-spec (`output_modalities` / `save_to` / `{{ run.id }}`), and sse-event-schema (`mediaUnits` on `cost:updated`). - All green (`pnpm turbo` 16/16) + Leakwatch-clean. - **Remaining (P3 + P4):** D8 adapter handle/url resolution, D7 the B5 sidecar re-materialize, **D9 the - binary media-egress capability + the SSRF mechanism half** (security-critical — needs a dedicated - security-review pass per ADR-0043); **D12/D13 `read_media` + the byte-delivery gate** (security-critical - — ADR-0044/security-review.md), D11 the terminal-state sweep, D15 the `output_modalities` load-check, - D16 the `save_to` write port, D17 the per-modality media cost governor, the keychain-bridge IPC test; - plus the **still-pending** canonical-home doc updates: config-spec (the media GC grace key — P4/D11) and - security-review (the SSRF/egress controls — P3/D9). _(database-schema, workflow-yaml-spec, and - sse-event-schema already landed with the P1+P2 code — see above.)_ The matrix row stays ◇ until the 1.AF - PR merges (Done-after-merge). + **Landed on `development` (P3 + P4/D13, pending merge):** + - **D9** the binary media-egress capability + the SSRF **mechanism** half: `deInlineMedia` re-hosts a `url` + source via an injected `MediaUrlFetch` hook (shared); `fetchMediaBytes` SSRF-validated host reference + (db — DNS-resolve + connect-by-validated-IP + per-hop redirect re-validation + streamed size-bound, + TLS never disabled, secret-free errors); wired at the `#emitDurable` choke point via the + `ExecutionHost.fetchMedia` port (a missing port ⇒ a `url` hard-fails, I3). + - **D8 + D7** the `FallbackChain` resolves a `handle` media source to the provider's in-flight source + **before egress** (the injected `resolveForEgress` hook — adapters stay pure) + the **byte-free** + `(provider, handle)` re-materialization sidecar (a non-base64 ref cached + reused; base64 never cached; + cross-provider re-materialize; retryable-advance on a re-materialization failure). + - **D13** the byte-delivery `Range` gate: `MediaStore.readRange` (host mechanism — path-jail + sha256 + + a defensive re-bound) + the engine-pure `validateByteRange` policy (fail-closed; reused by D12 + 1.AH). + - The dedicated **P3 egress/SSRF security-review pass** (independent adversarial — **0 blockers/highs**; + 4 lower follow-ups fixed). All green (`pnpm turbo` 16/16) + Leakwatch-clean. + **Remaining (P4):** **D12 `read_media`** (the 13th tool + the scope-set authz + the additive + `ToolPolicyDenyReason.media_scope_denied` + the `ToolDispatchContext` `byteLength`/session-scope fields + + the `media_references` row-writing lifecycle); D16 the `save_to` write port; D15 the `output_modalities` + load-check; D17 the per-modality media cost governor; D11 the terminal-state sweep + grace-window GC; the + keychain-bridge no-raw-key IPC test; the **P4 byte-delivery security-review pass**; plus the still-pending + canonical-home doc updates: built-in-tools (`read_media`), config-spec (the media GC grace key + the media + cost estimate), security-review (the byte-delivery + host SSRF mechanism sections). The matrix row stays ◇ + until the 1.AF PR(s) merge (Done-after-merge). - **1.AG — Output generation (Phase D).** Inline media-out (Gemini `responseModalities`, OpenAI agentic image-gen via the `providerExecuted`+normalized-`media` arm, OpenAI inline audio); the `generateMedia`/`pollMediaJob` separate-endpoint generators (gpt-image-1, Imagen, TTS sync; Sora/Veo diff --git a/packages/core/src/engine/agent-runner.ts b/packages/core/src/engine/agent-runner.ts index 2f34f1fe..4af543ef 100644 --- a/packages/core/src/engine/agent-runner.ts +++ b/packages/core/src/engine/agent-runner.ts @@ -66,6 +66,12 @@ export interface AgentRunnerDeps { readonly now?: ChainCapabilities['now']; /** Optional single out-of-band credential refresh (host-owned). */ readonly onAuthError?: ChainCapabilities['onAuthError']; + /** + * Host media-egress resolver (1.AF/D8) — turns a durable `handle` media source into the in-flight source + * a provider needs, before egress (backed by `MediaStore.resolveForEgress`). Forwarded into the chain so + * the adapter only ever sees a resolved source; absent on a text-only host (a handle is then sent as-is). + */ + readonly resolveForEgress?: ChainCapabilities['resolveForEgress']; /** Host capability for the `read_file` interpolation filter in a prompt (delegated workspace sandbox). */ readonly resolverCapabilities?: ResolverCapabilities; /** The filesystem scope tier for tool dispatch (default `'sandboxed'` — the safe tier). */ @@ -329,6 +335,7 @@ function chainCapabilities(deps: AgentRunnerDeps): ChainCapabilities { sleep: deps.sleep, ...(deps.now === undefined ? {} : { now: deps.now }), ...(deps.onAuthError === undefined ? {} : { onAuthError: deps.onAuthError }), + ...(deps.resolveForEgress === undefined ? {} : { resolveForEgress: deps.resolveForEgress }), }; } diff --git a/packages/core/src/engine/agent-turn.ts b/packages/core/src/engine/agent-turn.ts index 8066a8f4..c307cbac 100644 --- a/packages/core/src/engine/agent-turn.ts +++ b/packages/core/src/engine/agent-turn.ts @@ -85,7 +85,7 @@ export type PreEgressHook = (info: { /** The chain capabilities the host supplies (the platform-level subset of {@link FallbackChainOptions}). */ export type ChainCapabilities = Pick< FallbackChainOptions, - 'keyFor' | 'sleep' | 'now' | 'onAuthError' + 'keyFor' | 'sleep' | 'now' | 'onAuthError' | 'resolveForEgress' >; /** Everything one agent turn needs — no run/session correlation key, no `NodeExecContext`. */ diff --git a/packages/core/src/engine/engine.test.ts b/packages/core/src/engine/engine.test.ts index d173e3c6..1f5aa25d 100644 --- a/packages/core/src/engine/engine.test.ts +++ b/packages/core/src/engine/engine.test.ts @@ -187,6 +187,7 @@ function stubMediaStore(): { store: MediaStore; puts: { handle: string; bytes: U : Promise.resolve(found.bytes); }, resolveForEgress: () => Promise.reject(new Error('unused by this test')), + readRange: () => Promise.reject(new Error('unused by this test')), }; return { store, puts }; } @@ -373,6 +374,7 @@ describe('WorkflowEngine — media de-inline at the emit choke point (1.AF, ADR- put: () => Promise.reject(new Error('disk full')), get: () => Promise.reject(new Error('unused')), resolveForEgress: () => Promise.reject(new Error('unused')), + readRange: () => Promise.reject(new Error('unused')), }; const runStore = new InMemoryRunStore(); const host = createInMemoryHost({ store: runStore, mediaStore: rejectingStore }); @@ -385,6 +387,102 @@ describe('WorkflowEngine — media de-inline at the emit choke point (1.AF, ADR- expect(terminalsIn(events)[0]?.type).toBe('run:failed'); expect(JSON.stringify(events)).not.toContain('aGVsbG8='); }); + + it('re-hosts a url media node output to a handle via the host media-egress port (D9, no url persisted)', async () => { + const { store: mediaStore, puts } = stubMediaStore(); + const runStore = new InMemoryRunStore(); + const fetched: string[] = []; + const FETCH_BYTES = new Uint8Array([5, 6, 7]); + const host = createInMemoryHost({ + store: runStore, + mediaStore, + fetchMedia: (url) => { + fetched.push(url); + return Promise.resolve(FETCH_BYTES); + }, + }); + const urlPart = { + type: 'media' as const, + mimeType: 'image/png', + source: { kind: 'url' as const, url: 'https://media.example/a.png' }, + }; + const events = await drain( + engineWith({ work: () => ({ kind: 'completed', output: { image: urlPart } }) }, host).start({ + workflow: workflow(SEQUENTIAL), + }), + ); + expect(fetched).toEqual(['https://media.example/a.png']); // the host egress port was invoked + const put0 = puts[0]; + expect(put0).toBeDefined(); + const done = events.find((e) => e.type === 'node:completed' && e.nodeId === 'work'); + const output = done?.type === 'node:completed' ? done.output : undefined; + expect(output).toEqual({ + image: { + type: 'media', + mimeType: 'image/png', + source: { kind: 'handle', ref: put0?.handle }, + byteLength: 3, + }, + }); + // I3 — the url never reached the delivered stream or the persisted log (re-hosted to a handle). + expect(JSON.stringify(events)).not.toContain('media.example'); + const runId = events[0]?.runId; + if (runId !== undefined) { + expect(JSON.stringify(runStore.eventsFor(runId))).not.toContain('media.example'); + } + expect(terminalsIn(events)[0]?.type).toBe('run:completed'); + }); + + it('hard-fails a url media output when the host has a store but NO media-egress port (no leak)', async () => { + const { store: mediaStore, puts } = stubMediaStore(); + const runStore = new InMemoryRunStore(); + const host = createInMemoryHost({ store: runStore, mediaStore }); // store, but no fetchMedia port + const urlPart = { + type: 'media' as const, + mimeType: 'image/png', + source: { kind: 'url' as const, url: 'https://media.example/a.png' }, + }; + const events = await drain( + engineWith({ work: () => ({ kind: 'completed', output: urlPart }) }, host).start({ + workflow: workflow(SEQUENTIAL), + }), + ); + expect(terminalsIn(events)).toHaveLength(1); + expect(terminalsIn(events)[0]?.type).toBe('run:failed'); + expect(puts).toHaveLength(0); // an un-re-hostable url is fail-closed — nothing stored + expect(JSON.stringify(events)).not.toContain('media.example'); + }); + + it('fails the run (no leak) when the media-egress port THROWS on a url output (D9 fetch failure)', async () => { + // The third D9 branch: a fetchMedia hook IS wired but rejects (an SSRF block / network error / size + // overrun). The rejection propagates through deInlineMedia to #emitDurable's catch → one run:failed; the + // url + the failure reason stay out of every delivered + persisted event (secret-free, I3). + const { store: mediaStore, puts } = stubMediaStore(); + const runStore = new InMemoryRunStore(); + const host = createInMemoryHost({ + store: runStore, + mediaStore, + fetchMedia: () => Promise.reject(new Error('blocked_host')), + }); + const urlPart = { + type: 'media' as const, + mimeType: 'image/png', + source: { kind: 'url' as const, url: 'https://media.example/a.png' }, + }; + const events = await drain( + engineWith({ work: () => ({ kind: 'completed', output: urlPart }) }, host).start({ + workflow: workflow(SEQUENTIAL), + }), + ); + expect(terminalsIn(events)).toHaveLength(1); + expect(terminalsIn(events)[0]?.type).toBe('run:failed'); + expect(puts).toHaveLength(0); // the fetch failed before any put + expect(JSON.stringify(events)).not.toContain('media.example'); // url never persisted/delivered + const runId = events[0]?.runId; + if (runId !== undefined) { + expect(JSON.stringify(runStore.eventsFor(runId))).not.toContain('media.example'); + } + }); }); // --- cancellation ----------------------------------------------------------------------------- diff --git a/packages/core/src/engine/engine.ts b/packages/core/src/engine/engine.ts index d6174699..30bcdb7b 100644 --- a/packages/core/src/engine/engine.ts +++ b/packages/core/src/engine/engine.ts @@ -29,6 +29,7 @@ */ import { + DEFAULT_MAX_MEDIA_DOWNLOAD_BYTES, GateDecisionSchema, RETRYABLE_ERROR_CODES, RunEventSchema, @@ -37,6 +38,7 @@ import { type ExecutionMode, type GateDecision, type MaskedSecret, + type MediaUrlFetch, type NodeSkippedReason, type Retry, type RunEvent, @@ -1500,7 +1502,9 @@ class RunExecution { async #deInlineDraft(draft: RunEventDraft): Promise { const store = this.#host.mediaStore; if (store !== undefined) { - return (await deInlineMedia(draft, store)) as RunEventDraft; + // Pass the host media-egress hook (D9) so a `url` media source is re-hosted to a handle; undefined + // when the host has no egress mechanism, in which case a `url` hard-fails inside deInlineMedia (I3). + return (await deInlineMedia(draft, store, this.#mediaUrlFetch())) as RunEventDraft; } // No store: a draft carrying inline bytes OR an un-re-hosted url media part cannot be made // durable-safe — throw (the broadened #emitDurable catch + the #onOutcome/#begin backstops map it to @@ -1515,6 +1519,21 @@ class RunExecution { return draft; } + /** + * Build the `deInlineMedia` url-rehost hook (1.AF/D9) from the host media-egress port, bound to this + * run's size-bound **policy** ({@link DEFAULT_MAX_MEDIA_DOWNLOAD_BYTES}) and abort signal — so the host + * mechanism receives the engine-supplied bound. `undefined` when the host has no egress mechanism, in + * which case a `url` media source hard-fails inside `deInlineMedia` (an un-re-hosted url may never + * persist, I3). + */ + #mediaUrlFetch(): MediaUrlFetch | undefined { + const fetchMedia = this.#host.fetchMedia; + if (fetchMedia === undefined) { + return undefined; + } + return (url) => fetchMedia(url, DEFAULT_MAX_MEDIA_DOWNLOAD_BYTES, this.#abort.signal); + } + #elapsedMs(): number { return Date.parse(this.#host.clock.now()) - this.#startEpochMs; } diff --git a/packages/core/src/engine/execution-host.ts b/packages/core/src/engine/execution-host.ts index ab1c714e..bde81457 100644 --- a/packages/core/src/engine/execution-host.ts +++ b/packages/core/src/engine/execution-host.ts @@ -154,8 +154,25 @@ export interface ExecutionHost { * NO store injected is a loud configuration breach (`media_store_unavailable`), never a silent byte leak. */ readonly mediaStore?: MediaStore; + /** + * The host media-egress mechanism (1.AF/D9, [ADR-0043](../../../../docs/decisions/0043-media-egress-failover-rematerialization-ssrf.md)) + * — fetch the bytes at a public-HTTPS `url` with the SSRF-validated, size-bounded connect the engine + * binds into `deInlineMedia`'s url-rehost hook. The engine supplies the `maxBytes` **policy** + the run + * `AbortSignal`; the host performs the validated I/O (DNS-resolve + connect-by-validated-IP + per-hop + * redirect re-validation). **Optional + absent-tolerant**: with no egress mechanism a `url` media source + * hard-fails at the de-inline choke point (an un-re-hosted url may never persist, I3). The Node reference + * is `@relavium/db`'s `fetchMediaBytes`. + */ + readonly fetchMedia?: HostMediaFetch; } +/** The host media-egress port: a public-HTTPS `url` → its bytes, under an engine-supplied size bound. */ +export type HostMediaFetch = ( + url: string, + maxBytes: number, + signal?: AbortSignalLike, +) => Promise; + // --- In-memory reference implementation (engine tests + the local reference) ------------------- const TERMINAL_TYPES: ReadonlySet = new Set([ @@ -294,6 +311,8 @@ export function createInMemoryHost(options?: { baseEpochMs?: number; /** Inject a media store so a run that produces media de-inlines it (1.AF); omit for a text-only host. */ mediaStore?: MediaStore; + /** Inject a media-egress fetch so a `url` media source is re-hosted (1.AF/D9); omit to hard-fail urls. */ + fetchMedia?: HostMediaFetch; }): ExecutionHost & { store: RunStore } & Pick { let tick = options?.baseEpochMs ?? Date.parse('2026-01-01T00:00:00.000Z'); let idCounter = 0; @@ -307,6 +326,7 @@ export function createInMemoryHost(options?: { newAbortController: createAbortController, setTimer: timers.setTimer, ...(options?.mediaStore ? { mediaStore: options.mediaStore } : {}), + ...(options?.fetchMedia ? { fetchMedia: options.fetchMedia } : {}), fireTimers: timers.fireTimers, armedCount: timers.armedCount, }; diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 45d924df..aa70d48e 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -72,3 +72,17 @@ export { // handle. Node-side (node:crypto + node:fs); a host wires one into ExecutionHost.mediaStore. The pure // engine never imports this — it depends only on the @relavium/shared `MediaStore` interface. export { FilesystemMediaStore, InMemoryMediaStore } from './media-store.js'; + +// Media egress (1.AF/D9, ADR-0043) — the host-side SSRF-validated, size-bounded URL fetch the engine +// binds into a `MediaUrlFetch` hook so `deInlineMedia` can re-host a url media source to a handle. +// Node-side (node:dns + node:https + node:net). The pure engine never imports this. +export { + fetchMediaBytes, + nodeMediaEgressDeps, + MediaEgressError, + type FetchMediaBytesOptions, + type MediaEgressDeps, + type MediaEgressErrorCode, + type HopRequest, + type HopResponse, +} from './media-egress.js'; diff --git a/packages/db/src/media-egress.test.ts b/packages/db/src/media-egress.test.ts new file mode 100644 index 00000000..3d08795d --- /dev/null +++ b/packages/db/src/media-egress.test.ts @@ -0,0 +1,278 @@ +import { describe, expect, it } from 'vitest'; + +import { + fetchMediaBytes, + MediaEgressError, + type HopRequest, + type MediaEgressDeps, +} from './media-egress.js'; + +const PUBLIC_IP = '203.0.113.10'; // TEST-NET-3 documentation range — not in any SSRF private block +const PUBLIC_IP_2 = '198.51.100.7'; // TEST-NET-2 — a second public address for redirect targets + +/** A scripted hop the fake `openConnection` returns in order. */ +interface ScriptedHop { + readonly status: number; + readonly location?: string; + readonly body?: readonly Uint8Array[]; + /** Emit one chunk then throw a raw Node-style AbortError mid-stream (models a socket destroy on abort). */ + readonly bodyThrows?: boolean; +} + +function bodyOf(chunks: readonly Uint8Array[]): AsyncIterable { + return (async function* gen(): AsyncGenerator { + await Promise.resolve(); // a real body stream is async; satisfy require-await for the fake + for (const chunk of chunks) { + yield chunk; + } + })(); +} + +/** A body stream that yields one chunk then throws a raw Node AbortError (a socket destroy mid-read). */ +function throwingBody(): AsyncIterable { + return (async function* gen(): AsyncGenerator { + await Promise.resolve(); + yield new Uint8Array([1]); + const err = new Error('The operation was aborted'); + err.name = 'AbortError'; + throw err; + })(); +} + +/** Build fake deps + capture the connection calls and dispose count, so SSRF policy is deterministic. */ +function fakeDeps(config: { + readonly resolve?: Record; + readonly hops?: readonly ScriptedHop[]; +}): { + deps: MediaEgressDeps; + calls: HopRequest[]; + stats: { disposed: number }; +} { + const calls: HopRequest[] = []; + const stats = { disposed: 0 }; + let hopIndex = 0; + const deps: MediaEgressDeps = { + resolveHost: (hostname) => Promise.resolve(config.resolve?.[hostname] ?? [hostname]), + openConnection: (request) => { + calls.push(request); + const hop = config.hops?.[hopIndex]; + hopIndex += 1; + if (hop === undefined) { + return Promise.reject(new Error('test: no scripted hop')); + } + return Promise.resolve({ + status: hop.status, + location: hop.location, + body: hop.bodyThrows === true ? throwingBody() : bodyOf(hop.body ?? []), + dispose: () => { + stats.disposed += 1; + }, + }); + }, + }; + return { deps, calls, stats }; +} + +describe('fetchMediaBytes (1.AF/D9, ADR-0043 — SSRF-validated, size-bounded media egress)', () => { + it('fetches bytes over a 200, pinning the connection to the validated resolved IP', async () => { + const { deps, calls } = fakeDeps({ + resolve: { 'media.example': [PUBLIC_IP] }, + hops: [{ status: 200, body: [new Uint8Array([1, 2, 3, 4])] }], + }); + const bytes = await fetchMediaBytes('https://media.example/a.png', { maxBytes: 1000 }, deps); + expect([...bytes]).toEqual([1, 2, 3, 4]); + expect(calls).toHaveLength(1); + expect(calls[0]?.hostname).toBe('media.example'); // SNI/Host stays the hostname + expect(calls[0]?.pinnedIp).toBe(PUBLIC_IP); // connected by the validated IP (TOCTOU defense) + }); + + it('rejects a non-HTTPS url (insecure_url), opening no connection', async () => { + const { deps, calls } = fakeDeps({ hops: [{ status: 200 }] }); + await expect( + fetchMediaBytes('http://media.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'insecure_url' }); + expect(calls).toHaveLength(0); + }); + + it('rejects a url with embedded credentials (insecure_url)', async () => { + const { deps, calls } = fakeDeps({ hops: [{ status: 200 }] }); + await expect( + fetchMediaBytes('https://user:pass@media.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'insecure_url' }); + expect(calls).toHaveLength(0); + }); + + it('blocks a private host literal (blocked_host), never resolving or connecting', async () => { + const { deps, calls } = fakeDeps({ hops: [{ status: 200 }] }); + await expect( + fetchMediaBytes('https://127.0.0.1/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'blocked_host' }); + expect(calls).toHaveLength(0); + }); + + it('blocks a public hostname that RESOLVES to a private IP (blocked_host), opening no connection', async () => { + const { deps, calls } = fakeDeps({ + resolve: { 'rebind.example': ['10.0.0.1'] }, + hops: [{ status: 200 }], + }); + await expect( + fetchMediaBytes('https://rebind.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'blocked_host' }); + expect(calls).toHaveLength(0); + }); + + it('blocks when ANY of multiple resolved IPs is private (fail-closed over the whole record set)', async () => { + const { deps } = fakeDeps({ + resolve: { 'mixed.example': [PUBLIC_IP, '169.254.169.254'] }, // one public, one metadata + hops: [{ status: 200 }], + }); + await expect( + fetchMediaBytes('https://mixed.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'blocked_host' }); + }); + + it('allows a private target only under the explicit allowPrivate opt-in', async () => { + const { deps, calls } = fakeDeps({ + resolve: { 'localhost.example': ['127.0.0.1'] }, + hops: [{ status: 200, body: [new Uint8Array([9])] }], + }); + const bytes = await fetchMediaBytes( + 'https://localhost.example/a.png', + { maxBytes: 1000, allowPrivate: true }, + deps, + ); + expect([...bytes]).toEqual([9]); + expect(calls).toHaveLength(1); + }); + + it('follows a redirect to a public host (re-validating the new target) and disposes the redirect body', async () => { + const { deps, calls, stats } = fakeDeps({ + resolve: { 'a.example': [PUBLIC_IP], 'b.example': [PUBLIC_IP_2] }, + hops: [ + { status: 302, location: 'https://b.example/final.png' }, + { status: 200, body: [new Uint8Array([7, 7])] }, + ], + }); + const bytes = await fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps); + expect([...bytes]).toEqual([7, 7]); + expect(calls.map((c) => c.hostname)).toEqual(['a.example', 'b.example']); + expect(calls[1]?.pinnedIp).toBe(PUBLIC_IP_2); // the redirect target was independently resolved + pinned + expect(stats.disposed).toBeGreaterThanOrEqual(1); // the 302 body was disposed, not read + }); + + it('blocks a redirect to a private host (per-hop re-validation), even after a public first hop', async () => { + const { deps } = fakeDeps({ + resolve: { 'a.example': [PUBLIC_IP], 'evil.example': ['192.168.1.5'] }, + hops: [{ status: 307, location: 'https://evil.example/x' }, { status: 200 }], + }); + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'blocked_host' }); + }); + + it('blocks a redirect to a non-HTTPS target (insecure_url on the hop)', async () => { + const { deps } = fakeDeps({ + resolve: { 'a.example': [PUBLIC_IP] }, + hops: [{ status: 301, location: 'http://a.example/x' }], + }); + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'insecure_url' }); + }); + + it('fails with too_many_redirects past the limit', async () => { + const { deps } = fakeDeps({ + resolve: { 'a.example': [PUBLIC_IP] }, + hops: [ + { status: 302, location: 'https://a.example/1' }, + { status: 302, location: 'https://a.example/2' }, + { status: 302, location: 'https://a.example/3' }, + ], + }); + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000, maxRedirects: 1 }, deps), + ).rejects.toMatchObject({ code: 'too_many_redirects' }); + }); + + it('aborts an over-size body (too_large) and disposes the stream', async () => { + const { deps, stats } = fakeDeps({ + resolve: { 'big.example': [PUBLIC_IP] }, + hops: [{ status: 200, body: [new Uint8Array(8), new Uint8Array(8)] }], // 16 bytes + }); + await expect( + fetchMediaBytes('https://big.example/a.png', { maxBytes: 10 }, deps), + ).rejects.toMatchObject({ code: 'too_large' }); + expect(stats.disposed).toBeGreaterThanOrEqual(1); + }); + + it('fails with bad_status on a non-200, non-redirect response', async () => { + const { deps } = fakeDeps({ + resolve: { 'a.example': [PUBLIC_IP] }, + hops: [{ status: 404 }], + }); + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'bad_status' }); + }); + + it('normalizes a socket abort/destroy mid-body-read to a typed MediaEgressError (network)', async () => { + // The response promise has already resolved when the body stream throws; without normalization a raw + // Node AbortError would escape, breaking the "only MediaEgressError" contract (security-review finding). + const { deps, stats } = fakeDeps({ + resolve: { 'a.example': [PUBLIC_IP] }, + hops: [{ status: 200, bodyThrows: true }], + }); + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'network' }); + expect(stats.disposed).toBeGreaterThanOrEqual(1); // the stream was disposed on the way out + }); + + it('normalizes a resolver rejection to a typed MediaEgressError (never a raw DNS error)', async () => { + const deps: MediaEgressDeps = { + resolveHost: () => Promise.reject(new Error('getaddrinfo ENOTFOUND a.example')), + openConnection: () => Promise.reject(new Error('test: must not connect')), + }; + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'network' }); + }); + + it('blocks a resolver that returns a non-IP token (the pinned target must be an IP literal)', async () => { + const deps: MediaEgressDeps = { + resolveHost: () => Promise.resolve(['not-an-ip.example']), // a hostname, never a pinnable IP + openConnection: () => Promise.reject(new Error('test: must not connect')), + }; + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'blocked_host' }); + }); + + it('blocks an empty DNS resolution (no address to pin), opening no connection', async () => { + const { deps, calls } = fakeDeps({ + resolve: { 'nxdomain.example': [] }, // resolver returns zero addresses + hops: [{ status: 200 }], + }); + await expect( + fetchMediaBytes('https://nxdomain.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'blocked_host' }); + expect(calls).toHaveLength(0); // fail-closed before any openConnection (never pins to the hostname) + }); + + it('normalizes a malformed redirect Location to a typed MediaEgressError (no raw URL TypeError)', async () => { + const { deps } = fakeDeps({ + resolve: { 'a.example': [PUBLIC_IP] }, + hops: [{ status: 302, location: 'https://[' }], // unclosed IPv6 — new URL() throws (scheme incidental) + }); + await expect( + fetchMediaBytes('https://a.example/a.png', { maxBytes: 1000 }, deps), + ).rejects.toMatchObject({ code: 'network' }); + }); + + it('exposes a typed MediaEgressError', () => { + const err = new MediaEgressError('blocked_host', 'x'); + expect(err).toBeInstanceOf(Error); + expect(err.code).toBe('blocked_host'); + expect(err.name).toBe('MediaEgressError'); + }); +}); diff --git a/packages/db/src/media-egress.ts b/packages/db/src/media-egress.ts new file mode 100644 index 00000000..e232a936 --- /dev/null +++ b/packages/db/src/media-egress.ts @@ -0,0 +1,332 @@ +import { lookup as dnsLookup } from 'node:dns/promises'; +import { request as httpsRequest } from 'node:https'; +import { isIP } from 'node:net'; + +import { + extractHttpsHost, + isPrivateOrLocalHost, + urlHasCredentials, + type AbortSignalLike, +} from '@relavium/shared'; + +/** + * `fetchMediaBytes` (1.AF/D9, [ADR-0043](../../../docs/decisions/0043-media-egress-failover-rematerialization-ssrf.md) + * §2/§3) — the host **mechanism** half of media egress: the Node/filesystem-host reference + * implementation the engine binds into a {@link MediaUrlFetch} hook so `deInlineMedia` can re-host a + * `url` media source to bytes. The engine owns the **policy** (the one shared SSRF primitive + the size + * bound); this performs the validated I/O, in one place, never an adapter: + * + * - **HTTPS-only, no embedded credentials** — rejected via the shared {@link extractHttpsHost} / + * {@link urlHasCredentials} primitives (never a second hand-rolled parser). + * - **DNS-rebind defense (TOCTOU)** — resolve the hostname, validate **every** resolved IP against the + * shared {@link isPrivateOrLocalHost} range-block, then **connect by the validated IP** (a pinned + * `lookup`) so the address checked is the address connected to. + * - **Per-hop redirect re-validation** — every `3xx` `Location` re-runs the whole HTTPS + no-creds + + * resolve + range-block + pin cycle on the new target (a redirect-to-private / -to-http is blocked + * mid-fetch), bounded by `maxRedirects`. A redirect body is never read. + * - **Streamed, size-bounded** — the body is consumed chunk-by-chunk and aborted the moment it exceeds + * `maxBytes`; an over-size response is never fully buffered. + * - **TLS verification is never disabled** — the request connects to the pinned IP but keeps the + * original hostname as the SNI `servername`, so the certificate is validated against the hostname. + * + * Errors are a typed {@link MediaEgressError} whose message names a **reason only** — never the url, the + * resolved IP, a host stack, or bytes (ADR-0043 §4 secret-free discipline). The DNS resolver and the + * connection opener are injectable ({@link MediaEgressDeps}) so the SSRF policy + redirect + size-bound + * orchestration is deterministically unit-testable without real network/DNS; the default deps are Node. + */ + +/** Why a media egress fetch failed — a secret-free, reason-only discriminant. */ +export type MediaEgressErrorCode = + | 'insecure_url' // not HTTPS, embeds credentials, or a malformed authority + | 'blocked_host' // resolves to (or is) a private/loopback/link-local/metadata address + | 'too_many_redirects' + | 'too_large' // body exceeded the configured maximum download size + | 'bad_status' // a non-200, non-redirect HTTP status + | 'network'; // the connection failed / was aborted + +/** A typed media-egress failure. The `message` names a reason only — never the url/IP/bytes (rule 6). */ +export class MediaEgressError extends Error { + readonly code: MediaEgressErrorCode; + constructor(code: MediaEgressErrorCode, message: string) { + super(message); + this.name = 'MediaEgressError'; + this.code = code; + } +} + +export interface FetchMediaBytesOptions { + /** The per-fetch upper bound on the streamed body in bytes (the engine supplies this policy). */ + readonly maxBytes: number; + /** Overall request timeout in ms (default 30000). */ + readonly timeoutMs?: number; + /** Maximum number of redirects followed before failing (default 5). */ + readonly maxRedirects?: number; + /** + * Cancels the fetch (composed with the timeout). Typed as the platform-free `AbortSignalLike` so the + * engine's `AbortControllerLike.signal` (the run abort) wires in without a cast at the `HostMediaFetch` + * boundary; a real `AbortSignal` structurally satisfies it. + */ + readonly signal?: AbortSignalLike; + /** + * Allow a private/loopback target — the BYOK explicit local-endpoint opt-in (security-review.md). + * Default `false`: private/loopback/link-local/metadata addresses are blocked. + */ + readonly allowPrivate?: boolean; +} + +/** One redirect-free HTTP response the orchestrator inspects (status + Location + a body stream). */ +export interface HopResponse { + readonly status: number; + readonly location: string | undefined; + readonly body: AsyncIterable; + /** Abort the underlying socket — called when we stop reading early (a redirect, an error, an over-size body). */ + readonly dispose: () => void; +} + +/** One pinned request the connection opener must perform (no redirect following — the orchestrator owns that). */ +export interface HopRequest { + readonly url: string; + readonly hostname: string; + /** The pre-validated IP the connection MUST be pinned to (TOCTOU defense — never re-resolve here). */ + readonly pinnedIp: string; +} + +/** Injectable I/O primitives — Node by default; faked in tests so the SSRF policy is deterministic. */ +export interface MediaEgressDeps { + /** Resolve a hostname to its IP(s) (an IP literal resolves to itself). */ + readonly resolveHost: (hostname: string) => Promise; + /** Open ONE pinned HTTPS connection and return its (unread) response. */ + readonly openConnection: (request: HopRequest, signal: AbortSignal) => Promise; +} + +const DEFAULT_TIMEOUT_MS = 30_000; +const DEFAULT_MAX_REDIRECTS = 5; + +/** True for the redirect statuses we follow (a `Location` is required, re-validated per hop). */ +function isRedirectStatus(status: number): boolean { + return status === 301 || status === 302 || status === 303 || status === 307 || status === 308; +} + +/** + * Validate an egress URL's scheme + authority via the shared SSRF policy primitive and return its + * lowercased host. Throws `insecure_url` for a non-HTTPS scheme, a malformed authority, or embedded + * credentials — never a second hand-rolled parser. + */ +function validateEgressHost(url: string): string { + if (urlHasCredentials(url)) { + throw new MediaEgressError('insecure_url', 'media egress url must not embed credentials'); + } + const parsed = extractHttpsHost(url); + if (parsed === null) { + throw new MediaEgressError('insecure_url', 'media egress url must be a well-formed https url'); + } + if (parsed.hasCredentials) { + throw new MediaEgressError('insecure_url', 'media egress url must not embed credentials'); + } + return parsed.host; +} + +/** + * Resolve `host` and validate the host literal AND **every** resolved IP against the shared range-block. + * Fail-closed: any private/loopback/link-local/metadata address (unless `allowPrivate`) blocks the whole + * fetch — so a multi-record name with one private answer cannot slip through. + */ +async function resolveValidatedIps( + host: string, + deps: MediaEgressDeps, + allowPrivate: boolean, +): Promise { + if (!allowPrivate && isPrivateOrLocalHost(host)) { + throw new MediaEgressError('blocked_host', 'media egress target is a private/loopback address'); + } + const ips = await deps.resolveHost(host); + if (ips.length === 0) { + throw new MediaEgressError('blocked_host', 'media egress target did not resolve to an address'); + } + for (const ip of ips) { + // Every resolved value MUST be an IP literal — otherwise a (buggy/malicious) resolver returning a + // hostname would pass the range-block (a hostname is not a private IP) and become the pinned `lookup` + // target, defeating the connect-by-validated-IP guarantee. Fail-closed on a non-IP. + if (isIP(ip) === 0) { + throw new MediaEgressError('blocked_host', 'media egress resolver returned a non-IP address'); + } + if (!allowPrivate && isPrivateOrLocalHost(ip)) { + throw new MediaEgressError( + 'blocked_host', + 'media egress target resolves to a private/loopback address', + ); + } + } + return ips; +} + +/** Consume a body stream, aborting the moment it exceeds `maxBytes`; concat the bounded chunks. */ +async function readBounded( + body: AsyncIterable, + maxBytes: number, + dispose: () => void, +): Promise { + const chunks: Uint8Array[] = []; + let total = 0; + try { + for await (const chunk of body) { + total += chunk.length; + if (total > maxBytes) { + throw new MediaEgressError('too_large', 'media egress response exceeded the maximum size'); + } + chunks.push(chunk); + } + } finally { + dispose(); // abort the socket (harmless if the body already ended) + } + const out = new Uint8Array(total); + let offset = 0; + for (const chunk of chunks) { + out.set(chunk, offset); + offset += chunk.length; + } + return out; +} + +/** One validated hop's result: a redirect `Location` to follow, or the delivered size-bounded bytes. */ +type HopOutcome = + | { readonly kind: 'redirect'; readonly location: string } + | { readonly kind: 'bytes'; readonly bytes: Uint8Array }; + +/** + * Perform ONE validated hop: validate the url + resolve / range-block / pin the host, open the pinned + * connection, and either surface a redirect `Location` (the caller re-validates it on the next hop) or + * read the size-bounded body. Split out of {@link fetchMediaBytes} to keep its cognitive complexity in + * budget (sonar S3776); any raw throw here is normalized to a typed `MediaEgressError` by the caller. + */ +async function performHop( + target: string, + deps: MediaEgressDeps, + allowPrivate: boolean, + signal: AbortSignal, + maxBytes: number, +): Promise { + const host = validateEgressHost(target); + const ips = await resolveValidatedIps(host, deps, allowPrivate); + // Connect by the FIRST validated IP — every IP was range-checked + confirmed an IP literal above, so + // pinning means the address validated is the address connected to (no re-resolve TOCTOU window). + const response = await deps.openConnection( + { url: target, hostname: host, pinnedIp: ips[0] ?? host }, + signal, + ); + if (isRedirectStatus(response.status)) { + response.dispose(); // never read a redirect body + const location = response.location; + if (location === undefined || location.length === 0) { + throw new MediaEgressError('bad_status', 'media egress redirect had no Location'); + } + return { kind: 'redirect', location }; + } + if (response.status !== 200) { + response.dispose(); + throw new MediaEgressError('bad_status', 'media egress received a non-200 status'); + } + return { kind: 'bytes', bytes: await readBounded(response.body, maxBytes, response.dispose) }; +} + +/** + * Fetch the bytes at a public-HTTPS `url`, enforcing the full SSRF + size-bound policy. The host + * mechanism the engine binds into a `MediaUrlFetch` hook (the engine supplies `maxBytes` + the + * `AbortSignal`). See the file header for the security contract. Its ONLY thrown type is + * {@link MediaEgressError} — every raw resolver / socket / `new URL` / body-read error is normalized. + */ +export async function fetchMediaBytes( + url: string, + options: FetchMediaBytesOptions, + deps: MediaEgressDeps = nodeMediaEgressDeps, +): Promise { + const maxRedirects = options.maxRedirects ?? DEFAULT_MAX_REDIRECTS; + const allowPrivate = options.allowPrivate ?? false; + const controller = new AbortController(); + const abort = (): void => controller.abort(); + if (options.signal?.aborted === true) { + controller.abort(); + } + options.signal?.addEventListener('abort', abort); // removed in the finally below + const timer = setTimeout(abort, options.timeoutMs ?? DEFAULT_TIMEOUT_MS); + try { + let target = url; + for (let redirects = 0; ; redirects += 1) { + if (redirects > maxRedirects) { + throw new MediaEgressError( + 'too_many_redirects', + 'media egress exceeded the redirect limit', + ); + } + const outcome = await performHop( + target, + deps, + allowPrivate, + controller.signal, + options.maxBytes, + ); + if (outcome.kind === 'bytes') { + return outcome.bytes; + } + // A relative Location resolves against the current url; the next iteration re-validates it (per-hop). + target = new URL(outcome.location, target).toString(); + } + } catch (error) { + if (error instanceof MediaEgressError) { + throw error; // a typed failure (blocked_host / too_large / bad_status / …) — preserve the discriminant + } + // Any RAW throw is normalized to the typed, secret-free network failure — a resolver DNS error, an + // openConnection socket error, a malformed-Location `new URL` TypeError, or an aborted body read — so + // fetchMediaBytes's ONLY thrown type is MediaEgressError (the function's contract; never a raw leak). + throw new MediaEgressError('network', 'media egress request failed'); + } finally { + clearTimeout(timer); + options.signal?.removeEventListener('abort', abort); + } +} + +/** The default Node deps: `node:dns` lookup (IP literal → itself) + a pinned `node:https` GET. */ +export const nodeMediaEgressDeps: MediaEgressDeps = { + resolveHost: async (hostname: string): Promise => { + if (isIP(hostname) !== 0) { + return [hostname]; // already an IP literal — no DNS round-trip + } + const records = await dnsLookup(hostname, { all: true }); + return records.map((record) => record.address); + }, + openConnection: (request: HopRequest, signal: AbortSignal): Promise => + new Promise((resolve, reject) => { + const parsed = new URL(request.url); + const family = isIP(request.pinnedIp) === 6 ? 6 : 4; + const clientRequest = httpsRequest( + { + protocol: 'https:', + hostname: request.hostname, + port: parsed.port === '' ? 443 : Number(parsed.port), + path: `${parsed.pathname}${parsed.search}`, + method: 'GET', + servername: request.hostname, // SNI + certificate hostname — TLS verification stays ON + // Pin to the pre-validated IP: the agent connects to exactly this address, never re-resolving. + lookup: (_hostname, _opts, callback) => callback(null, request.pinnedIp, family), + signal, + }, + (incoming) => { + const location = incoming.headers.location; + resolve({ + status: incoming.statusCode ?? 0, + location: typeof location === 'string' ? location : undefined, + body: incoming, + dispose: () => { + incoming.destroy(); + clientRequest.destroy(); + }, + }); + }, + ); + // A secret-free network failure — never echo the underlying message (it can carry the host/IP). + clientRequest.on('error', () => + reject(new MediaEgressError('network', 'media egress request failed')), + ); + clientRequest.end(); + }), +}; diff --git a/packages/db/src/media-store.test.ts b/packages/db/src/media-store.test.ts index f5bba7da..72caf0c2 100644 --- a/packages/db/src/media-store.test.ts +++ b/packages/db/src/media-store.test.ts @@ -84,3 +84,37 @@ describe('InMemoryMediaStore (1.AF — reference impl)', () => { expect([...(await store.get(handle))]).toEqual([1, 2, 3]); // still intact (get copied) }); }); + +describe('MediaStore.readRange (1.AF/D13 — byte-delivery Range gate)', () => { + it('InMemoryMediaStore reads a valid inclusive range and rejects an out-of-bounds one', async () => { + const store = new InMemoryMediaStore(); + const handle = await store.put(HELLO); // [0x68,0x65,0x6c,0x6c,0x6f] = "hello" + expect([...(await store.readRange(handle, { start: 1, end: 3 }))]).toEqual([0x65, 0x6c, 0x6c]); // "ell" + expect([...(await store.readRange(handle, { start: 0, end: 4 }))]).toEqual([...HELLO]); // whole + await expect(store.readRange(handle, { start: 0, end: 5 })).rejects.toThrow(/out of bounds/); + await expect(store.readRange(handle, { start: 3, end: 1 })).rejects.toThrow( + /reversed|>= start/, + ); + await expect(store.readRange(handle, { start: -1, end: 2 })).rejects.toThrow(/non-negative/); + }); + + it('InMemoryMediaStore.readRange rejects a malformed handle (never reads an unknown blob)', async () => { + const store = new InMemoryMediaStore(); + await expect(store.readRange('nonsense', { start: 0, end: 0 })).rejects.toThrow(/handle/); + }); + + it('FilesystemMediaStore reads a validated range (reusing the path-jail + sha256 integrity check)', async () => { + const root = mkdtempSync(join(tmpdir(), 'relavium-media-range-')); + try { + const store = new FilesystemMediaStore(root); + const handle = await store.put(HELLO); + expect([...(await store.readRange(handle, { start: 1, end: 2 }))]).toEqual([0x65, 0x6c]); // "el" + await expect(store.readRange(handle, { start: 2, end: 99 })).rejects.toThrow(/out of bounds/); + await expect(store.readRange('../../etc/passwd', { start: 0, end: 1 })).rejects.toThrow( + /handle/, + ); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }); +}); diff --git a/packages/db/src/media-store.ts b/packages/db/src/media-store.ts index 9137783d..7673aa8a 100644 --- a/packages/db/src/media-store.ts +++ b/packages/db/src/media-store.ts @@ -2,7 +2,13 @@ import { createHash, randomUUID } from 'node:crypto'; import { mkdir, readFile, rename, writeFile } from 'node:fs/promises'; import { dirname, join, resolve, sep } from 'node:path'; -import { MEDIA_HANDLE_PATTERN, type MediaSource, type MediaStore } from '@relavium/shared'; +import { + MEDIA_HANDLE_PATTERN, + validateByteRange, + type ByteRange, + type MediaSource, + type MediaStore, +} from '@relavium/shared'; /** * Host-side `MediaStore` implementations (1.AF, ADR-0042) — the Node CLI / VS Code blob store the @@ -39,6 +45,21 @@ function toBase64Source(bytes: Uint8Array): MediaSource { return { kind: 'base64', data: Buffer.from(bytes).toString('base64') }; } +/** + * Defensively slice an **inclusive** {@link ByteRange} from already-loaded bytes (1.AF/D13). The engine + * pre-validates the range against `media_objects.byteLength` ({@link validateByteRange}); this re-bounds + * against the ACTUAL stored size and fails closed on a bad range — never trusting the caller's size. + */ +function sliceRange(bytes: Uint8Array, range: ByteRange): Uint8Array { + const checked = validateByteRange(range, bytes.length); + if (!checked.ok) { + throw new Error(`media readRange: ${checked.reason}`); + } + // Slice with the VALIDATED snapshot (checked.range), never re-reading the input `range` — an + // accessor-backed / mutated range object cannot TOCTOU past the validated bounds. + return bytes.slice(checked.range.start, checked.range.end + 1); // inclusive end ⇒ +1 (exclusive slice) +} + /** * A filesystem content-addressed store (CAS). Bytes live under `//` (sharded by * the first hash byte). The digest is validated 64-lowercase-hex (from {@link MEDIA_HANDLE_PATTERN}), so @@ -87,6 +108,14 @@ export class FilesystemMediaStore implements MediaStore { return bytes; } + // The `signal` of the `MediaStore.readRange` contract is omitted (a fewer-param method satisfies the + // interface): the Node reference reads the bounded blob whole (it is already size-capped by the put/egress + // path) and reuses get()'s path-jail + sha256 integrity check, then defensively re-bounds the inclusive + // range. The real desktop Rust CAS (1.AH) streams the validated range with an abort. + async readRange(handle: string, range: ByteRange): Promise { + return sliceRange(await this.get(handle), range); + } + // `provider` is intentionally not a parameter yet (a fewer-param method satisfies the interface): // 1.AF resolves to inline base64 for every provider. The provider-aware file re-upload optimization // for over-ceiling media is the egress/sidecar work (1.AF/D8, ADR-0043); the engine, not the adapter, @@ -133,4 +162,10 @@ export class InMemoryMediaStore implements MediaStore { async resolveForEgress(handle: string): Promise { return toBase64Source(await this.get(handle)); } + + // The `signal` of the `MediaStore.readRange` contract is omitted (a fewer-param method still satisfies + // the interface): the in-memory blob is already resident, so there is no streamed I/O to abort. + async readRange(handle: string, range: ByteRange): Promise { + return sliceRange(await this.get(handle), range); + } } diff --git a/packages/llm/src/fallback-chain.test.ts b/packages/llm/src/fallback-chain.test.ts index fdb5bc47..86e4146a 100644 --- a/packages/llm/src/fallback-chain.test.ts +++ b/packages/llm/src/fallback-chain.test.ts @@ -49,6 +49,8 @@ interface FakeProvider { function makeProvider(opts: { id: ProviderId; supports?: { tools?: boolean; streaming?: boolean }; + /** A full capability override (e.g. a media-capable provider for the egress re-materialization tests). */ + capabilities?: CapabilityFlags; generate?: (req: LlmRequest, key: string, call: number) => Promise; stream?: (req: LlmRequest, key: string, call: number) => AsyncIterable; }): FakeProvider { @@ -57,7 +59,7 @@ function makeProvider(opts: { let streamCalls = 0; const provider: LlmProvider = { id: opts.id, - supports: caps(opts.supports), + supports: opts.capabilities ?? caps(opts.supports), generate(req, key) { calls.push(req); genCalls += 1; @@ -1394,3 +1396,175 @@ describe('cost + credential robustness', () => { expect(error?.type === 'error' && error.error.message).not.toContain(secret); }); }); + +// --- media egress re-materialization (1.AF, D7/D8, ADR-0043) ----------------------------------- + +describe('FallbackChain media egress re-materialization (D7/D8)', () => { + const MEDIA_CAPS: CapabilityFlags = { + tools: true, + streaming: true, + parallelToolCalls: false, + vision: true, + promptCache: false, + reasoning: false, + media: { + input: { image: true, audio: true, video: true, document: true }, + outputCombinations: [], + }, + }; + const HANDLE = `media://sha256-${'a'.repeat(64)}`; + const handleReq: LlmRequest = { + model: 'incoming', + messages: [ + { + role: 'user', + content: [ + { type: 'text', text: 'describe' }, + { type: 'media', mimeType: 'image/png', source: { kind: 'handle', ref: HANDLE } }, + ], + }, + ], + }; + const sentSource = (fake: FakeProvider, call = 0): unknown => { + const part = fake.calls[call]?.messages[0]?.content[1]; + return part?.type === 'media' ? part.source : null; + }; + + it('resolves a handle media source to its in-flight source before the provider sees it (D8)', async () => { + const provider = makeProvider({ + id: 'openai', + capabilities: MEDIA_CAPS, + generate: resolves('ok'), + }); + const resolveCalls: Array<{ handle: string; provider: ProviderId }> = []; + const { options } = makeOptions({ + resolveForEgress: (handle, p) => { + resolveCalls.push({ handle, provider: p }); + return Promise.resolve({ kind: 'base64', data: 'aGVsbG8=' }); + }, + }); + await new FallbackChain([entry(provider, 'gpt-x')], options).generate(handleReq); + expect(resolveCalls).toEqual([{ handle: HANDLE, provider: 'openai' }]); + expect(sentSource(provider)).toEqual({ kind: 'base64', data: 'aGVsbG8=' }); // no handle to the adapter + }); + + it('leaves a handle source unchanged when no resolveForEgress hook is wired (no-op)', async () => { + const provider = makeProvider({ + id: 'openai', + capabilities: MEDIA_CAPS, + generate: resolves('ok'), + }); + const { options } = makeOptions(); + await new FallbackChain([entry(provider, 'gpt-x')], options).generate(handleReq); + expect(sentSource(provider)).toEqual({ kind: 'handle', ref: HANDLE }); + }); + + it('caches a non-base64 provider ref and reuses it on a later same-provider call (byte-free sidecar, D7)', async () => { + const provider = makeProvider({ + id: 'openai', + capabilities: MEDIA_CAPS, + generate: resolves('ok'), + }); + let resolveCount = 0; + const { options } = makeOptions({ + resolveForEgress: () => { + resolveCount += 1; + return Promise.resolve({ kind: 'url', url: 'https://files.openai/abc' }); + }, + }); + const chain = new FallbackChain([entry(provider, 'gpt-x')], options); + await chain.generate(handleReq); + await chain.generate(handleReq); // a later call on the SAME instance + expect(resolveCount).toBe(1); // the provider ref was cached + reused (no re-upload) + expect(sentSource(provider, 1)).toEqual({ kind: 'url', url: 'https://files.openai/abc' }); + }); + + it('never caches a base64 source (byte-free sidecar) — re-resolves on each call', async () => { + const provider = makeProvider({ + id: 'openai', + capabilities: MEDIA_CAPS, + generate: resolves('ok'), + }); + let resolveCount = 0; + const { options } = makeOptions({ + resolveForEgress: () => { + resolveCount += 1; + return Promise.resolve({ kind: 'base64', data: 'aGVsbG8=' }); + }, + }); + const chain = new FallbackChain([entry(provider, 'gpt-x')], options); + await chain.generate(handleReq); + await chain.generate(handleReq); + expect(resolveCount).toBe(2); // base64 carries bytes — never cached; re-resolved (sidecar stays byte-free) + }); + + it('re-materializes per-provider across a failover (a foreign ref never crosses the boundary, D7)', async () => { + const primary = makeProvider({ + id: 'openai', + capabilities: MEDIA_CAPS, + generate: rejects('openai', 'overloaded'), + }); + const fallback = makeProvider({ + id: 'anthropic', + capabilities: MEDIA_CAPS, + generate: resolves('ok'), + }); + const seen: ProviderId[] = []; + const { options } = makeOptions({ + resolveForEgress: (_handle, p) => { + seen.push(p); + return Promise.resolve({ kind: 'url', url: `https://files.${p}/abc` }); + }, + }); + const chain = new FallbackChain( + [entry(primary, 'gpt-x'), entry(fallback, 'claude-x')], + options, + ); + await chain.generate(handleReq); + expect(seen).toEqual(['openai', 'anthropic']); // resolved independently for each provider + expect(sentSource(fallback)).toEqual({ kind: 'url', url: 'https://files.anthropic/abc' }); + }); + + it('advances to the next provider when re-materialization fails (retryable-advance, D7)', async () => { + const primary = makeProvider({ + id: 'openai', + capabilities: MEDIA_CAPS, + generate: resolves('primary'), + }); + const fallback = makeProvider({ + id: 'anthropic', + capabilities: MEDIA_CAPS, + generate: resolves('fallback'), + }); + const { options, trace } = makeOptions({ + resolveForEgress: (_handle, p) => + p === 'openai' + ? Promise.reject(new Error('upload failed')) + : Promise.resolve({ kind: 'base64', data: 'aGVsbG8=' }), + }); + const chain = new FallbackChain( + [entry(primary, 'gpt-x'), entry(fallback, 'claude-x')], + options, + ); + const res = await chain.generate(handleReq); + expect(res.content).toEqual([{ type: 'text', text: 'fallback' }]); // advanced past the failed materialization + expect(primary.calls).toHaveLength(0); // primary never got a request — materialization failed first + const failed = trace.find((r) => r.outcome === 'failed'); + expect(failed?.error?.message).toMatch(/re-materialization failed/); // visible + secret-free + }); + + it('makes a re-materialization failure fatal when there is no next provider', async () => { + const provider = makeProvider({ + id: 'openai', + capabilities: MEDIA_CAPS, + generate: resolves('ok'), + }); + const { options } = makeOptions({ + resolveForEgress: () => Promise.reject(new Error('upload failed')), + }); + const chain = new FallbackChain([entry(provider, 'gpt-x')], options); + const error = await rejectedError(chain.generate(handleReq)); + expect(error.message).toMatch(/re-materialization failed/); + expect(provider.calls).toHaveLength(0); + }); +}); diff --git a/packages/llm/src/fallback-chain.ts b/packages/llm/src/fallback-chain.ts index f9d19b8a..d3607abe 100644 --- a/packages/llm/src/fallback-chain.ts +++ b/packages/llm/src/fallback-chain.ts @@ -1,4 +1,4 @@ -import type { BackoffStrategy } from '@relavium/shared'; +import type { BackoffStrategy, ContentPart, MediaSource } from '@relavium/shared'; import type { CostTracker, CostUpdate } from './cost-tracker.js'; import { isRetryable, LlmProviderError, makeLlmError } from './llm-error.js'; @@ -153,6 +153,17 @@ export interface FallbackChainOptions { * `false`/absent makes the auth failure fatal. Never becomes a retry loop. */ readonly onAuthError?: (provider: ProviderId) => boolean | Promise; + /** + * Resolve a durable media `handle` to the in-flight {@link MediaSource} a specific provider's egress + * needs (1.AF/D8, [ADR-0043](../../../docs/decisions/0043-media-egress-failover-rematerialization-ssrf.md)). + * Host-injected (backed by `MediaStore.resolveForEgress`) so the **chain stays byte-free + platform-free** + * — it calls a function, never holds a `MediaStore`. Before each attempted entry the chain re-materializes + * every `handle` source in the request for that entry's provider (so the adapter only ever sees an + * already-resolved source); a non-base64 provider ref is cached per `(provider, handle)` and re-materialized + * on a cross-provider advance, a base64 source is never cached (the sidecar stays byte-free). Absent ⇒ a + * `handle` source is sent to the adapter unchanged (no-op). + */ + readonly resolveForEgress?: (handle: string, provider: ProviderId) => Promise; } const DEFAULT_BACKOFF_BASE_MS = 250; @@ -211,6 +222,13 @@ function isContentChunk(chunk: StreamChunk): boolean { return chunk.type !== 'stop' && chunk.type !== 'error'; } +/** True when any message carries a durable `handle` media source needing egress re-materialization (D8). */ +function hasHandleMedia(req: LlmRequest): boolean { + return req.messages.some((message) => + message.content.some((part) => part.type === 'media' && part.source.kind === 'handle'), + ); +} + /** What one `generate` attempt produced. */ type GenerateAttempt = | { readonly status: 'success'; readonly result: LlmResult } @@ -226,6 +244,12 @@ export class FallbackChain { readonly #cooldownMs: number; /** Per-provider cooldown expiry (ms), persisted across calls on this instance (rate-limit nuance). */ readonly #cooldownUntil = new Map(); + // The media egress re-materialization sidecar (ADR-0043 §4, D7): a per-`(provider, handle)` cache of + // NON-base64 resolved sources (a provider-hosted ref). It is byte-free (a base64 source is never cached), + // never persisted/logged/checkpointed, and survives across generate/stream calls on this instance so a + // multi-call tool loop reuses a provider ref instead of re-uploading; a cross-provider advance misses + // the cache (a different key) and re-materializes, so a foreign provider's ref never crosses a boundary. + readonly #egressSidecar = new Map(); // The cross-CALL reasoning strip latch (ADR-0039). Unlike the per-call `ChainRun.#lastProvider`, // this survives across generate/stream calls so a multi-turn tool loop on ONE chain instance strips // a prior provider's signed reasoning before it can reach a different provider's next call. A chain @@ -279,7 +303,11 @@ export class FallbackChain { continue; } const entryReq = run.beginEntry(entry); // strips on a provider boundary — only for attempted entries - const result = await this.#runEntryGenerate(entry, entryReq, req, run); + const materialized = await this.#materializeForEntry(entry, entryReq, run); + if (materialized === undefined) { + continue; // a failed media re-materialization advances to the next provider (retryable-advance) + } + const result = await this.#runEntryGenerate(entry, materialized, req, run); if (result !== undefined) { return result; } @@ -345,7 +373,11 @@ export class FallbackChain { continue; } const entryReq = run.beginEntry(entry); // strips on a provider boundary — only for attempted entries - const action = yield* this.#runEntryStream(entry, entryReq, req, run); + const materialized = await this.#materializeForEntry(entry, entryReq, run); + if (materialized === undefined) { + continue; // a failed media re-materialization advances to the next provider (retryable-advance) + } + const action = yield* this.#runEntryStream(entry, materialized, req, run); if (action === 'done') { return; } @@ -515,6 +547,77 @@ export class FallbackChain { } } + /** + * Re-materialize every durable `handle` media source in `entryReq` to the in-flight source `provider` + * needs (D8), via the injected {@link FallbackChainOptions.resolveForEgress} hook. Returns the resolved + * request, or `undefined` when the resolution failed — recorded as a visible failed attempt so the caller + * advances to the next provider (retryable-advance, ADR-0043 §4). A secret-free message names the reason + * only — never the handle, a resolved path, a host stack, or bytes. + */ + async #materializeForEntry( + entry: FallbackPlanEntry, + entryReq: LlmRequest, + run: ChainRun, + ): Promise { + try { + return await this.#materializeMedia(entryReq, entry.provider.id); + } catch { + const error = makeLlmError({ + provider: entry.provider.id, + kind: 'unknown', + message: 'media re-materialization failed before egress', + }); + run.lastError = error; + this.#emit({ ...run.next(entry), error }); + return undefined; + } + } + + /** + * Resolve every `handle` media source in the request to the in-flight source `provider` needs, leaving + * `base64`/`url` sources untouched. Non-mutating (returns a fresh request) and allocates only when the + * request actually carries a handle source (the dominant text/base64 case pays a cheap scan). + */ + async #materializeMedia(req: LlmRequest, provider: ProviderId): Promise { + const resolve = this.#options.resolveForEgress; + if (resolve === undefined || !hasHandleMedia(req)) { + return req; + } + const messages: LlmMessage[] = []; + for (const message of req.messages) { + const content: ContentPart[] = []; + for (const part of message.content) { + content.push( + part.type === 'media' && part.source.kind === 'handle' + ? { ...part, source: await this.#resolveHandle(part.source.ref, provider, resolve) } + : part, + ); + } + messages.push({ ...message, content }); + } + return { ...req, messages }; + } + + /** Resolve one handle for `provider`, caching only a non-base64 ref in the byte-free egress sidecar. */ + async #resolveHandle( + handle: string, + provider: ProviderId, + resolve: (handle: string, provider: ProviderId) => Promise, + ): Promise { + const key = `${provider}${handle}`; + const cached = this.#egressSidecar.get(key); + if (cached !== undefined) { + return cached; + } + const resolved = await resolve(handle, provider); + if (resolved.kind !== 'base64') { + // Cache only a provider ref (url/handle) — a base64 source carries bytes and is never cached, so the + // sidecar stays byte-free (ADR-0043 §4). A later same-provider re-use then skips the re-upload. + this.#egressSidecar.set(key, resolved); + } + return resolved; + } + /** Whether to skip an entry without consuming an attempt (cooldown or unmet capability). */ #skipReason( entry: FallbackPlanEntry, diff --git a/packages/shared/src/content.test.ts b/packages/shared/src/content.test.ts index d4320a21..3872db48 100644 --- a/packages/shared/src/content.test.ts +++ b/packages/shared/src/content.test.ts @@ -20,6 +20,7 @@ import { persistableMediaRefine, refineInFlightMediaPart, urlHasCredentials, + validateByteRange, } from './content.js'; import type { AbortSignalLike, ContentPart, DurableContentPart, MediaPart } from './content.js'; @@ -864,3 +865,48 @@ describe('extractHttpsHost + urlHasCredentials (SSRF URL policy)', () => { }); }); }); + +describe('validateByteRange (1.AF/D13 — the engine-pure byte-delivery Range policy)', () => { + it('accepts a valid inclusive range within bounds', () => { + expect(validateByteRange({ start: 0, end: 4 }, 10)).toEqual({ + ok: true, + range: { start: 0, end: 4 }, + }); + expect(validateByteRange({ start: 9, end: 9 }, 10)).toEqual({ + ok: true, + range: { start: 9, end: 9 }, + }); // last byte + }); + + it('rejects a negative bound (fail-closed)', () => { + expect(validateByteRange({ start: -1, end: 4 }, 10).ok).toBe(false); + expect(validateByteRange({ start: 0, end: -1 }, 10).ok).toBe(false); + }); + + it('rejects a reversed range (end < start)', () => { + const r = validateByteRange({ start: 5, end: 2 }, 10); + expect(r.ok).toBe(false); + expect(r.ok === false && r.reason).toMatch(/reversed|>= start/); + }); + + it('rejects an out-of-bounds end (end >= byteLength)', () => { + expect(validateByteRange({ start: 0, end: 10 }, 10).ok).toBe(false); // end == length is out of bounds + expect(validateByteRange({ start: 0, end: 99 }, 10).ok).toBe(false); + expect(validateByteRange({ start: 0, end: 0 }, 0).ok).toBe(false); // empty store: nothing readable + }); + + it('rejects non-integer bounds (no raw parseInt trust)', () => { + expect(validateByteRange({ start: 0.5, end: 4 }, 10).ok).toBe(false); + expect(validateByteRange({ start: 0, end: Number.NaN }, 10).ok).toBe(false); + }); + + it('fails closed on its own bad byteLength (NaN / Infinity / negative)', () => { + // The exported policy must not trust its caller's byteLength — a NaN/Infinity would otherwise make the + // `end >= byteLength` check silently pass an out-of-bounds range. + expect(validateByteRange({ start: 0, end: 0 }, Number.NaN).ok).toBe(false); + expect(validateByteRange({ start: 0, end: 0 }, Number.POSITIVE_INFINITY).ok).toBe(false); + expect(validateByteRange({ start: 0, end: 0 }, -1).ok).toBe(false); + const r = validateByteRange({ start: 0, end: 0 }, Number.NaN); + expect(r.ok === false && r.reason).toMatch(/byteLength/); + }); +}); diff --git a/packages/shared/src/content.ts b/packages/shared/src/content.ts index e46118d9..1831b0e8 100644 --- a/packages/shared/src/content.ts +++ b/packages/shared/src/content.ts @@ -53,6 +53,15 @@ export const MEDIA_MESSAGE_CAPS = { maxInlineBytesPerMessage: 2 * 1024 * 1024, } as const; +/** + * The default per-fetch upper bound on a re-hosted media `url` download (1.AF/D9, + * [ADR-0043](../../../docs/decisions/0043-media-egress-failover-rematerialization-ssrf.md) §2): the + * engine supplies this size-bound **policy** to the host media-egress mechanism, which streams the body + * and aborts the moment it exceeds the bound (never fully buffered). A configurable override is a + * config-spec concern; this is the conservative default. Tunable constant, not frozen shape. + */ +export const DEFAULT_MAX_MEDIA_DOWNLOAD_BYTES = 25 * 1024 * 1024; + /** * The `url` media carrier landing gate (ADR-0031 §Reserved shape): the SSRF range-primitive has * landed (1.AE — `extractHttpsHost`, `isPrivateOrLocalHost`, `urlHasCredentials`), so URL sources @@ -770,6 +779,62 @@ export interface MediaStore { * adapters stay pure string→string and never hold a `MediaStore` (ADR-0031 §adapter rule). */ resolveForEgress(handle: string, provider: LlmProviderId): Promise; + /** + * Read a validated byte {@link ByteRange} of a handle's bytes (1.AF/D13, + * [ADR-0044](../../../docs/decisions/0044-media-access-governance-read-media-save-to-cost.md) §1) — the + * host byte-delivery **mechanism**: `realpath`+`commonpath` fail-closed path resolution, symlinks OFF, + * streamed + size-bounded I/O. The engine owns the **policy** — it validates the range against the + * handle's `byteLength` ({@link validateByteRange}) BEFORE calling this; the host re-bounds defensively + * against the actual stored size (fail-closed, never trusting a client size). The `read_media` built-in + * (1.AF) and the 1.AH desktop `read_media(ref)` command share this one mechanism — the gate is never + * written twice. + */ + readRange(handle: string, range: ByteRange, signal?: AbortSignalLike): Promise; +} + +/** + * A byte range over a handle's bytes (1.AF/D13, ADR-0044 §1) — **inclusive** of both ends (HTTP + * `Range: bytes=start-end` semantics), so the served length is `end - start + 1`. The engine validates it + * against the handle's `byteLength` ({@link validateByteRange}) before the host reads it. + */ +export interface ByteRange { + readonly start: number; + readonly end: number; +} + +/** + * Validate a requested {@link ByteRange} against a handle's known `byteLength` (1.AF/D13) — the engine-pure + * byte-delivery **policy** (security-review.md §Media byte delivery). Fail-closed: rejects a non-integer, + * negative, reversed (`end < start`), or out-of-bounds (`end >= byteLength`) range — never a raw `parseInt` + * without a bound, never trusting a client-supplied size. Returns the validated range, or a secret-free + * reason (a range bound, never bytes). Reused by the `read_media` tool + the 1.AH desktop command, so the + * Range gate is written once. + */ +export function validateByteRange( + range: ByteRange, + byteLength: number, +): { ok: true; range: ByteRange } | { ok: false; reason: string } { + // Guard the policy's OWN `byteLength` input first — this exported primitive fails closed on a bad bound + // rather than trusting its caller (a future `read_media` caller passes a `nonNegativeInt.optional()`). + // `Number.isInteger` also rejects NaN/Infinity, which would otherwise make the `end >= byteLength` check + // silently pass an out-of-bounds range. + if (!Number.isInteger(byteLength) || byteLength < 0) { + return { ok: false, reason: 'byteLength must be a non-negative integer' }; + } + const { start, end } = range; + if (!Number.isInteger(start) || !Number.isInteger(end)) { + return { ok: false, reason: 'range bounds must be integers' }; + } + if (start < 0 || end < 0) { + return { ok: false, reason: 'range bounds must be non-negative' }; + } + if (end < start) { + return { ok: false, reason: 'range end must be >= start (not reversed)' }; + } + if (end >= byteLength) { + return { ok: false, reason: 'range end is out of bounds for the stored byte length' }; + } + return { ok: true, range: { start, end } }; } /** @@ -781,10 +846,27 @@ export interface MediaStore { * checkpoint snapshots a refine cannot reach). */ export interface DeInlineMedia { - (parts: readonly ContentPart[], store: MediaStore): Promise; - (value: unknown, store: MediaStore): Promise; + ( + parts: readonly ContentPart[], + store: MediaStore, + fetchUrl?: MediaUrlFetch, + ): Promise; + (value: unknown, store: MediaStore, fetchUrl?: MediaUrlFetch): Promise; } +/** + * The host-injected media-egress fetch (1.AF/D9, [ADR-0043](../../../docs/decisions/0043-media-egress-failover-rematerialization-ssrf.md) + * §2/§3): re-host a public-HTTPS `url` media source to its raw bytes, which the engine then + * content-addresses via {@link MediaStore.put}. The **host** performs the validated I/O — DNS resolve, + * connect by the validated IP, and re-run the one shared SSRF primitive ({@link isPrivateOrLocalHost} / + * {@link extractHttpsHost}) on **every redirect hop** — and enforces a streamed, size-bounded body so + * an over-size response is aborted and never fully buffered. The platform-pure engine supplies only the + * **decision** to fetch (and binds the per-fetch size bound + `AbortSignal` when it constructs the hook), + * so the seam type stays a narrow `url → bytes` function — never a socket in `@relavium/shared`. When no + * hook is wired, `deInlineMedia` hard-fails a `url` source (an un-re-hosted url may never persist, I3). + */ +export type MediaUrlFetch = (url: string) => Promise; + /* ------------------------------------------------------------------------------------------------ * SSRF range-block — the one shared primitive (1.AE, security-review.md, ADR-0031 §Guardrails) * diff --git a/packages/shared/src/media-deinline.test.ts b/packages/shared/src/media-deinline.test.ts index a98b7833..8563feae 100644 --- a/packages/shared/src/media-deinline.test.ts +++ b/packages/shared/src/media-deinline.test.ts @@ -37,6 +37,7 @@ function makeStubStore(): { }, resolveForEgress: () => Promise.reject(new Error('resolveForEgress is not exercised by deInlineMedia')), + readRange: () => Promise.reject(new Error('readRange is not exercised by deInlineMedia')), }; return { store, puts }; } @@ -152,6 +153,51 @@ describe('deInlineMedia (1.AF, ADR-0042 §2 — flight→durable transform)', () expect(puts).toHaveLength(0); }); + // --- D9: WITH an injected fetch hook, a canonical url media part is RE-HOSTED to a handle. + it('re-hosts a url media part to a handle via the injected fetch hook (D9 — no url in output)', async () => { + const { store, puts } = makeStubStore(); + const fetched: string[] = []; + const FETCH_BYTES = new Uint8Array([1, 2, 3, 4]); + const fetchUrl = (url: string): Promise => { + fetched.push(url); + return Promise.resolve(FETCH_BYTES); + }; + const urlPart: unknown = [ + { + type: 'media', + mimeType: 'image/png', + source: { kind: 'url', url: 'https://x.example/a.png' }, + }, + ]; + const out = (await deInlineMedia(urlPart, store, fetchUrl)) as { + source: unknown; + byteLength: number; + }[]; + expect(fetched).toEqual(['https://x.example/a.png']); // the host hook was called with the url + const put0 = puts[0]; + expect(put0).toBeDefined(); + expect(put0?.bytes).toEqual(FETCH_BYTES); // the fetched bytes were content-addressed + expect(out[0]?.source).toEqual({ kind: 'handle', ref: put0?.handle }); + expect(out[0]?.byteLength).toBe(4); + expect(JSON.stringify(out)).not.toContain('x.example'); // the url is gone — re-hosted to a handle (I3) + }); + + it('still hard-fails a mimeType-less url part EVEN WITH a fetch hook (nothing to content-address)', async () => { + const { store, puts } = makeStubStore(); + let fetchCalls = 0; + const fetchUrl = (): Promise => { + fetchCalls += 1; + return Promise.resolve(new Uint8Array([1])); + }; + const bare: unknown = { + type: 'media', + source: { kind: 'url', url: 'https://x.example/a.png' }, + }; + await expect(deInlineMedia(bare, store, fetchUrl)).rejects.toThrow(/no mimeType/); + expect(puts).toHaveLength(0); // fail-closed before the hook — no fetch, no put + expect(fetchCalls).toBe(0); // the hook is never invoked — it fails closed BEFORE any fetch + }); + // --- I3 leak regression: non-canonical byte carriers must HARD-FAIL, never pass through (review HIGH #1) it('hard-fails (no leak, no put) on a base64 data: URI string in an opaque value', async () => { const { store, puts } = makeStubStore(); @@ -207,6 +253,20 @@ describe('deInlineMedia (1.AF, ADR-0042 §2 — flight→durable transform)', () expect(puts).toHaveLength(0); }); + it('hard-fails (TypeError naming the real fault) on a base64 media source whose data is not a string', async () => { + const { store, puts } = makeStubStore(); + // Co-locate with a real carrier so the scan triggers the walk — a lone non-string-data source is NOT + // flagged (its data isn't a string), so it would otherwise fast-path through unchanged (and fail the + // durable Zod schema later). The walk reaches mediaPartBytes (isInflightMediaPart only needs a string + // mimeType + a string kind), and the message names the actual fault (non-string data), not "unsupported kind". + const bad: unknown = [ + { type: 'media', mimeType: 'image/png', source: { kind: 'base64', data: 42 } }, + base64MediaPart, + ]; + await expect(deInlineMedia(bad, store)).rejects.toThrow(/source\.data must be a string/); + expect(puts).toHaveLength(0); // throws on item 0 before the valid carrier is put + }); + it('hard-fails on an unknown media source kind (co-located with a real carrier so the scan runs)', async () => { const { store } = makeStubStore(); // A standalone { kind:'blob' } is byte/url-free, so the scan skips it (returns it unchanged); co-locating diff --git a/packages/shared/src/media-deinline.ts b/packages/shared/src/media-deinline.ts index e651aa4a..bfeed8a8 100644 --- a/packages/shared/src/media-deinline.ts +++ b/packages/shared/src/media-deinline.ts @@ -8,6 +8,7 @@ import { type ContentPart, type DurableContentPart, type MediaStore, + type MediaUrlFetch, } from './content.js'; /** @@ -22,13 +23,18 @@ import { * overload, an opaque event payload / node output / `tool_call.args` / `tool_result.result` a Zod * refine cannot recurse into. * + * A **`url`** media source (1.AF/D9, [ADR-0043](../../../docs/decisions/0043-media-egress-failover-rematerialization-ssrf.md) + * §3): when an optional `fetchUrl` host hook is injected, a canonical `url` media part is **re-hosted** — + * the host fetch performs the SSRF-validated, streamed, size-bounded connect, and the returned bytes are + * content-addressed via `MediaStore.put` exactly like a base64 source. When **no** hook is wired (or the + * url is malformed / has no mimeType to content-address against), the `url` source **hard-fails** — an + * un-re-hosted url may never persist (I3). + * * What it THROWS on (caught at the engine choke point → a single `run:failed`, or a stripped terminal): * a base64 `data:` URI string, a loose `{ kind:'base64' }` source not wrapped in a media part, a raw - * binary buffer, a media part with an unknown source kind or an unknown modality, and a **`url`** media - * source (re-hosting a url to a handle needs the host media-egress fetch and is the separate engine step - * — [ADR-0043](../../../docs/decisions/0043-media-egress-failover-rematerialization-ssrf.md), 1.AF/D9 — - * which re-hosts *before* this transform; until then an un-re-hosted url must never pass). Error - * messages name the carrier kind only — never the bytes/URL/handle/secret. + * binary buffer, a media part with an unknown source kind or an unknown modality, and an un-re-hostable + * `url` (no hook / malformed / no mimeType). Error messages name the carrier kind only — never the + * bytes/URL/handle/secret. * * The walk is **non-mutating** (returns new structures, never edits the input), **cycle-safe**, and * preserves `Array`/`Map`/`Set`/plain-object shape and shared references (one clone per distinct node). @@ -38,17 +44,26 @@ import { export function deInlineMedia( parts: readonly ContentPart[], store: MediaStore, + fetchUrl?: MediaUrlFetch, ): Promise; -export function deInlineMedia(value: unknown, store: MediaStore): Promise; -export async function deInlineMedia(value: unknown, store: MediaStore): Promise { +export function deInlineMedia( + value: unknown, + store: MediaStore, + fetchUrl?: MediaUrlFetch, +): Promise; +export async function deInlineMedia( + value: unknown, + store: MediaStore, + fetchUrl?: MediaUrlFetch, +): Promise { // No-unsafe-media fast path — the dominant text/handle-only emit pays only a cheap cycle-safe scan, // no store round-trip and no clone. This scan ALSO flags a url media part (containsDurableUnsafeMedia, // not the byte-only containsInlineMediaBytes), so a url-only payload is never skipped → it reaches the - // walk and throws (it must not silently persist an un-re-hosted url). + // walk and is re-hosted (with a hook) or throws (without one) — never silently persisted. if (!containsDurableUnsafeMedia(value)) { return value; } - return rewrite(value, store, new Map()); + return rewrite(value, store, new Map(), fetchUrl); } function isRecord(value: unknown): value is Record { @@ -69,6 +84,7 @@ function isInflightMediaPart(node: Record): boolean { async function rewriteMediaPart( node: Record, store: MediaStore, + fetchUrl: MediaUrlFetch | undefined, ): Promise> { const source = node['source']; const mimeType = node['mimeType']; @@ -79,27 +95,13 @@ async function rewriteMediaPart( if (kind === 'handle') { return node; // already durable — nothing to do } - if (kind === 'url') { - throw new Error( - 'deInlineMedia cannot re-host a url media source — the engine media-egress step (1.AF, ADR-0043) must materialize it to a handle first', - ); - } - const data = source['data']; - if (kind !== 'base64' || typeof data !== 'string') { - // An unknown source kind on a media part cannot be made durable-safe — fail closed (never pass through). - throw new Error( - `deInlineMedia: unsupported media source kind '${String(kind)}' on a media part`, - ); - } - // Fail-closed on an unknown modality (mirrors the seam ingestion refine) — mimeType is bounded (≤255) - // and not a secret/byte payload, so it is safe to name. + // Fail-closed on an unknown modality (mirrors the seam ingestion refine) — checked BEFORE resolving + // bytes so a bad mimeType never triggers a base64 decode or a url fetch. mimeType is bounded (≤255) + // and not a secret/byte payload, so it is safe to name. Applies to every non-handle carrier. if (mediaModalityOf(mimeType) === undefined) { throw new Error(`deInlineMedia: unsupported media mimeType '${mimeType}'`); } - const bytes = decodeBase64(data); - if (bytes === undefined) { - throw new Error('deInlineMedia: media source.data is not valid base64'); - } + const bytes = await mediaPartBytes(source, kind, fetchUrl); const handle = await store.put(bytes, mimeType); // Build the durable part: handle-only source + Y3 byteLength; preserve the text hints / duration. const durable: Record = { @@ -114,10 +116,56 @@ async function rewriteMediaPart( return durable; } +/** + * Resolve a non-handle media source to its raw bytes: decode a `base64` source, or **re-host** a `url` + * source via the injected host fetch (the SSRF-validated, size-bounded connect — D9). Fail-closed on an + * unknown source kind, a malformed source, or a `url` with no fetch hook wired (an un-re-hosted url must + * never reach a durable position, I3). + */ +async function mediaPartBytes( + source: Record, + kind: unknown, + fetchUrl: MediaUrlFetch | undefined, +): Promise { + if (kind === 'base64') { + const data = source['data']; + if (typeof data !== 'string') { + // A `typeof` guard ⇒ TypeError, and the message names the ACTUAL fault (a non-string `data`), not a + // misleading "unsupported kind" — base64 IS supported here. Secret-free: the data value is never + // interpolated (I3). The sibling domain/value checks below stay plain Error. + throw new TypeError('deInlineMedia: base64 media source.data must be a string'); + } + const bytes = decodeBase64(data); + if (bytes === undefined) { + throw new Error('deInlineMedia: media source.data is not valid base64'); + } + return bytes; + } + if (kind === 'url') { + const url = source['url']; + if (fetchUrl === undefined || typeof url !== 'string') { + // No host media-egress fetch wired (or a malformed url) — an un-re-hosted url must never pass. + throw new Error( + 'deInlineMedia cannot re-host a url media source — the engine media-egress step (1.AF, ADR-0043) must materialize it to a handle first', + ); + } + // The host hook performs the SSRF-validated, streamed, size-bounded connect (D9); we only consume bytes. + return fetchUrl(url); + } + // An unknown source kind on a media part cannot be made durable-safe — fail closed (never pass through). + // `kind` is interpolated bounded (slice 64) — on the unknown `unknown`-overload walk it is only typed + // `typeof === 'string'` (unlike the ≤255-bounded mimeType), so an opaque payload could otherwise supply + // an arbitrarily long string; the canonical values (base64/handle/url) are all short. + throw new Error( + `deInlineMedia: unsupported media source kind '${String(kind).slice(0, 64)}' on a media part`, + ); +} + async function rewrite( value: unknown, store: MediaStore, cache: Map, + fetchUrl: MediaUrlFetch | undefined, ): Promise { if (typeof value === 'string') { if (isBase64DataUri(value)) { @@ -141,7 +189,7 @@ async function rewrite( throw new Error('deInlineMedia: a raw binary buffer may not cross the durable boundary'); } - const container = await rewriteContainer(value, store, cache); + const container = await rewriteContainer(value, store, cache, fetchUrl); if (container !== null) { return container.clone; } @@ -150,7 +198,7 @@ async function rewrite( } if (isInflightMediaPart(value)) { - const durable = await rewriteMediaPart(value, store); + const durable = await rewriteMediaPart(value, store, fetchUrl); cache.set(value, durable); return durable; } @@ -161,8 +209,10 @@ async function rewrite( // (I3). A url part WITH a mimeType already throws inside rewriteMediaPart above; this is the opaque case. const urlSource = value['source']; if (value['type'] === 'media' && isRecord(urlSource) && urlSource['kind'] === 'url') { + // Distinct suffix from the mediaPartBytes url throw above: this arm is the mimeType-LESS url part + // (no content type to content-address against), so it can never be re-hosted even with a fetch hook. throw new Error( - 'deInlineMedia cannot re-host a url media source — the engine media-egress step (1.AF, ADR-0043) must materialize it to a handle first', + 'deInlineMedia cannot re-host a url media source with no mimeType — there is no content type to content-address against (1.AF, ADR-0043, I3)', ); } // A loose base64 source NOT wrapped in a media part has no mimeType to content-address — it cannot be @@ -176,7 +226,7 @@ async function rewrite( const clone: Record = {}; cache.set(value, clone); for (const [key, nested] of Object.entries(value)) { - clone[key] = await rewrite(nested, store, cache); + clone[key] = await rewrite(nested, store, cache, fetchUrl); } return clone; } @@ -191,12 +241,13 @@ async function rewriteContainer( value: object, store: MediaStore, cache: Map, + fetchUrl: MediaUrlFetch | undefined, ): Promise<{ clone: unknown } | null> { if (Array.isArray(value)) { const clone: unknown[] = []; cache.set(value, clone); for (const item of value) { - clone.push(await rewrite(item, store, cache)); + clone.push(await rewrite(item, store, cache, fetchUrl)); } return { clone }; } @@ -204,7 +255,7 @@ async function rewriteContainer( const clone = new Map(); cache.set(value, clone); for (const [k, v] of value) { - clone.set(await rewrite(k, store, cache), await rewrite(v, store, cache)); + clone.set(await rewrite(k, store, cache, fetchUrl), await rewrite(v, store, cache, fetchUrl)); } return { clone }; } @@ -212,7 +263,7 @@ async function rewriteContainer( const clone = new Set(); cache.set(value, clone); for (const item of value) { - clone.add(await rewrite(item, store, cache)); + clone.add(await rewrite(item, store, cache, fetchUrl)); } return { clone }; }