From 574f58442a1ccce71658cbfa17547cc4f6691d5c Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sat, 5 Sep 2026 06:33:00 +0000 Subject: [PATCH] feat: enforce input_schema at call time (ADR-016 INVALID_INPUT leg) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - OperationSpec.input_schema was advertise-only: services/schema disclosed it but no dispatch entry point consulted it (the only enforced schema was publish_schema per-chunk on Pub ops, P-03). - compile input_schema once at registration, same fail-closed rule as publish_schema/CF-003: an un-compilable schema is a registration error, never a silently-skipped contract. Validator cache mirrored on fork and in OperationRegistryBuilder like the publish validators. - check after the ACL gate in all three dispatch entry points: invoke, invoke_streaming, invoke_sink (via resolve_sink_handler, preserving the P-08 single-source-of-truth property). Violations return INVALID_INPUT with the input echoed in details. - raw-JSON-Schema semantics (permissive on unknown keys); adapters wanting closed-by-default keep their own hardening (alkhttp's CompiledInputSchema composes unchanged). - motivated by alktty review #001 L1: the channels open-op wrapper hands the registry-checked input to the OpenHandler as the authoritative params, which requires the registry to validate it. Verification: cargo test 596 lib (6 new: invoke/streaming/sink enforcement, fail-closed registration, permissive-{} compile, fork-carries-validator); --all-features 613; clippy -D warnings clean; fmt clean; doc 0 warnings; publish --dry-run ok. alkhttp 438+16 tests pass against 0.3.1 (registry) — re-verify against the published 0.4.0 after upload. --- CHANGELOG.md | 39 +++++ Cargo.lock | 2 +- Cargo.toml | 2 +- src/registry/registration.rs | 324 ++++++++++++++++++++++++++++++++++- 4 files changed, 358 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b167454..9e065ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,45 @@ 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/). +## [0.4.0] - 2026-09-05 + +`input_schema` is now enforced at call time (the advertise-vs-enforce +leg ADR-016 promised: `INVALID_INPUT` is the registry's +schema-mismatch error code). Minor bump — behavioral break for any +caller that was passing schema-violating inputs to ops declaring a +non-trivial `input_schema` and relying on the validation being +documentation-only. + +### Changed + +- **`OperationSpec.input_schema` is enforced by the registry on every + dispatch.** Until now the schema was advertise-only: `services/schema` + disclosed it, the wire and gateway paths disclosed it, but no dispatch + entry point consulted it (the only schema alkcall actually enforced + was `publish_schema` per-chunk on Pub ops, P-03). The schema now + compiles once at registration (same fail-closed rule as + `publish_schema`/CF-003: an un-compilable schema is a registration + error, never a silently-skipped contract) and is checked after the + ACL gate in all three dispatch entry points — `invoke`, + `invoke_streaming`, and `invoke_sink` (via `resolve_sink_handler`, + preserving the P-08 single-source-of-truth property). Violations + return `CallError::invalid_input("input failed input_schema + validation")` with the input echoed in `details`. Raw-JSON-Schema + semantics (permissive on unknown keys) — adapters that want + closed-by-default enforcement keep doing their own hardening, as + alkhttp's `CompiledInputSchema` already does; the two checks compose. + Built-in ops are unaffected: alkcall's own specs declare the empty + object schema (accepts anything) and `services/schema` already + returned `INVALID_INPUT` for a missing `name` before its handler ran. + Motivated by alktty's code review #001 (L1): the channels open-op + wrapper hands the registry-checked `input` to the `OpenHandler` as + the authoritative params, which requires the registry to actually + validate it. +- `OperationRegistry` gained an `input_validator` cache (compiled at + registration, mirrored on `fork` and in `OperationRegistryBuilder` + like the publish validators); `input_validator()` is `pub(crate)` — + the public surface is unchanged. + ## [0.3.1] - 2026-09-04 Bug-fix release: `services/list-peers` can now list peer-announced diff --git a/Cargo.lock b/Cargo.lock index 4f6b385..164bcfc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -27,7 +27,7 @@ dependencies = [ [[package]] name = "alkcall" -version = "0.3.1" +version = "0.4.0" dependencies = [ "async-trait", "bytes", diff --git a/Cargo.toml b/Cargo.toml index d53b79d..67a8bca 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "alkcall" -version = "0.3.1" +version = "0.4.0" edition = "2021" rust-version = "1.85" license = "MIT OR Apache-2.0" diff --git a/src/registry/registration.rs b/src/registry/registration.rs index 5fc9f74..97580de 100644 --- a/src/registry/registration.rs +++ b/src/registry/registration.rs @@ -109,6 +109,7 @@ impl HandlerRegistration { pub struct OperationRegistry { operations: RwLock>, validators: RwLock>>, + input_validators: RwLock>>, } impl OperationRegistry { @@ -116,23 +117,27 @@ impl OperationRegistry { Self { operations: RwLock::new(HashMap::new()), validators: RwLock::new(HashMap::new()), + input_validators: RwLock::new(HashMap::new()), } } /// Fork the registry: a deep copy of every registration (handlers - /// and provenance included) plus the cached publish-schema - /// validators (review 004 F-02/F-03). The fork is an independent - /// registry — later registrations on the original do not appear on - /// the fork and vice versa. It is the per-session composition seam - /// (ADR-047 §4 amendment): fork the base registry, register the - /// per-session ops (generic channel ops, openables, bootstrap ops) - /// against the fork, and dispatch the session over the fork. + /// and provenance included) plus the cached publish-schema and + /// input-schema validators (review 004 F-02/F-03). The fork is an + /// independent registry — later registrations on the original do + /// not appear on the fork and vice versa. It is the per-session + /// composition seam (ADR-047 §4 amendment): fork the base + /// registry, register the per-session ops (generic channel ops, + /// openables, bootstrap ops) against the fork, and dispatch the + /// session over the fork. pub fn fork(&self) -> OperationRegistry { let operations = self.operations.read().clone(); let validators = self.validators.read().clone(); + let input_validators = self.input_validators.read().clone(); OperationRegistry { operations: RwLock::new(operations), validators: RwLock::new(validators), + input_validators: RwLock::new(input_validators), } } @@ -163,6 +168,36 @@ impl OperationRegistry { Ok(()) } + /// Compile a spec's `input_schema` for call-time enforcement + /// (ADR-016: `INVALID_INPUT` is the registry's schema-mismatch + /// error code; the advertise-vs-enforce leg every protocol crate + /// leans on — the channels open-op path passes the registry-checked + /// `input` to its `OpenHandler` as the authoritative params). + /// Un-compilable schemas are rejected at registration time — the + /// same fail-closed rule as `publish_schema` (CF-003): registering + /// an un-validatable input contract must never produce an operation + /// that silently skips validation. A compiled validator is cached + /// so the dispatch path does zero per-request work; `None` means + /// "no schema declared" (the empty-object schema alkcall's own ops + /// declare accepts anything, but it is still compiled so the + /// dispatch path never re-parses it). + fn validate_and_cache_input_schema( + input_validators: &mut HashMap>, + spec: &OperationSpec, + ) -> Result<(), String> { + let compiled = match jsonschema::options().build(&spec.input_schema) { + Ok(v) => Some(v), + Err(e) => { + return Err(format!( + "input_schema for operation {:?} failed to compile: {e}", + spec.name + )); + } + }; + input_validators.insert(spec.name.clone(), compiled); + Ok(()) + } + pub fn register(&self, registration: HandlerRegistration) -> Result<(), String> { let expected = match registration.spec.op_type { OperationType::Query | OperationType::Mutation => "Once", @@ -181,6 +216,10 @@ impl OperationRegistry { )); } Self::validate_and_cache_publish_schema(&mut self.validators.write(), ®istration.spec)?; + Self::validate_and_cache_input_schema( + &mut self.input_validators.write(), + ®istration.spec, + )?; self.operations .write() .insert(registration.spec.name.clone(), registration); @@ -199,6 +238,41 @@ impl OperationRegistry { self.validators.read().get(name).cloned().flatten() } + /// The compiled call-time validator for an op's `input_schema` + /// (ADR-016: `INVALID_INPUT` is the registry's schema-mismatch + /// error). Compiled at registration time (CF-003 fail-closed — an + /// un-compilable schema is a registration error, never a silent + /// skip). Every declared schema compiles to `Some`; the `Option` + /// exists for map-miss on an unknown op name. + pub(crate) fn input_validator(&self, name: &str) -> Option { + self.input_validators.read().get(name).cloned().flatten() + } + + /// Validate `input` against the op's compiled `input_schema` + /// (ADR-016). Returns the `INVALID_INPUT` envelope on violation. + /// Called by all three dispatch entry points (`invoke`, + /// `invoke_streaming`, `invoke_sink` via `resolve_sink_handler`) + /// after the ACL check and before the handler — the registry-level + /// gate every protocol crate's dispatch funnels through. + fn check_input_schema( + &self, + registration: &HandlerRegistration, + input: &Value, + request_id: &str, + ) -> Result<(), ResponseEnvelope> { + let Some(validator) = self.input_validator(®istration.spec.name) else { + return Ok(()); + }; + if validator.is_valid(input) { + return Ok(()); + } + Err(ResponseEnvelope::error( + request_id.to_string(), + CallError::invalid_input("input failed input_schema validation") + .with_details(serde_json::json!({ "input": input })), + )) + } + pub fn list_operations(&self) -> Vec { self.operations .read() @@ -248,6 +322,10 @@ impl OperationRegistry { return ResponseEnvelope::forbidden(request_id, message); } + if let Err(envelope) = self.check_input_schema(®istration, &input, &request_id) { + return envelope; + } + match ®istration.handler { HandlerKind::Once(handler) => { let handler = Arc::clone(handler); @@ -316,6 +394,10 @@ impl OperationRegistry { })); } + if let Err(envelope) = self.check_input_schema(®istration, &input, &request_id) { + return Box::pin(stream::once(async move { envelope })); + } + let streaming_handler = match ®istration.handler { HandlerKind::Stream(h) => Arc::clone(h), HandlerKind::Once(_) => { @@ -395,6 +477,8 @@ impl OperationRegistry { return Err(ResponseEnvelope::forbidden(request_id, message)); } + self.check_input_schema(®istration, input, &request_id)?; + match ®istration.handler { HandlerKind::Sink(h) => Ok(Arc::clone(h)), HandlerKind::Once(_) => Err(ResponseEnvelope::error( @@ -441,6 +525,7 @@ impl Default for OperationRegistry { pub struct OperationRegistryBuilder { operations: HashMap, validators: HashMap>, + input_validators: HashMap>, } impl OperationRegistryBuilder { @@ -448,6 +533,7 @@ impl OperationRegistryBuilder { Self { operations: HashMap::new(), validators: HashMap::new(), + input_validators: HashMap::new(), } } @@ -460,6 +546,7 @@ impl OperationRegistryBuilder { Self { operations: registry.operations.read().clone(), validators: registry.validators.read().clone(), + input_validators: registry.input_validators.read().clone(), } } @@ -468,6 +555,10 @@ impl OperationRegistryBuilder { &mut self.validators, ®istration.spec, )?; + OperationRegistry::validate_and_cache_input_schema( + &mut self.input_validators, + ®istration.spec, + )?; let name = registration.spec.name.clone(); self.operations.insert(name, registration); Ok(self) @@ -477,6 +568,7 @@ impl OperationRegistryBuilder { OperationRegistry { operations: RwLock::new(self.operations), validators: RwLock::new(self.validators), + input_validators: RwLock::new(self.input_validators), } } @@ -2631,4 +2723,222 @@ mod tests { let response = registry.invoke("late/op", serde_json::json!({}), ctx).await; assert!(response.result.is_ok()); } + + // --- input_schema call-time enforcement (ADR-016 INVALID_INPUT leg) ---- + + use futures::stream::StreamExt; + + fn spec_with_input_schema(name: &str, op_type: OperationType, schema: Value) -> OperationSpec { + OperationSpec::new( + name, + op_type, + Visibility::External, + schema, + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ) + } + + #[tokio::test] + async fn invoke_rejects_input_failing_input_schema() { + let registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + spec_with_input_schema( + "echo", + OperationType::Query, + serde_json::json!({ + "type": "object", + "properties": { "n": { "type": "integer" } }, + "required": ["n"] + }), + ), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + + let ctx = root_context("req-schema-1", None, None, false, ScopedPeerEnv::empty()); + let response = registry.invoke("echo", serde_json::json!({}), ctx).await; + let err = response + .result + .expect_err("missing required field must be rejected"); + assert_eq!(err.code, "INVALID_INPUT"); + + let ctx = root_context("req-schema-2", None, None, false, ScopedPeerEnv::empty()); + let response = registry + .invoke("echo", serde_json::json!({"n": "not-an-int"}), ctx) + .await; + assert!(response.result.is_err(), "type mismatch must be rejected"); + + let ctx = root_context("req-schema-3", None, None, false, ScopedPeerEnv::empty()); + let response = registry + .invoke("echo", serde_json::json!({"n": 1}), ctx) + .await; + assert_eq!(response.result.unwrap(), serde_json::json!({"n": 1})); + } + + #[tokio::test] + async fn invoke_streaming_rejects_input_failing_input_schema() { + let registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + spec_with_input_schema( + "sub", + OperationType::Sub, + serde_json::json!({ + "type": "object", + "properties": { "channel": { "type": "string" } }, + "required": ["channel"] + }), + ), + HandlerKind::Stream(make_streaming_handler(|_input, ctx| { + Box::pin(futures::stream::once(async move { + ResponseEnvelope::ok(ctx.request_id, serde_json::json!({})) + })) as Pin + Send>> + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + + let ctx = root_context("req-sub-1", None, None, false, ScopedPeerEnv::empty()); + let mut responses: Vec<_> = registry + .invoke_streaming("sub", serde_json::json!({}), ctx) + .collect() + .await; + assert_eq!(responses.len(), 1); + let response = responses.pop().expect("one envelope"); + let err = response + .result + .expect_err("schema violation must be rejected"); + assert_eq!(err.code, "INVALID_INPUT"); + + let ctx = root_context("req-sub-2", None, None, false, ScopedPeerEnv::empty()); + let responses: Vec<_> = registry + .invoke_streaming("sub", serde_json::json!({"channel": "c"}), ctx) + .collect() + .await; + assert_eq!(responses.len(), 1); + assert!(responses[0].result.is_ok()); + } + + #[tokio::test] + async fn invoke_sink_rejects_input_failing_input_schema() { + let registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + spec_with_input_schema( + "pub", + OperationType::Pub, + serde_json::json!({ + "type": "object", + "properties": { "n": { "type": "integer" } }, + "required": ["n"] + }), + ), + HandlerKind::Sink(collect_sink_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + + let (chunk_tx, chunk_rx) = futures::channel::mpsc::channel::>(8); + drop(chunk_tx); + let publish_stream: PublishStream = Box::pin(chunk_rx); + let ctx = root_context("req-pub-1", None, None, false, ScopedPeerEnv::empty()); + let response = registry + .invoke_sink("pub", serde_json::json!({}), publish_stream, ctx) + .await; + let err = response + .result + .expect_err("schema violation must be rejected"); + assert_eq!(err.code, "INVALID_INPUT"); + } + + #[test] + fn uncompilable_input_schema_is_rejected_at_registration() { + let registry = OperationRegistry::new(); + let result = registry.register(HandlerRegistration::new( + spec_with_input_schema( + "broken", + OperationType::Query, + serde_json::json!({ + "required": "n" + }), + ), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )); + assert!( + result.is_err(), + "un-compilable input_schema must fail registration" + ); + assert!( + result.unwrap_err().contains("input_schema"), + "error must name the failing schema" + ); + } + + #[test] + fn empty_object_input_schema_compiles_and_accepts_anything() { + let registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + spec_with_input_schema("echo", OperationType::Query, serde_json::json!({})), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + assert!( + registry.input_validator("echo").is_some(), + "a declared schema compiles even when permissive" + ); + } + + #[tokio::test] + async fn fork_carries_input_validator_and_enforces() { + let registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + spec_with_input_schema( + "echo", + OperationType::Query, + serde_json::json!({ + "type": "object", + "properties": { "n": { "type": "integer" } }, + "required": ["n"] + }), + ), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + + let fork = registry.fork(); + let ctx = root_context("req-fork-schema", None, None, false, ScopedPeerEnv::empty()); + let response = fork.invoke("echo", serde_json::json!({}), ctx).await; + let err = response + .result + .expect_err("the fork must carry the compiled input validator"); + assert_eq!(err.code, "INVALID_INPUT"); + } }