Fix hosted customer-storage security boundaries

Cargo.toml

@@ -26,7 +26,7 @@ hound = { version = "3.5", optional = true }
 jsonwebtoken = { version = "9", default-features = false, features = ["use_pem"] }
 mime_guess = "2.0"
 rand = { version = "0.9", features = ["std", "std_rng"] }
-reqwest = { version = "0.13", features = ["form", "json", "multipart"] }
+reqwest = { version = "0.13", features = ["form", "json", "multipart", "stream"] }
 schemars = "1.0"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
@@ -37,6 +37,7 @@ songbird = { version = "0.6", default-features = false, features = ["driver", "g
 sqlx = { package = "sqlx-core", version = "0.8", default-features = false, features = ["_rt-tokio", "_tls-rustls-ring-webpki", "chrono", "json"] }
 sqlx-postgres = { version = "0.8", features = ["chrono", "json"] }
 tokio = { version = "1.47", features = ["fs", "macros", "net", "process", "rt-multi-thread", "signal", "sync", "time"] }
+tokio-util = { version = "0.7", features = ["io"] }
 tower-http = { version = "0.6", features = ["fs"] }
 tracing = "0.1"
 urlencoding = "2"

docs/deployment/hosted-worker.md

@@ -27,6 +27,9 @@ conditions:
 - a notice channel is configured;
 - a retention period from 1 through 365 days is configured;
 - the monthly recording cap exists and has remaining seconds; and
+- a versioned opaque `customer_s3` or `customer_r2` destination is pinned with
+  its exact normalized account/bucket host and
+  `delete_after_verified_delivery`; and
 - at least one human participant is currently in the requested channel.
 
 Missing configuration, a failed first fetch, stale configuration, malformed
@@ -43,12 +46,13 @@ All requests use `Authorization: Bearer <workload token>` and include
 the server must bind the token to the configured worker ID and should keep
 these internal routes on private ingress or require workload mTLS.
 
-This incremental adapter does not yet implement a hosted storage destination.
-Every current provider (`customer_s3`, `customer_r2`, `google_drive`, and
-`managed_transient`) therefore fails closed at the worker, even when the
-control plane reports it as ready. Hosted recording remains disabled until a
-provider adapter uploads, verifies, and deletes or retains transient local data
-according to policy. Self-hosted local/PVC capture is unchanged.
+Hosted capture remains raw-audio record-only. The worker can deliver finalized
+WAV segments to `customer_s3` or `customer_r2` through short-lived,
+exact-object grants issued by the private control plane. It never receives or
+stores the customer's reusable storage credentials. Unsupported providers,
+missing destination ID/revision, and any transient policy except
+`delete_after_verified_delivery` fail closed. Self-hosted local/PVC capture and
+transcript content behavior are unchanged.
 
 `GET /internal/v1/worker/guild-configurations`
 
@@ -71,6 +75,11 @@ according to policy. Self-hosted local/PVC capture is unchanged.
       "remainingRecordingSeconds": 35900,
       "storageProvider": "customer_s3",
       "storageDestinationLabel": "Customer archive",
+      "storageDestinationId": "dst_01",
+      "storageDestinationRevision": "dstrev_04",
+      "storageAllowedHost": "customer-audio.s3.us-east-1.amazonaws.com",
+      "storageObjectKeyPrefix": "tenant-opaque-prefix/",
+      "transientDeletePolicy": "delete_after_verified_delivery",
       "ready": true,
       "blockedReasons": [],
       "desiredRecordingGeneration": 42
@@ -200,13 +209,84 @@ capture releases the reservation. Invalid or missing audio stays pending only
 inside the bounded settlement window and becomes a visible terminal
 operator-reconciliation item afterward. Recovery settles an abandoned capture;
 it never resumes capture or renews recording authority after a process crash.
+Header-only zero-duration WAVs never enter artifact delivery or delivery
+backpressure. Their paths are atomically placed in a separate crash-safe local
+cleanup outbox with session-stop evidence; cleanup revalidates that each target
+is a regular zero-sample WAV inside `captureDir` before removing it.
 
 Recovery requires durable shared access to both Postgres and `captureDir`
 across pod replacement. A hard crash can discard audio written after the most
 recent WAV checkpoint, so recovery intentionally may undercount by less than
 the five-second checkpoint interval rather than bill unpersisted wall-clock
-time. All hosted storage providers remain disabled in this branch, so no
-reservation or audio capture can enter this path yet.
+time.
+
+At finalization, the worker hashes each regular, non-symlink WAV and atomically
+persists the runtime artifact, immutable manifest, encrypted reservation lease,
+delivery job, usage job, stop evidence, and removal of capture recovery. It
+then calls `POST /internal/v1/worker/artifact-deliveries/prepare` with only the
+reservation authority and exact object manifest:
+
+```json
+{
+  "reservationId": "reservation_01",
+  "leaseToken": "opaque-reservation-lease",
+  "recordingId": "recording_01",
+  "artifactId": "artifact_01",
+  "artifactKind": "raw_audio_wav",
+  "segmentIndex": 1,
+  "contentLength": 123456,
+  "sha256": "64-lowercase-hex-characters",
+  "contentType": "audio/wav"
+}
+```
+
+Hosted capture disables per-SSRC source-stem creation, so the mixed WAV segment
+list is the exhaustive sensitive-audio manifest. Self-hosted source-stem
+behavior remains available and unchanged.
+
+The control plane derives organization, guild, destination, and object key from
+the reservation and responds with `operationId`, positive `generation`, exact
+echoes of `recordingId`, `artifactId`, `artifactKind`, and `segmentIndex`, an
+opaque safe `objectKey`, pinned destination ID/revision/provider, the exact
+configuration-pinned `allowedUploadHost`, and a short-lived HTTPS `PUT` URL
+plus signed headers. The URL must use effective HTTPS port 443, and its path
+must be exactly `/<objectKey>`; an explicit alternate port is a different
+origin and is rejected. The worker accepts no redirects and does not let the
+response choose the value used to approve its own host. Only AWS S3 hosts
+ending in `amazonaws.com` with an `s3` DNS label or the exact
+`<account>.r2.cloudflarestorage.com` host shape may be pinned by configuration.
+S3 uses bucket-qualified virtual-host style. Because R2 uses an
+account-qualified host with path-style buckets, the independently pinned
+`storageObjectKeyPrefix` begins with the destination bucket and tenant prefix.
+Every returned object key must remain under the pinned prefix for both
+providers.
+
+The worker parses the actual SigV4 query, not only the detached JSON expiry. It
+requires `AWS4-HMAC-SHA256`, a valid `s3/aws4_request` credential scope,
+`X-Amz-Date`, `X-Amz-Expires` of at most ten minutes, a 64-character lowercase
+signature, and a sorted `X-Amz-SignedHeaders` containing `host` plus every
+outgoing header. The SigV4 expiry must exactly equal `expiresAt` and remain
+inside the live policy window. Signed upload headers bind content type, length,
+and `x-amz-meta-callscribe-sha256`. AWS S3 additionally requires
+`x-amz-checksum-sha256` containing the base64 raw digest; R2 rejects that
+unsupported full-object checksum header. Uploads time out within eight minutes
+under a fifteen-minute database claim.
+
+After upload, the worker posts the generation and exact recording, artifact,
+kind, segment, and object-key identity to
+`/internal/v1/worker/artifact-deliveries/<operation-id>/verify`. The response
+contains an audit receipt carrying the same full identity and a separate
+short-lived presigned provider `HEAD` capability. The worker validates that
+capability with the same exact host, object path, and SigV4 rules, then calls
+the provider directly. A matching control-plane receipt alone never authorizes
+deletion. The provider `HEAD` must return the exact content length and
+`x-amz-meta-callscribe-sha256`; AWS S3 must also return its provider-verified
+`x-amz-checksum-sha256`. Only then does the worker persist verification, delete
+the local WAV, and mark delivery final. It never stores either presigned URL or
+its headers. Failed uploads or verification retain the local WAV for fenced
+retry, and an unsafe delivery older than ten minutes backpressures later hosted
+starts for that organization/guild. After twenty attempts the job is terminal
+and still backpressures capture for operator reconciliation.
 
 ## Rollout
 

migrations/20260812030000_hosted_artifact_delivery_outbox.sql

@@ -0,0 +1,156 @@
+-- Pin the customer-owned storage destination and transient deletion policy to
+-- the crash-recovery row. A later configuration refresh must never redirect an
+-- already-recorded artifact to a different destination.
+ALTER TABLE call_scribe_hosted_capture_recovery
+    ADD COLUMN IF NOT EXISTS organization_id TEXT,
+    ADD COLUMN IF NOT EXISTS guild_id TEXT,
+    ADD COLUMN IF NOT EXISTS storage_provider TEXT,
+    ADD COLUMN IF NOT EXISTS storage_destination_id TEXT,
+    ADD COLUMN IF NOT EXISTS storage_destination_revision TEXT,
+    ADD COLUMN IF NOT EXISTS storage_allowed_host TEXT,
+    ADD COLUMN IF NOT EXISTS storage_object_key_prefix TEXT,
+    ADD COLUMN IF NOT EXISTS transient_delete_policy TEXT;
+
+-- This is only an upgrade guard for rows created by a prerelease worker which
+-- could not enable hosted storage. The unsupported marker remains fail-closed.
+UPDATE call_scribe_hosted_capture_recovery
+SET organization_id = COALESCE(organization_id, 'org_private_alpha'),
+    guild_id = COALESCE(guild_id, 'unsupported_legacy'),
+    storage_provider = COALESCE(storage_provider, 'unsupported_legacy'),
+    storage_destination_id = COALESCE(storage_destination_id, 'unsupported_legacy'),
+    storage_destination_revision = COALESCE(storage_destination_revision, 'unsupported_legacy'),
+    storage_allowed_host = COALESCE(storage_allowed_host, 'unsupported.invalid'),
+    storage_object_key_prefix = COALESCE(storage_object_key_prefix, 'unsupported/'),
+    transient_delete_policy = COALESCE(transient_delete_policy, 'retain_pending_operator')
+WHERE organization_id IS NULL
+   OR guild_id IS NULL
+   OR storage_provider IS NULL
+   OR storage_destination_id IS NULL
+   OR storage_destination_revision IS NULL
+   OR storage_allowed_host IS NULL
+   OR storage_object_key_prefix IS NULL
+   OR transient_delete_policy IS NULL;
+
+ALTER TABLE call_scribe_hosted_capture_recovery
+    ALTER COLUMN organization_id SET NOT NULL,
+    ALTER COLUMN guild_id SET NOT NULL,
+    ALTER COLUMN storage_provider SET NOT NULL,
+    ALTER COLUMN storage_destination_id SET NOT NULL,
+    ALTER COLUMN storage_destination_revision SET NOT NULL,
+    ALTER COLUMN storage_allowed_host SET NOT NULL,
+    ALTER COLUMN storage_object_key_prefix SET NOT NULL,
+    ALTER COLUMN transient_delete_policy SET NOT NULL;
+
+CREATE TABLE IF NOT EXISTS call_scribe_hosted_artifact_delivery_outbox (
+    artifact_id TEXT PRIMARY KEY
+        REFERENCES call_scribe_artifacts(id) ON DELETE RESTRICT,
+    organization_id TEXT NOT NULL,
+    guild_id TEXT NOT NULL,
+    recording_id TEXT NOT NULL
+        REFERENCES call_scribe_capture_sessions(id) ON DELETE RESTRICT,
+    reservation_id TEXT NOT NULL,
+    encrypted_lease_token BYTEA,
+    encryption_nonce BYTEA,
+    artifact_kind TEXT NOT NULL CHECK (artifact_kind = 'raw_audio_wav'),
+    segment_index INTEGER NOT NULL CHECK (segment_index BETWEEN 1 AND 16),
+    local_path TEXT NOT NULL,
+    content_length BIGINT NOT NULL CHECK (content_length > 0),
+    sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'),
+    content_type TEXT NOT NULL CHECK (content_type = 'audio/wav'),
+    storage_provider TEXT NOT NULL CHECK (storage_provider IN ('customer_s3', 'customer_r2')),
+    storage_destination_id TEXT NOT NULL,
+    storage_destination_revision TEXT NOT NULL,
+    storage_allowed_host TEXT NOT NULL,
+    storage_object_key_prefix TEXT NOT NULL,
+    transient_delete_policy TEXT NOT NULL
+        CHECK (transient_delete_policy = 'delete_after_verified_delivery'),
+    status TEXT NOT NULL DEFAULT 'pending'
+        CHECK (status IN ('pending', 'in_progress', 'verified', 'delivered', 'failed')),
+    operation_id TEXT,
+    operation_generation BIGINT CHECK (operation_generation IS NULL OR operation_generation > 0),
+    operation_object_key TEXT,
+    receipt JSONB,
+    claim_owner TEXT,
+    claim_token TEXT,
+    claim_until TIMESTAMPTZ,
+    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
+    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    first_attempt_at TIMESTAMPTZ,
+    verified_at TIMESTAMPTZ,
+    local_deleted_at TIMESTAMPTZ,
+    completed_at TIMESTAMPTZ,
+    last_error TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    UNIQUE (reservation_id, artifact_kind, segment_index),
+    CHECK (
+        (operation_id IS NULL AND operation_generation IS NULL AND operation_object_key IS NULL)
+        OR (operation_id IS NOT NULL AND operation_generation IS NOT NULL
+            AND operation_object_key IS NOT NULL)
+    ),
+    CHECK (
+        (status = 'pending'
+            AND encrypted_lease_token IS NOT NULL
+            AND encryption_nonce IS NOT NULL
+            AND claim_token IS NULL
+            AND claim_until IS NULL)
+        OR (status = 'in_progress'
+            AND encrypted_lease_token IS NOT NULL
+            AND encryption_nonce IS NOT NULL
+            AND claim_owner IS NOT NULL
+            AND claim_token IS NOT NULL
+            AND claim_until IS NOT NULL)
+        OR (status = 'verified'
+            AND encrypted_lease_token IS NULL
+            AND encryption_nonce IS NULL
+            AND receipt IS NOT NULL
+            AND verified_at IS NOT NULL
+            AND ((claim_owner IS NULL AND claim_token IS NULL AND claim_until IS NULL)
+                OR (claim_owner IS NOT NULL AND claim_token IS NOT NULL
+                    AND claim_until IS NOT NULL)))
+        OR (status = 'delivered'
+            AND encrypted_lease_token IS NULL
+            AND encryption_nonce IS NULL
+            AND receipt IS NOT NULL
+            AND verified_at IS NOT NULL
+            AND local_deleted_at IS NOT NULL
+            AND completed_at IS NOT NULL
+            AND claim_token IS NULL
+            AND claim_until IS NULL)
+        OR (status = 'failed'
+            AND encrypted_lease_token IS NULL
+            AND encryption_nonce IS NULL
+            AND claim_token IS NULL
+            AND claim_until IS NULL)
+    )
+);
+
+CREATE INDEX IF NOT EXISTS call_scribe_hosted_artifact_delivery_claim_idx
+    ON call_scribe_hosted_artifact_delivery_outbox(next_attempt_at, claim_until, created_at)
+    WHERE status IN ('pending', 'in_progress', 'verified');
+
+CREATE INDEX IF NOT EXISTS call_scribe_hosted_artifact_delivery_backpressure_idx
+    ON call_scribe_hosted_artifact_delivery_outbox(organization_id, guild_id, created_at)
+    WHERE status <> 'delivered';
+
+-- Header-only zero-duration WAVs have no deliverable audio and no remaining
+-- reservation authority. Keep their local removal crash-safe without putting
+-- an impossible artifact into the delivery/backpressure state machine.
+CREATE TABLE IF NOT EXISTS call_scribe_hosted_zero_duration_cleanup_outbox (
+    local_path TEXT PRIMARY KEY,
+    organization_id TEXT NOT NULL,
+    recording_id TEXT NOT NULL
+        REFERENCES call_scribe_capture_sessions(id) ON DELETE RESTRICT,
+    claim_token TEXT,
+    claim_until TIMESTAMPTZ,
+    attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0),
+    next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    last_error TEXT,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    CHECK ((claim_token IS NULL AND claim_until IS NULL)
+        OR (claim_token IS NOT NULL AND claim_until IS NOT NULL))
+);
+
+CREATE INDEX IF NOT EXISTS call_scribe_hosted_zero_duration_cleanup_claim_idx
+    ON call_scribe_hosted_zero_duration_cleanup_outbox(next_attempt_at, claim_until, created_at);