diff --git a/CHANGELOG.md b/CHANGELOG.md index 1bb768a..41ad8bc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,49 @@ All notable changes to this crate are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this crate adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] + +Consumer-findings remediation (CF-001..004 from +`docs/reviews/consumer-findings-ledger.md` — alkhttp as the first real +consumer). One behavior change noted below; otherwise additive. + +### Changed + +- **Un-compilable `publish_schema` values are rejected at registration + time** (CF-003). `OperationRegistry::register` and every + `OperationRegistryBuilder` method now return `Err` when a Pub op's + `publish_schema` fails to compile. Previously the bad schema was + accepted and the dispatch path failed **open** (warn log, chunks + unvalidated). The compiled validator is now cached per-op + (`OperationRegistry::publish_validator`) and consumed by the dispatch + path — the per-request compile is gone. Code that registered + un-compilable schemas will now get a registration error instead of a + silently-unvalidated op. + +### Fixed + +- **`services/schema` no longer discloses Internal or ACL-restricted op + specs** (CF-004). The schema handler now applies the same gates as + `invoke()` — Internal-visibility rejection and `AccessControl::check` + with matching identity resolution. Restricted ops return spec-404 + (`NOT_FOUND`), consistent with "restricted ops don't exist" elsewhere; + no information leaks about the restricted surface. The gate is inside + the handler, so every transport is covered. +- **Demux `TooLarge` skip no longer allocates from the peer's header** + (CF-002). The skip now streams through a fixed 64 KiB buffer, and a + cumulative 256 MiB skipped-bytes budget tears down connections that + loop oversized headers. Previously the buffer was sized from the + untrusted `u32` length (up to ~4 GiB pinned per dribbling peer). +- **Write failures before request delivery are retryable** (CF-001). + New `CallError::connection_closed` (`CONNECTION_CLOSED`, `retryable: + true`) is returned when the `call.requested` frame write fails on any + consumer path (`call`/`subscribe`/`publish`, both stream modes) — the + call provably never reached the producer, so reconnect/retry is safe. + Mid-publish write failures and producer-side connection-teardown + failures remain non-retryable `INTERNAL` (delivery ambiguous). The + new code string is additive; the `retryable` flag is the + machine-readable signal for consumers. + ## [0.1.1] - 2026-08-17 A minor release that renames the ALPN prefix from `alknet/` to `alk/` diff --git a/docs/reviews/consumer-findings-ledger.md b/docs/reviews/consumer-findings-ledger.md index 7f86daf..26738f7 100644 --- a/docs/reviews/consumer-findings-ledger.md +++ b/docs/reviews/consumer-findings-ledger.md @@ -11,141 +11,101 @@ Format: date | found-in (alkhttp context) | severity | status. ## Open -### CF-004 — `services_schema_handler` discloses Internal/ACL-restricted op specs — no visibility or AccessControl check (2026-08-30) +(none) + +--- + +## Resolved + +### CF-004 — `services_schema_handler` discloses Internal/ACL-restricted op specs — no visibility or AccessControl check (2026-08-30) — RESOLVED 2026-08-31 - **Found in:** alkhttp Review 002, finding PRJ-16 (`alkhttp/docs/reviews/002-post-remediation-review.md`, Part F', verified against tree `91483a7` + alkcall source). Filed to this ledger 2026-08-30. `src/registry/discovery.rs:327-343` (`services_schema_handler`) — - the handler does a bare `registry.registration(&name)` and returns + the handler did a bare `registry.registration(&name)` and returned `spec_to_json(®.spec)` verbatim, with **no `Visibility` check and - no `AccessControl::check(peer_identity)`**. The built-in op itself is - External with default ACL, so alkhttp's per-route pre-checks (which - guard only the *outer* op name `services/schema`) cannot see the - inner name a caller smuggles in the input: `POST /call - {"operation":"services/schema","input":{"name":"secret/op"}}` returns - the full spec — input/output/error schemas and - `access_control.required_scopes` — of any Internal or admin-scoped - op, unauthenticated. The GET `/schema` route and the MCP `schema` - tool both enforce internal-op + ACL pre-checks on the same - underlying operation; the `/call` path (and the MCP `call` tool - dispatching through it) is the hole. -- **Impact:** unauthenticated disclosure of the complete contract of - every Internal operation, including the exact scopes needed for - escalation (the "reads the scopes needed to escalate" shape review - 001's PRJ-06 called out). Disclosure only — the op's handler is not - executed. -- **Suggested direction:** enforce inside the handler (the complete - fix, covers every transport): resolve the peer identity from - `OperationContext`, then apply the same two checks the HTTP route - does — `is_internal_op`-equivalent visibility rejection (spec-404) - and `AccessControl::check` → FORBIDDEN envelope. Defensive note: some - existing wire tests exercise `services/schema` with default-ACL ops - and would still pass; the new behavior is only observable against - Internal/ACL-restricted registrations. -- **alkhttp side:** an alkhttp-local mitigation (input pre-check in - `call_handler`'s dispatch path) is planned as review-002 remediation - regardless — but it is a per-transport patch; the handler-level fix - here is the one that closes the hole for the wire and every future - transport. -- **Status:** open — no alkcall change made. + no `AccessControl::check(peer_identity)`**. +- **Fix:** the handler now applies the same two gates as + `OperationRegistry::invoke` — the Internal-visibility rejection + (`!ctx.internal` → spec-404) and `AccessControl::check` with the same + identity resolution invoke() uses (`handler_identity` when internal, + `identity` otherwise). Restricted ops return `NOT_FOUND` (spec-404) + rather than `FORBIDDEN` — no information leaks about the restricted + surface's existence or shape. The gate is in the handler itself, so + every transport (wire `/call`, HTTP routes, MCP tools) is covered. +- **Status:** resolved — 2026-08-30. -### CF-003 — `publish_schema` compile failure is fail-open on the wire dispatch path (2026-08-30) +### CF-003 — `publish_schema` compile failure is fail-open on the wire dispatch path (2026-08-30) — RESOLVED 2026-08-30 - **Found in:** alkhttp Review 001 post-remediation task `review-001-publish-schema-validation-robust` (the GW-01/#1 follow-up work). While fixing alkhttp's HTTP-side instance of the pattern, the identical instance was verified on the wire dispatch path: `src/protocol/dispatch.rs:352-366` (`Dispatcher::dispatch`, - `OperationType::Pub` arm) — when a Pub op's `publish_schema` fails to - **compile** (`jsonschema::options().build(&schema)` errors, e.g. a - non-array `"required"`, a non-string `"type"`, an invalid `pattern` - regex), the pump logs `warn!("publish_schema failed to compile; - chunks will not be validated")` and proceeds with `validator: None`. -- **Impact:** silent unvalidated ingest, per-request. A Pub op (local - registration or remote import) whose schema is un-compilable by the - pinned jsonschema dialect silently receives arbitrary JSON over the - wire that the registration *declares* validated — the handler's - schema guarantee is transport-dependent and quietly false. One broken - schema in the registry turns that op into an unvalidated ingest path - with only a warn log as the tell. (Per-request recompile cost is the - secondary issue; alkhttp's gateway cache eliminated both.) -- **Suggested direction:** fail-closed — on compile failure, reject the - chunk (or abort the sink dispatch) with `INTERNAL`-class - `CallError` naming the compile failure in the log only (untrusted - schema text off the wire), never a silent skip; cache the compiled - validator per registration (hot-reload by schema-value comparison; - see alkhttp's `src/gateway/schema_cache.rs` for the worked shape). - Complementary hygiene: reject un-compilable `publish_schema` values - at **registration time** (`OperationRegistry::register` observing - `spec.publish_schema`, or `with_publish_schema` validating eagerly) - so the failure never reaches traffic. -- **alkhttp side:** fixed in alkhttp `1572a9d` (gateway cache + lazy - resolution + stream-terminal fail-closed). The wire path is out of - this crate's scope (AGENTS.md: no forking alkcall types; the pump is - alkcall-internal), so the same invariant currently holds only on the - HTTP path. -- **Status:** open — no alkcall change made. + `OperationType::Pub` arm) — when a Pub op's `publish_schema` failed to + **compile**, the pump logged a warn and proceeded with + `validator: None` (silent unvalidated ingest, plus per-request + recompile cost). +- **Fix:** fail-closed at the source — `publish_schema` is now compiled + at **registration time** in both insertion points + (`OperationRegistry::register` and `OperationRegistryBuilder::store`); + an un-compilable schema is a registration error, so an + unvalidated-ingest path can never be constructed. The compiled + validator is cached per-op (`OperationRegistry::publish_validator`) + and the dispatch path consumes the cache — the per-request compile is + gone. The `from_call` import path inherits the guarantee (imports + register through the same choke point); forwarded-chunk validation on + the producer side is the remote peer's dispatch path and now inherits + the same fail-closed property. +- **Behavior change (semver-relevant):** `register`/builder methods + return `Err` for an un-compilable `publish_schema` (previously: + accepted). Code that registered garbage schemas will now get a + registration error instead of a silently-unvalidated op. +- **Status:** resolved — 2026-08-30. -### CF-002 — demux `TooLarge` skip allocates the full peer-declared length up front (up to ~4 GiB from an 8-byte header) (2026-08-30) +### CF-002 — demux `TooLarge` skip allocates the full peer-declared length up front (up to ~4 GiB from an 8-byte header) (2026-08-30) — RESOLVED 2026-08-30 - **Found in:** alkhttp Review 001, finding WS-12 (cross-crate; `docs/reviews/001-initial-implementation-review.md`, Part B, verified against tree `4a825d3`). Filed to this ledger 2026-08-30. - `src/channels/adapter.rs:144` (`run_demux_loop_for_client`) — when a - parsed chunk header carries `length > MAX_CHUNK_LEN`, the - `ChunkError::TooLarge` arm allocates + `src/channels/adapter.rs:144` (`run_demux_loop_for_client`) — the + `ChunkError::TooLarge` arm allocated `let mut discard = vec![0u8; length as usize];` **before** reading: - the buffer is sized from the peer's untrusted 8-byte header, and - `length` is a `u32` (`0xFFFFFFFF` ≈ 4 GiB). -- **Impact:** memory-shape DoS, not a correctness bug (the resync/skip - itself is correct — alkcall's own resync test covers it). Any - authenticated peer can send header `[ch][len = 0xFFFFFFFF]` and then - dribble; the allocation is pinned for as long as the payload takes to - arrive (a dribbling peer can hold it indefinitely). Via the alkhttp WS - path this is trivially reachable by any authenticated browser; K - concurrent connections × 4 GiB headers → OOM before the dribble even - matters. The *normal* payload arm (`:159`, `vec![0u8; - header.length as usize]`) is safe — `parse_header` rejects - `length > MAX_CHUNK_LEN`, so it is bounded at 16 MiB; only the - `TooLarge` skip arm is unbounded. -- **Suggested direction:** stream-skip with a bounded fixed buffer - (e.g. a 64 KiB stack/fixed buffer, `read_exact` loop consuming - `length` bytes) instead of the pre-sized allocation. Optionally bound - cumulative skipped bytes per connection before tearing it down, so a - peer cannot loop `TooLarge` headers to burn bandwidth/CPU forever. -- **alkhttp side:** nothing to change there — this is demux-internal. - alkhttp's own read-side hardening (review-001-ws-pump-consolidation: - explicit inbound `max_message_size`/`max_frame_size`, idle-read - timeout) bounds the WS ingress, but a header that survives those caps - still hits this arm, so the fix belongs on the alkcall side. -- **Status:** open — no alkcall change made. + the buffer was sized from the peer's untrusted 8-byte header (`u32` + `length`, up to ~4 GiB). +- **Fix:** stream-skip with a fixed 64 KiB buffer + `read_exact` loop + consuming `length` bytes — no allocation sized from peer input. + Plus a cumulative skipped-bytes budget (256 MiB) that tears down the + connection when a peer loops `TooLarge` headers to burn bandwidth/CPU. + The budget resets on each valid chunk, so legitimate isolated + oversized chunks (the resync path, covered by the existing test) are + unaffected. The existing `demux_resyncs_after_oversized_chunk` test + passes unchanged. +- **Status:** resolved — 2026-08-30. -### CF-001 — `call_single_stream` write-failure maps to non-retryable `INTERNAL` (2026-08-29) +### CF-001 — `call_single_stream` write-failure maps to non-retryable `INTERNAL` (2026-08-29) — RESOLVED 2026-08-30 - **Found in:** alkhttp `from_wss` drop-monitor race tests (WS-02/CON-02, - review-001-ws-eof-signal). When the transport mux dies mid-call, the - write path fails fast and alkcall's `call_single_stream` write-failure - mapping produces a non-retryable - `INTERNAL: failed to write request frame` — even though the call never - reached the producer and a reconnect/retry would be safe. -- **Impact:** consumers that drop connections under load (WS EOF, network - flaps) get non-retryable errors for in-flight calls that are provably - not processed. Callers must either tolerate-or-discriminate two - envelope shapes for the same transport-death race, or miss retry - opportunities. -- **Suggested direction:** classify write failures that occur before any - response frame could arrive as retryable (`CONNECTION_CLOSED` or - equivalent) — or expose the distinction so consumers can decide. -- **alkhttp side:** the race test tolerates both outcomes for now - (asserts retryable only when the call is provably in-flight). If/when - alkcall fixes the mapping, alkhttp's ws-eof-signal test should tighten - back to retryable-only. -- **Status:** open — no alkcall change made. - ---- - -## Resolved - -(empty) \ No newline at end of file + review-001-ws-eof-signal). When the transport mux died mid-call, the + write path failed fast and the write-failure mapping produced a + non-retryable `INTERNAL: failed to write request frame` — even though + the call never reached the producer and a reconnect/retry would be + safe. +- **Fix:** new retryable `CallError::connection_closed` constructor + (`CONNECTION_CLOSED` code, `retryable: true`), applied **only** to + write failures where the call is provably undelivered: + `call.requested` frame write failures on all consumer paths + (`call`/`subscribe`/`publish`, single-stream and stream-per-request; + the publish pump tags its write stages so only the request-frame + stage is retryable). Mid-publish and completed-frame write failures + stay `INTERNAL` (delivery ambiguous — retry unsafe), and the + producer-side `fail_all(...)` on connection close stays `INTERNAL`. + The new code string is additive to the wire error vocabulary; + `CONNECTION_CLOSED` is new — consumers should treat unknown codes per + their existing policy (the `retryable` flag is the machine-readable + signal). +- **Status:** resolved — 2026-08-30. alkhttp follow-up: the + `review-001-ws-eof-signal` race test can now tighten back to + retryable-only asserts. \ No newline at end of file diff --git a/src/channels/adapter.rs b/src/channels/adapter.rs index d3434d2..4c9ccbf 100644 --- a/src/channels/adapter.rs +++ b/src/channels/adapter.rs @@ -128,6 +128,15 @@ impl ChannelsAdapter { reader: Box, policy: Option<&Arc>, ) { + // Stack buffer for skipping oversized payloads — sized once, never + // from peer-declared lengths (a peer-declared `length` is untrusted + // up to u32::MAX ≈ 4 GiB; pre-sizing off it is an allocation-shape + // DoS). A cumulative cap tears down dribbling peers that loop + // TooLarge headers to burn bandwidth indefinitely. + const SKIP_BUF_LEN: usize = 64 * 1024; + const MAX_CONSECUTIVE_SKIPPED_BYTES: u64 = 256 * 1024 * 1024; + let mut skip_buf = vec![0u8; SKIP_BUF_LEN]; + let mut skipped_since_last_valid: u64 = 0; let mut reader = reader; let mut header_buf = [0u8; CHUNK_HEADER_LEN]; loop { @@ -141,10 +150,23 @@ impl ChannelsAdapter { max = super::wire::MAX_CHUNK_LEN, "demux: chunk too large, skipping payload bytes" ); - let mut discard = vec![0u8; length as usize]; - if let Err(e) = reader.read_exact(&mut discard).await { - warn!(error = %e, "demux: failed to skip oversized payload"); - break; + let mut remaining = length as u64; + while remaining > 0 { + let take = remaining.min(SKIP_BUF_LEN as u64) as usize; + if let Err(e) = reader.read_exact(&mut skip_buf[..take]).await { + warn!(error = %e, "demux: failed to skip oversized payload"); + return; + } + remaining -= take as u64; + } + skipped_since_last_valid += length as u64; + if skipped_since_last_valid > MAX_CONSECUTIVE_SKIPPED_BYTES { + warn!( + skipped_since_last_valid, + max = MAX_CONSECUTIVE_SKIPPED_BYTES, + "demux: skipped-bytes budget exceeded; tearing down connection" + ); + return; } continue; } @@ -170,6 +192,7 @@ impl ChannelsAdapter { } }; manager.route_payload(header.channel_id, payload).await; + skipped_since_last_valid = 0; } Err(e) => { if e.kind() == std::io::ErrorKind::UnexpectedEof { diff --git a/src/protocol/connection.rs b/src/protocol/connection.rs index 2eccc13..8c54e92 100644 --- a/src/protocol/connection.rs +++ b/src/protocol/connection.rs @@ -235,7 +235,7 @@ impl CallConnection { }; if let Err(err) = self.write_request(send, &request_id, payload).await { - let call_error = CallError::internal(err); + let call_error = CallError::connection_closed(err); self.pending .lock() .handle_error(&request_id, call_error.clone()); @@ -276,7 +276,8 @@ impl CallConnection { let envelope = EventEnvelope::requested(&request_id, payload); if let Err(err) = writer.write_frame(&envelope).await { - let call_error = CallError::internal(format!("failed to write request frame: {err}")); + let call_error = + CallError::connection_closed(format!("failed to write request frame: {err}")); self.pending .lock() .handle_error(&request_id, call_error.clone()); @@ -353,7 +354,7 @@ impl CallConnection { }; if let Err(err) = self.write_request(send, &request_id, payload).await { - let call_error = CallError::internal(err); + let call_error = CallError::connection_closed(err); self.pending .lock() .handle_error(&request_id, call_error.clone()); @@ -385,7 +386,8 @@ impl CallConnection { let envelope = EventEnvelope::requested(&request_id, payload); if let Err(err) = writer.write_frame(&envelope).await { - let call_error = CallError::internal(format!("failed to write request frame: {err}")); + let call_error = + CallError::connection_closed(format!("failed to write request frame: {err}")); self.pending .lock() .handle_error(&request_id, call_error.clone()); @@ -498,8 +500,17 @@ impl CallConnection { }); let write_result = pump_publish_to_wire(send, &request_id, payload, stream).await; - if let Err(err) = write_result { - let call_error = CallError::internal(err); + if let Err((stage, err)) = write_result { + // A write failure before the request frame could be delivered is + // provably undelivered — retry is safe. Mid-publish failures leave + // delivery ambiguous; the INTERNAL mapping (non-retryable) is the + // safe default there. + let call_error = match stage { + PublishWriteStage::Request => CallError::connection_closed(err), + PublishWriteStage::Published | PublishWriteStage::Completed => { + CallError::internal(err) + } + }; self.pending .lock() .handle_error(&request_id, call_error.clone()); @@ -533,7 +544,8 @@ impl CallConnection { let requested = EventEnvelope::requested(&request_id, payload); if let Err(err) = writer.write_frame(&requested).await { - let call_error = CallError::internal(format!("failed to write request frame: {err}")); + let call_error = + CallError::connection_closed(format!("failed to write request frame: {err}")); self.pending .lock() .handle_error(&request_id, call_error.clone()); @@ -603,33 +615,48 @@ impl CallConnection { } } +/// Which frame of a publish pump failed to write. The `Request` stage is +/// provably undelivered (the producer never received `call.requested`, so a +/// reconnect/retry is safe); the later stages leave delivery ambiguous. +enum PublishWriteStage { + Request, + Published, + Completed, +} + async fn pump_publish_to_wire( send: W, request_id: &str, payload: Value, mut stream: Pin + Send>>, -) -> Result<(), String> +) -> Result<(), (PublishWriteStage, String)> where W: AsyncWrite + Unpin, { let mut writer = FrameFramedWriter::new(send); let requested = EventEnvelope::requested(request_id, payload); - writer - .write_frame(&requested) - .await - .map_err(|e| format!("failed to write request frame: {e}"))?; + writer.write_frame(&requested).await.map_err(|e| { + ( + PublishWriteStage::Request, + format!("failed to write request frame: {e}"), + ) + })?; while let Some(chunk) = stream.next().await { let envelope = EventEnvelope::published(request_id, chunk); - writer - .write_frame(&envelope) - .await - .map_err(|e| format!("failed to write published frame: {e}"))?; + writer.write_frame(&envelope).await.map_err(|e| { + ( + PublishWriteStage::Published, + format!("failed to write published frame: {e}"), + ) + })?; } let completed = EventEnvelope::completed(request_id); - writer - .write_frame(&completed) - .await - .map_err(|e| format!("failed to write completed frame: {e}"))?; + writer.write_frame(&completed).await.map_err(|e| { + ( + PublishWriteStage::Completed, + format!("failed to write completed frame: {e}"), + ) + })?; Ok(()) } @@ -1880,6 +1907,144 @@ mod tests { assert_eq!(read, envelope); } + // CF-001 — a transport write failure before the request frame could be + // delivered maps to retryable CONNECTION_CLOSED (the call provably + // never reached the producer). Mid-publish failures stay INTERNAL. + + /// AsyncWrite that always errors — simulates a dead mux mid-call. + struct BrokenWrite; + impl AsyncWrite for BrokenWrite { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + _buf: &[u8], + ) -> Poll> { + Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "mux dead", + ))) + } + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + + fn broken_write_bi_stream() -> crate::core::types::BiStream { + // The joined stream holds the read end; the write half always errors. + let (read_end, _write_end) = tokio::io::duplex(1024); + crate::core::types::BiStream::from_joined(read_end, BrokenWrite) + } + + #[tokio::test] + async fn single_stream_call_write_failure_is_retryable_connection_closed() { + let bi_stream = broken_write_bi_stream(); + let (writer, reader) = split_single_stream(bi_stream); + let conn = CallConnection::new_single_stream(stub_connection(), writer); + drop(reader); + + let response = conn.call("test/echo", serde_json::json!({})).await; + match response.result { + Err(e) => { + assert_eq!(e.code, "CONNECTION_CLOSED"); + assert!(e.retryable); + } + other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"), + } + } + + #[tokio::test] + async fn multi_stream_call_write_request_failure_is_retryable() { + // A `Connection` whose open_bi succeeds but whose write half is + // dead: the request write fails, the call provably undelivered. + use crate::core::types::StreamError; + use crate::core::types::{BiStream, BidiStreamSource, Connection}; + use std::net::SocketAddr; + + struct OneBiStream(std::sync::Mutex>); + #[async_trait::async_trait] + impl BidiStreamSource for OneBiStream { + async fn accept_bi(&self) -> Result { + Err(StreamError::ConnectionClosed) + } + async fn open_bi(&self) -> Result { + let stream = self.0.lock().expect("open once").take(); + stream.ok_or(StreamError::StreamClosed) + } + fn remote_addr(&self) -> Option { + None + } + fn close(&self, _code: u32, _reason: &str) {} + } + + let conn = Connection::from_source( + OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))), + b"alk/call".to_vec(), + ); + let call_conn = CallConnection::new(conn); + let response = call_conn.call("test/echo", serde_json::json!({})).await; + match response.result { + Err(e) => { + assert_eq!(e.code, "CONNECTION_CLOSED"); + assert!(e.retryable); + } + other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"), + } + } + + #[tokio::test] + async fn multi_stream_publish_request_frame_write_failure_is_retryable() { + use crate::core::types::StreamError; + use crate::core::types::{BiStream, BidiStreamSource, Connection}; + use std::net::SocketAddr; + + struct OneBiStream(std::sync::Mutex>); + #[async_trait::async_trait] + impl BidiStreamSource for OneBiStream { + async fn accept_bi(&self) -> Result { + Err(StreamError::ConnectionClosed) + } + async fn open_bi(&self) -> Result { + let stream = self.0.lock().expect("open once").take(); + stream.ok_or(StreamError::StreamClosed) + } + fn remote_addr(&self) -> Option { + None + } + fn close(&self, _code: u32, _reason: &str) {} + } + + let conn = Connection::from_source( + OneBiStream(std::sync::Mutex::new(Some(broken_write_bi_stream()))), + b"alk/call".to_vec(), + ); + let call_conn = CallConnection::new(conn); + let response = call_conn + .publish( + "fs/upload", + serde_json::json!({}), + Box::pin(futures::stream::iter(vec![serde_json::json!({"c": 1})])), + ) + .await; + match response.result { + Err(e) => { + // The pump died on the *request* frame — nothing delivered. + assert_eq!(e.code, "CONNECTION_CLOSED"); + assert!(e.retryable); + } + other => panic!("expected retryable CONNECTION_CLOSED, got {other:?}"), + } + } + + #[test] + fn connection_closed_error_is_retryable() { + let err = CallError::connection_closed("test"); + assert_eq!(err.code, "CONNECTION_CLOSED"); + assert!(err.retryable); + } + #[tokio::test] async fn shared_frame_writer_concurrent_writes_do_not_interleave() { let (client_end, server_end) = tokio::io::duplex(64 * 1024); diff --git a/src/protocol/dispatch.rs b/src/protocol/dispatch.rs index 68012f0..cc43de5 100644 --- a/src/protocol/dispatch.rs +++ b/src/protocol/dispatch.rs @@ -349,21 +349,7 @@ impl Dispatcher { Ok(h) => h, Err(envelope) => return DispatchResult::Once(envelope), }; - let publish_validator = self - .registry - .registration(&operation_name) - .and_then(|r| r.spec.publish_schema.as_ref()) - .and_then(|schema| match jsonschema::options().build(schema) { - Ok(v) => Some(v), - Err(e) => { - warn!( - operation = %operation_name, - error = %e, - "publish_schema failed to compile; chunks will not be validated", - ); - None - } - }); + let publish_validator = self.registry.publish_validator(&operation_name); let (chunk_tx, chunk_rx) = mpsc::channel::>(PUBLISH_CHANNEL_BUFFER); let publish_stream: PublishStream = Box::pin(chunk_rx); diff --git a/src/protocol/wire.rs b/src/protocol/wire.rs index e29a7cd..7917e27 100644 --- a/src/protocol/wire.rs +++ b/src/protocol/wire.rs @@ -113,6 +113,10 @@ impl CallError { Self::new("TIMEOUT", message, true) } + pub fn connection_closed(message: impl Into) -> Self { + Self::new("CONNECTION_CLOSED", message, true) + } + pub fn invalid_operation_type(message: impl Into) -> Self { Self::new("INVALID_OPERATION_TYPE", message, false) } diff --git a/src/registry/discovery.rs b/src/registry/discovery.rs index bd8e431..cbf70dc 100644 --- a/src/registry/discovery.rs +++ b/src/registry/discovery.rs @@ -4,7 +4,7 @@ use serde_json::{json, Value}; use super::context::OperationContext; use super::registration::{Handler, OperationRegistry}; -use super::spec::{AccessControl, OperationSpec, OperationType, Visibility}; +use super::spec::{AccessControl, AccessResult, OperationSpec, OperationType, Visibility}; use crate::protocol::wire::{CallError, ResponseEnvelope}; const NAME_SERVICES_LIST: &str = "services/list"; @@ -337,13 +337,35 @@ pub fn services_schema_handler(registry: Arc) -> Handler { ); } }; - match registry.registration(&name) { - Some(reg) => { - let spec_json = spec_to_json(®.spec); - ResponseEnvelope::ok(ctx.request_id, spec_json) - } - None => ResponseEnvelope::not_found(ctx.request_id, &name), + let registration = match registry.registration(&name) { + Some(reg) => reg, + None => return ResponseEnvelope::not_found(ctx.request_id, &name), + }; + + // Same gates as `OperationRegistry::invoke` — the schema of a + // restricted op is disclosed only to callers who could invoke + // it. Identity resolution mirrors invoke(): `handler_identity` + // (composition authority) when internal, `identity` otherwise. + if registration.spec.visibility == Visibility::Internal && !ctx.internal { + return ResponseEnvelope::not_found(ctx.request_id, &name); } + let identity = if ctx.internal { + ctx.handler_identity + .as_ref() + .and_then(|ca| ca.as_identity()) + } else { + ctx.identity.clone() + }; + if let AccessResult::Forbidden(_) = registration.spec.access_control.check( + identity.as_ref(), + None, + ctx.ownership.as_deref(), + ) { + return ResponseEnvelope::not_found(ctx.request_id, &name); + } + + let spec_json = spec_to_json(®istration.spec); + ResponseEnvelope::ok(ctx.request_id, spec_json) }) }) } @@ -720,6 +742,81 @@ mod tests { } } + // CF-004 — the schema handler must apply the same visibility + ACL + // gates as invoke(). Unrestricted ops stay fetchable; Internal, + // ACL-restricted (unauthorized), and *authorized* ACL-restricted + // shapes below. Restricted disclosures return NOT_FOUND (spec-404), + // matching "restricted ops don't exist" everywhere else. + + #[tokio::test] + async fn services_schema_hides_internal_op_from_external_caller() { + let registry = registry_with_access_controlled_ops(); + let handler = services_schema_handler(Arc::clone(®istry)); + let ctx = root_context("req-cf4-1"); + let response = handler(serde_json::json!({ "name": "internal/hidden" }), ctx).await; + match response.result { + Err(e) => assert_eq!(e.code, "NOT_FOUND"), + other => panic!("expected NOT_FOUND for internal op, got {other:?}"), + } + } + + #[tokio::test] + async fn services_schema_shows_internal_op_to_internal_caller() { + let registry = registry_with_access_controlled_ops(); + let handler = services_schema_handler(Arc::clone(®istry)); + let mut ctx = root_context("req-cf4-2"); + ctx.internal = true; + ctx.handler_identity = Some(CompositionAuthority::new("agent-chat", [])); + let response = handler(serde_json::json!({ "name": "internal/hidden" }), ctx).await; + let spec = response.result.expect("internal caller sees internal op"); + assert_eq!(spec.get("name"), Some(&json!("internal/hidden"))); + } + + #[tokio::test] + async fn services_schema_hides_acl_restricted_op_from_unauthorized_caller() { + let registry = registry_with_access_controlled_ops(); + let handler = services_schema_handler(Arc::clone(®istry)); + let ctx = root_context_with_identity( + "req-cf4-3", + Some(identity_with_scopes("regular-peer", &["user"])), + ); + let response = handler(serde_json::json!({ "name": "admin/secret" }), ctx).await; + match response.result { + Err(e) => assert_eq!(e.code, "NOT_FOUND"), + other => panic!("expected NOT_FOUND for unauthorized ACL, got {other:?}"), + } + } + + #[tokio::test] + async fn services_schema_shows_acl_restricted_op_to_authorized_caller() { + let registry = registry_with_access_controlled_ops(); + let handler = services_schema_handler(Arc::clone(®istry)); + let ctx = root_context_with_identity( + "req-cf4-4", + Some(identity_with_scopes("admin-peer", &["admin"])), + ); + let response = handler(serde_json::json!({ "name": "admin/secret" }), ctx).await; + let spec = response.result.expect("authorized caller sees the spec"); + assert_eq!(spec.get("name"), Some(&json!("admin/secret"))); + assert_eq!( + spec.get("access_control") + .and_then(|a| a.get("required_scopes")), + Some(&json!(["admin"])) + ); + } + + #[tokio::test] + async fn services_schema_unrestricted_op_fetchable_without_identity() { + let registry = registry_with_access_controlled_ops(); + let handler = services_schema_handler(Arc::clone(®istry)); + let ctx = root_context("req-cf4-5"); + let response = handler(serde_json::json!({ "name": "public/echo" }), ctx).await; + let spec = response + .result + .expect("default-ACL op fetchable unauthenticated"); + assert_eq!(spec.get("name"), Some(&json!("public/echo"))); + } + #[tokio::test] async fn services_list_handler_registered_and_invocable_via_registry() { let registry = registry_with_ops(); diff --git a/src/registry/registration.rs b/src/registry/registration.rs index 7f2937e..e3e73b3 100644 --- a/src/registry/registration.rs +++ b/src/registry/registration.rs @@ -96,15 +96,44 @@ impl HandlerRegistration { pub struct OperationRegistry { operations: HashMap, + validators: HashMap>, } impl OperationRegistry { pub fn new() -> Self { Self { operations: HashMap::new(), + validators: HashMap::new(), } } + /// Compile a spec's `publish_schema` (CF-003). Un-compilable schemas + /// are rejected at registration time — registering one is an error, + /// never a silent unvalidated ingest path. A compiled validator is + /// cached so the dispatch path does zero per-request work; the cached + /// value also distinguishes "no schema declared" (`None`) from a + /// schema that failed to compile, though the latter can no longer + /// reach this registry. + fn validate_and_cache_publish_schema( + validators: &mut HashMap>, + spec: &OperationSpec, + ) -> Result<(), String> { + let compiled = match &spec.publish_schema { + None => None, + Some(schema) => match jsonschema::options().build(schema) { + Ok(v) => Some(v), + Err(e) => { + return Err(format!( + "publish_schema for operation {:?} failed to compile: {e}", + spec.name + )); + } + }, + }; + validators.insert(spec.name.clone(), compiled); + Ok(()) + } + pub fn register(&mut self, registration: HandlerRegistration) -> Result<(), String> { let expected = match registration.spec.op_type { OperationType::Query | OperationType::Mutation => "Once", @@ -122,6 +151,7 @@ impl OperationRegistry { registration.spec.op_type, expected, actual )); } + Self::validate_and_cache_publish_schema(&mut self.validators, ®istration.spec)?; self.operations .insert(registration.spec.name.clone(), registration); Ok(()) @@ -131,6 +161,14 @@ impl OperationRegistry { self.operations.get(name) } + /// The cached per-chunk validator for a Pub op's `publish_schema` + /// (ADR-046 §4). `None` when the op declares no schema. Compiled at + /// registration time — a schema that fails to compile never reaches + /// this registry, so the dispatch path cannot fail open (CF-003). + pub fn publish_validator(&self, name: &str) -> Option { + self.validators.get(name).cloned().flatten() + } + pub fn list_operations(&self) -> Vec<&OperationSpec> { self.operations .values() @@ -371,21 +409,34 @@ impl Default for OperationRegistry { pub struct OperationRegistryBuilder { operations: HashMap, + validators: HashMap>, } impl OperationRegistryBuilder { pub fn new() -> Self { Self { operations: HashMap::new(), + validators: HashMap::new(), } } fn store(mut self, registration: HandlerRegistration) -> Result { + OperationRegistry::validate_and_cache_publish_schema( + &mut self.validators, + ®istration.spec, + )?; let name = registration.spec.name.clone(); self.operations.insert(name, registration); Ok(self) } + pub fn build(self) -> OperationRegistry { + OperationRegistry { + operations: self.operations, + validators: self.validators, + } + } + fn wrap_once(spec: &OperationSpec, handler: Handler) -> Result { match spec.op_type { OperationType::Query | OperationType::Mutation => Ok(HandlerKind::Once(handler)), @@ -572,12 +623,6 @@ impl OperationRegistryBuilder { pub fn with(self, registration: HandlerRegistration) -> Result { self.store(registration) } - - pub fn build(self) -> OperationRegistry { - OperationRegistry { - operations: self.operations, - } - } } impl Default for OperationRegistryBuilder { @@ -1852,6 +1897,129 @@ mod tests { ) } + // --- CF-003: publish_schema validated at registration time ------------- + + fn make_sink_noop() -> SinkHandler { + make_sink_handler(|_input, ctx, _stream| async move { + ResponseEnvelope::ok(ctx.request_id, serde_json::json!({})) + }) + } + + fn registering_pub_with_schema(schema: Value) -> Result<(), String> { + let mut registry = OperationRegistry::new(); + registry.register(HandlerRegistration::new( + pub_spec("fs/upload", AccessControl::default()).with_publish_schema(schema), + HandlerKind::Sink(make_sink_noop()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + } + + #[test] + fn register_rejects_uncompilable_publish_schema() { + // `required` must be an array — jsonschema rejects this at compile. + let err = registering_pub_with_schema(serde_json::json!({ + "type": "object", + "required": "not-an-array" + })) + .expect_err("un-compilable schema must be rejected at registration"); + assert!( + err.contains("failed to compile"), + "error names the failure: {err}" + ); + assert!(err.contains("fs/upload"), "error names the op: {err}"); + } + + #[test] + fn register_rejects_invalid_type_and_pattern_schema() { + assert!(registering_pub_with_schema(serde_json::json!({ + "type": 123 + })) + .is_err()); + + assert!(registering_pub_with_schema(serde_json::json!({ + "type": "object", + "properties": { "x": { "pattern": "(unclosed" } } + })) + .is_err()); + } + + #[test] + fn register_accepts_compilable_publish_schema_and_caches_validator() { + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + pub_spec("fs/upload", AccessControl::default()).with_publish_schema( + serde_json::json!({ + "type": "object", + "properties": { "bytes": { "type": "string" } }, + "required": ["bytes"] + }), + ), + HandlerKind::Sink(make_sink_noop()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .expect("compilable schema registers"); + assert!( + registry.publish_validator("fs/upload").is_some(), + "validator cached at registration" + ); + } + + #[test] + fn publish_validator_none_when_no_schema_declared() { + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + pub_spec("fs/upload", AccessControl::default()), + HandlerKind::Sink(make_sink_noop()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .expect("no-schema pub registers"); + assert!(registry.publish_validator("fs/upload").is_none()); + assert!(registry.publish_validator("no/such").is_none()); + } + + #[test] + fn registry_builder_rejects_uncompilable_publish_schema() { + let result = OperationRegistryBuilder::new().with_local_sink( + pub_spec("fs/upload", AccessControl::default()) + .with_publish_schema(serde_json::json!({ "type": "object", "required": "bad" })), + make_sink_noop(), + None, + None, + Capabilities::new(), + ); + assert!(result.is_err(), "builder path validates too"); + } + + #[test] + fn failed_validation_leaves_no_validator_cache_entry() { + let mut registry = OperationRegistry::new(); + let result = registry.register(HandlerRegistration::new( + pub_spec("fs/upload", AccessControl::default()) + .with_publish_schema(serde_json::json!({ "type": "object", "required": "bad" })), + HandlerKind::Sink(make_sink_noop()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )); + assert!(result.is_err()); + assert!( + registry.registration("fs/upload").is_none(), + "op not registered on schema failure" + ); + } + fn internal_pub_spec(acl: AccessControl) -> OperationSpec { OperationSpec::new( "fs/upload",