diff --git a/docs/architecture/README.md b/docs/architecture/README.md index 63705e5..e2130e9 100644 --- a/docs/architecture/README.md +++ b/docs/architecture/README.md @@ -68,6 +68,7 @@ types — the former `alknet-core` and `alknet-call` merged). | [068](decisions/068-gateway-publish-endpoint.md) | Gateway `/publish` Endpoint | 6th gateway endpoint for `OperationType::Pub` (producer→consumer streaming); NDJSON request body → `call.published` chunks | | [069](decisions/069-webtransport-out-of-scope.md) | WebTransport Out of Scope | h3/WebTransport removed from alkhttp scope entirely (an alknet concern); supersedes the deferral framing of ADR-044 | | [070](decisions/070-from-wss-consumer-adapter.md) | `from_wss` Consumer Adapter | Import a remote node's operations over WSS — same-protocol importer, channels-over-WS as transport; `wss` feature gate | +| [071](decisions/071-dispatch-schema-guard.md) | Dispatch-Spine `services/schema` Op-Path Guard | `GatewayDispatch` applies the GET `/schema` visibility+ACL checks to the meta-op's inner `name` (review-002 PRJ-16); alkcall CF-004 is the complete fix, this stays as defense-in-depth | ## Relevant Open Questions diff --git a/docs/architecture/decisions/071-dispatch-schema-guard.md b/docs/architecture/decisions/071-dispatch-schema-guard.md new file mode 100644 index 0000000..7b77de9 --- /dev/null +++ b/docs/architecture/decisions/071-dispatch-schema-guard.md @@ -0,0 +1,75 @@ +# ADR-071: The dispatch-spine `services/schema` op-path guard + +## Status + +Accepted + +## Context + +Review-002 PRJ-16 ([major, security]) found that the GET `/schema` +disclosure fixes (GW-02, SRV-02) had a sibling hole through the **op +path**: `services/schema` is an External operation with default ACL, so +the gateway's outer-name pre-checks admit it — and alkcall's +`services_schema_handler` performs a bare +`registry.registration(name)` → spec projection with **no visibility +and no AccessControl check** of its own. Any HTTP caller could fetch +the full spec (input/output/error schemas, `required_scopes`) of an +Internal or ACL-restricted operation with: + +``` +POST /call {"operation":"services/schema","input":{"name":"secret/op"}} +``` + +— unauthenticated, and identically through the MCP `call` and `batch` +tools. The same GET `/schema` route that returns 404 for that op was +thus circumventable in one request. (Disclosure only; the target op's +handler never executes.) + +The complete fix is alkcall-side: the handler itself must check — +filed as **CF-004** in `alkcall/docs/reviews/consumer-findings-ledger.md`. +But the alkcall lead time was unknown, and the hole is in this crate's +wire surface today. + +## Decision + +**`GatewayDispatch` blocks `services/schema` invocations whose inner +`name` input would be denied by the GET `/schema` route for the same +identity.** Before dispatch, when the resolved operation is the +`services/schema` meta-op, the spine applies the same two checks the +route applies to `?name=`: + +- Inner op is `Visibility::Internal` → `NOT_FOUND` (404 on HTTP). +- Inner op denies the caller's `AccessControl` → `FORBIDDEN` + (401 unauthenticated / 403 authenticated under the gateway error + mapping — identical to the GET `/schema` denials). + +The guard lives in the dispatch spine (`invoke` + `invoke_streaming`), +not in the route handlers, because every HTTP transport and the MCP +`call`/`batch` tools flow through it — one interception point covers +`/call`, `/batch`, `/subscribe` (defense-in-depth: the registry rejects +the Query-typed meta-op on the streaming path anyway), and the MCP +tools. Sink dispatch (`/publish`) cannot reach `services/schema`: the +registry rejects non-`Pub` operations before the handler runs and the +sink ignores the input, so nothing is projected there. + +The visibility + ACL check itself is one shared function +(`gateway::dispatch::schema_disclosure_denial`) used by the HTTP GET +`/schema` route, the dispatch-spine guard, and the MCP `schema` tool, +so the transports cannot drift. + +## Consequences + +- **Transport-level invariant (PRJ-16's wording):** no alkhttp + transport can fetch, through any path, a spec the GET `/schema` + route would deny for the same identity. +- **When CF-004 lands, the guard stays.** The alkcall-side handler + check is the complete fix for every transport (wire, overlays, peer + composition); this per-transport check remains as defense-in-depth. + Do not remove it. +- Wire-observable behavior changes only for the previously-leaking + requests: they now get the same 404/403 the GET route returns. + Legitimate `services/schema` calls (allowed inner names) are + unaffected. +- The guard matches the outer registration's `spec.name` against the + `services/schema` constant rather than the raw request string, so + leading-slash variants (`/services/schema`) hit the same check. \ No newline at end of file diff --git a/src/adapters/to_mcp.rs b/src/adapters/to_mcp.rs index e9c67c3..81c393e 100644 --- a/src/adapters/to_mcp.rs +++ b/src/adapters/to_mcp.rs @@ -34,7 +34,6 @@ use std::sync::Arc; use alkcall::core::auth::Identity; use alkcall::protocol::wire::{CallError, ResponseEnvelope}; use alkcall::registry::registration::OperationRegistry; -use alkcall::registry::spec::{AccessResult, Visibility}; use rmcp::model::{ CallToolRequestParams, CallToolResult, Implementation, JsonObject, ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, @@ -292,23 +291,16 @@ fn json_type_name(value: &Value) -> &'static str { /// Symmetric with HTTP `GET /schema` (PRJ-06): internal ops are /// invisible (NOT_FOUND regardless of caller), ACL-forbidden ops /// return FORBIDDEN. The MCP `schema` tool must never return the full -/// spec of an op the caller could not call. +/// spec of an op the caller could not call. The check is the shared +/// `schema_disclosure_denial` (review-002 PRJ-16) so the HTTP route, +/// the dispatch-spine `services/schema` guard, and this tool cannot +/// drift. fn schema_visibility_and_access_denial( registry: &OperationRegistry, operation: &str, identity: Option<&Identity>, ) -> Option { - let name = operation.strip_prefix('/').unwrap_or(operation); - let registration = registry.registration(name)?; - if registration.spec.visibility == Visibility::Internal { - return Some(CallError::not_found(operation)); - } - if let AccessResult::Forbidden(message) = - registration.spec.access_control.check(identity, None, None) - { - return Some(CallError::forbidden(message)); - } - None + crate::gateway::schema_disclosure_denial(registry, operation, identity) } fn map_search_response(response: ResponseEnvelope, query: Option<&str>) -> CallToolResult { @@ -1126,6 +1118,174 @@ mod tests { ); } + #[tokio::test] + async fn call_tool_via_services_schema_with_internal_name_returns_not_found() { + let mut inner = OperationRegistry::new(); + inner + .register(HandlerRegistration::new( + internal_spec("secret/op"), + HandlerKind::Once(make_echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + let inner = Arc::new(inner); + let mut dispatch_registry = OperationRegistry::new(); + dispatch_registry + .register(HandlerRegistration::new( + internal_spec("secret/op"), + HandlerKind::Once(make_echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + dispatch_registry + .register(HandlerRegistration::new( + services_list_spec(), + HandlerKind::Once(services_list_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + Some(ScopedPeerEnv::empty()), + Capabilities::new(), + )) + .unwrap(); + dispatch_registry + .register(HandlerRegistration::new( + services_schema_spec(), + HandlerKind::Once(services_schema_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + Some(ScopedPeerEnv::empty()), + Capabilities::new(), + )) + .unwrap(); + let gateway = ToMcpGateway::new(dispatch(Arc::new(dispatch_registry))); + + let mut args = Map::new(); + args.insert( + "operation".to_string(), + Value::String("services/schema".to_string()), + ); + args.insert( + "input".to_string(), + serde_json::json!({ "name": "secret/op" }), + ); + let result = invoke_tool(&gateway, "call", Some(args), None).await; + assert_eq!( + result.is_error, + Some(true), + "the MCP call tool must deny a spec the schema tool denies (PRJ-16)" + ); + let structured = result.structured_content.expect("structured error present"); + assert_eq!( + structured.get("code"), + Some(&Value::String("NOT_FOUND".to_string())) + ); + assert!( + structured.get("input_schema").is_none(), + "the internal op's spec must not leak: {structured}" + ); + } + + #[tokio::test] + async fn batch_tool_via_services_schema_with_internal_name_yields_not_found_entry() { + let mut inner = OperationRegistry::new(); + inner + .register(HandlerRegistration::new( + internal_spec("secret/op"), + HandlerKind::Once(make_echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + inner + .register(HandlerRegistration::new( + external_spec("echo/run", OperationType::Query, AccessControl::default()), + HandlerKind::Once(make_echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + let inner = Arc::new(inner); + let mut dispatch_registry = OperationRegistry::new(); + dispatch_registry + .register(HandlerRegistration::new( + internal_spec("secret/op"), + HandlerKind::Once(make_echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + dispatch_registry + .register(HandlerRegistration::new( + services_list_spec(), + HandlerKind::Once(services_list_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + Some(ScopedPeerEnv::empty()), + Capabilities::new(), + )) + .unwrap(); + dispatch_registry + .register(HandlerRegistration::new( + services_schema_spec(), + HandlerKind::Once(services_schema_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + Some(ScopedPeerEnv::empty()), + Capabilities::new(), + )) + .unwrap(); + dispatch_registry + .register(HandlerRegistration::new( + external_spec("echo/run", OperationType::Query, AccessControl::default()), + HandlerKind::Once(make_echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + let gateway = ToMcpGateway::new(dispatch(Arc::new(dispatch_registry))); + + let mut args = Map::new(); + args.insert( + "calls".to_string(), + serde_json::json!([ + { "operation": "services/schema", "input": { "name": "secret/op" } }, + { "operation": "services/schema", "input": { "name": "echo/run" } }, + ]), + ); + let result = invoke_tool(&gateway, "batch", Some(args), None).await; + assert_eq!(result.is_error, Some(false)); + let structured = result.structured_content.expect("structured present"); + let results = structured + .get("results") + .and_then(Value::as_array) + .expect("results array"); + assert_eq!(results.len(), 2); + assert_eq!(results[0].get("isError"), Some(&Value::Bool(true))); + assert_eq!( + results[0] + .get("error") + .and_then(|e| e.get("code")) + .cloned() + .unwrap_or(Value::Null), + Value::String("NOT_FOUND".to_string()) + ); + assert_eq!(results[1].get("isError"), Some(&Value::Bool(false))); + } + #[tokio::test] async fn batch_over_cap_returns_invalid_input_without_dispatching() { let registry = full_registry_with_ops(vec![( diff --git a/src/gateway/dispatch.rs b/src/gateway/dispatch.rs index 94cc5d2..9d3f98a 100644 --- a/src/gateway/dispatch.rs +++ b/src/gateway/dispatch.rs @@ -29,6 +29,31 @@ //! `deadline: None` (subscriptions are unbounded per alkcall ADR-021, //! and a `/publish` body is bounded by the client's upload, not a //! fixed window). +//! +//! # The `services/schema` op-path guard (review-002 PRJ-16) +//! +//! The registry-level pre-checks fire on the **outer** operation name +//! only, but `services/schema` is an External op whose *input* names +//! another operation — and the underlying handler does a bare +//! `registry.registration(name)` → spec projection with no visibility +//! or AccessControl check of its own (alkcall CF-004). Without a guard, +//! `POST /call {"operation":"services/schema","input":{"name":X}}` +//! returns the full spec of any Internal op, defeating the GET +//! `/schema` fix (GW-02/SRV-02) through the op path. [`invoke`] and +//! [`invoke_streaming`] therefore apply the same is-internal + +//! access-control checks to the inner `name` that GET `/schema` applies +//! (404 for Internal, FORBIDDEN for ACL denial) before dispatching. +//! This is the alkhttp-local layer of the fix; the complete fix is the +//! alkcall-side handler check (CF-004, +//! `alkcall/docs/reviews/consumer-findings-ledger.md`). When CF-004 +//! lands, this guard remains as defense-in-depth: the per-transport +//! check stays so no alkhttp transport can fetch a spec the GET +//! `/schema` route would deny for the same identity. +//! +//! Sink dispatch (`invoke_sink`, the `/publish` path) cannot reach +//! `services/schema`: the registry rejects non-`Pub` operations before +//! the handler runs, and the sink handler never reads the input, so +//! no spec is projected there. use std::collections::HashMap; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -41,11 +66,14 @@ use alkcall::protocol::wire::{CallError, ResponseEnvelope}; use alkcall::registry::context::{AbortPolicy, OperationContext, ScopedPeerEnv}; use alkcall::registry::env::LocalOperationEnv; use alkcall::registry::registration::OperationRegistry; +use alkcall::registry::spec::{AccessResult, Visibility}; use futures::stream::BoxStream; use serde_json::Value; const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); +const SERVICES_SCHEMA: &str = "services/schema"; + /// The shared dispatch spine over the registry, invoking operations for /// the neutral `ResponseEnvelope` result shape both gateway projections /// map to their wire formats. Identity arrives per-call as @@ -88,6 +116,11 @@ impl GatewayDispatch { ) -> ResponseEnvelope { self.invoke_count.fetch_add(1, Ordering::Relaxed); let operation_name = strip_leading_slash(op).to_string(); + if let Some(error) = + schema_via_call_denial(&self.registry, &operation_name, &input, identity.as_ref()) + { + return ResponseEnvelope::error(uuid::Uuid::new_v4().to_string(), error); + } let request_id = uuid::Uuid::new_v4().to_string(); let context = self.build_root_context(&request_id, &operation_name, identity); let result = tokio::time::timeout( @@ -116,6 +149,14 @@ impl GatewayDispatch { input: Value, ) -> BoxStream<'static, ResponseEnvelope> { let operation_name = strip_leading_slash(op).to_string(); + if let Some(error) = + schema_via_call_denial(&self.registry, &operation_name, &input, identity.as_ref()) + { + let request_id = uuid::Uuid::new_v4().to_string(); + return Box::pin(futures::stream::once(async move { + ResponseEnvelope::error(request_id, error) + })); + } let request_id = uuid::Uuid::new_v4().to_string(); let context = self.build_root_context_streaming(&request_id, &operation_name, identity); self.registry @@ -212,6 +253,53 @@ fn strip_leading_slash(operation_id: &str) -> &str { operation_id.strip_prefix('/').unwrap_or(operation_id) } +/// The `services/schema` op-path guard (review-002 PRJ-16): when the +/// dispatched operation is the schema meta-op, apply the GET `/schema` +/// route's visibility + AccessControl checks to the **inner** `name` +/// input — 404 for Internal, FORBIDDEN for ACL denial — so no transport +/// through this spine can fetch a spec the GET `/schema` route would +/// deny for the same identity. The complete fix is the alkcall-side +/// handler check (CF-004); this guard stays as defense-in-depth after +/// that lands. +fn schema_via_call_denial( + registry: &OperationRegistry, + operation: &str, + input: &Value, + identity: Option<&Identity>, +) -> Option { + let name = input.get("name").and_then(Value::as_str)?; + if !matches!( + registry.registration(operation), + Some(registration) if registration.spec.name == SERVICES_SCHEMA + ) { + return None; + } + schema_disclosure_denial(registry, name, identity) +} + +/// The per-transport schema-disclosure check shared by the HTTP GET +/// `/schema` route, the dispatch-spine `services/schema` guard, and the +/// MCP `schema` tool: Internal ops are invisible (404 / NOT_FOUND +/// regardless of caller), ACL-forbidden ops are denied (403 / +/// FORBIDDEN). One implementation so the transports cannot drift. +pub(crate) fn schema_disclosure_denial( + registry: &OperationRegistry, + operation: &str, + identity: Option<&Identity>, +) -> Option { + let name = strip_leading_slash(operation); + let registration = registry.registration(name)?; + if registration.spec.visibility == Visibility::Internal { + return Some(CallError::not_found(operation)); + } + if let AccessResult::Forbidden(message) = + registration.spec.access_control.check(identity, None, None) + { + return Some(CallError::forbidden(message)); + } + None +} + #[cfg(test)] mod tests { use super::*; @@ -374,4 +462,162 @@ mod tests { } assert_eq!(ticks.len(), 3); } + + fn registry_with_services_schema_over(inner_ops: Vec) -> Arc { + use alkcall::registry::discovery::{services_schema_handler, services_schema_spec}; + + let inner = Arc::new({ + let mut registry = OperationRegistry::new(); + for op_spec in &inner_ops { + registry + .register(HandlerRegistration::new( + op_spec.clone(), + HandlerKind::Once(make_handler(|input, ctx| async move { + ResponseEnvelope::ok(ctx.request_id, input) + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + } + registry + }); + let mut registry = OperationRegistry::new(); + for op_spec in &inner_ops { + registry + .register(HandlerRegistration::new( + op_spec.clone(), + HandlerKind::Once(make_handler(|input, ctx| async move { + ResponseEnvelope::ok(ctx.request_id, input) + })), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + } + registry + .register(HandlerRegistration::new( + services_schema_spec(), + HandlerKind::Once(services_schema_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + Arc::new(registry) + } + + #[tokio::test] + async fn invoke_of_services_schema_with_internal_inner_name_is_blocked_pre_dispatch() { + let registry = registry_with_services_schema_over(vec![spec( + "secret/op", + Visibility::Internal, + OperationType::Query, + )]); + let dispatch = GatewayDispatch::new(registry); + let envelope = dispatch + .invoke( + None, + "services/schema", + serde_json::json!({ "name": "secret/op" }), + ) + .await; + match envelope.result { + Err(error) => { + assert_eq!(error.code, "NOT_FOUND"); + assert!(error.message.contains("secret/op")); + } + Ok(v) => panic!("the internal op spec must not be returned, got {v:?}"), + } + } + + #[tokio::test] + async fn invoke_of_services_schema_with_authorized_inner_name_still_projects() { + let registry = registry_with_services_schema_over(vec![spec( + "public/op", + Visibility::External, + OperationType::Query, + )]); + let dispatch = GatewayDispatch::new(registry); + let envelope = dispatch + .invoke( + None, + "services/schema", + serde_json::json!({ "name": "public/op" }), + ) + .await; + assert!( + envelope.result.is_ok(), + "an allowed inner name must still project, got {envelope:?}" + ); + assert_eq!( + envelope + .result + .as_ref() + .ok() + .and_then(|v| v.get("name")) + .and_then(Value::as_str), + Some("public/op") + ); + } + + #[tokio::test] + async fn invoke_of_services_schema_with_forbidden_inner_name_is_denied() { + let restricted = OperationSpec::new( + "admin/op", + OperationType::Query, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl { + required_scopes: vec!["admin".to_string()], + ..Default::default() + }, + None, + ); + let registry = registry_with_services_schema_over(vec![restricted]); + let dispatch = GatewayDispatch::new(registry); + let envelope = dispatch + .invoke( + Some(Identity { + id: "user".to_string(), + scopes: vec!["user".to_string()], + resources: HashMap::new(), + }), + "services/schema", + serde_json::json!({ "name": "admin/op" }), + ) + .await; + match envelope.result { + Err(error) => assert_eq!(error.code, "FORBIDDEN"), + Ok(v) => panic!("the ACL-denied spec must not be returned, got {v:?}"), + } + } + + #[tokio::test] + async fn invoke_streaming_of_services_schema_with_internal_inner_name_is_blocked() { + let registry = registry_with_services_schema_over(vec![spec( + "secret/op", + Visibility::Internal, + OperationType::Query, + )]); + let dispatch = GatewayDispatch::new(registry); + let mut stream = dispatch.invoke_streaming( + None, + "services/schema", + serde_json::json!({ "name": "secret/op" }), + ); + let envelopes: Vec = stream.by_ref().collect().await; + assert_eq!(envelopes.len(), 1, "the guard error is the only item"); + match &envelopes[0].result { + Err(error) => assert_eq!(error.code, "NOT_FOUND"), + Ok(v) => panic!("the internal op spec must not be returned, got {v:?}"), + } + } } diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index c419bb6..d003d33 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -3,6 +3,8 @@ pub mod error; pub mod routes; pub(crate) mod schema_cache; +#[cfg(feature = "mcp")] +pub(crate) use dispatch::schema_disclosure_denial; pub use dispatch::GatewayDispatch; pub use routes::CallRequest; diff --git a/src/gateway/routes.rs b/src/gateway/routes.rs index 1207165..570eb88 100644 --- a/src/gateway/routes.rs +++ b/src/gateway/routes.rs @@ -1087,6 +1087,73 @@ mod tests { Arc::new(registry) } + fn registry_with_discovery_and_internal_op() -> Arc { + let mut inner = OperationRegistry::new(); + inner + .register(HandlerRegistration::new( + internal_spec("secret/op"), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + inner + .register(HandlerRegistration::new( + external_spec("echo/run", AccessControl::default()), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + let inner = Arc::new(inner); + let mut registry = OperationRegistry::new(); + registry + .register(HandlerRegistration::new( + services_list_spec(), + HandlerKind::Once(services_list_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + registry + .register(HandlerRegistration::new( + services_schema_spec(), + HandlerKind::Once(services_schema_handler(Arc::clone(&inner))), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + registry + .register(HandlerRegistration::new( + internal_spec("secret/op"), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + registry + .register(HandlerRegistration::new( + external_spec("echo/run", AccessControl::default()), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )) + .unwrap(); + Arc::new(registry) + } + fn unused_provider() -> Arc { Arc::new(StaticIdentityProvider::new()) } @@ -1385,6 +1452,186 @@ mod tests { assert_eq!(status, StatusCode::NOT_FOUND); } + #[tokio::test] + async fn call_services_schema_with_internal_name_returns_404() { + let router = build_router(registry_with_internal_op(), unused_provider()); + let req = json_request( + "POST", + "/call", + json!({ "operation": "services/schema", "input": { "name": "secret/op" } }), + ); + let (status, body) = send(router, req).await; + assert_eq!( + status, + StatusCode::NOT_FOUND, + "the op path must deny a spec the GET /schema route denies (PRJ-16): {body}" + ); + assert_eq!(body.get("code"), Some(&json!("NOT_FOUND"))); + assert!( + body.get("input_schema").is_none() && body.get("output").is_none(), + "the spec must not leak in any form: {body}" + ); + } + + #[tokio::test] + async fn call_services_schema_with_acl_restricted_name_returns_401_unauthenticated() { + let discovery = registry_with_discovery_and_ops(vec![HandlerRegistration::new( + external_spec( + "admin/secret", + AccessControl { + required_scopes: vec!["admin".to_string()], + ..Default::default() + }, + ), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )]); + let router = build_router(discovery, unused_provider()); + let req = json_request( + "POST", + "/call", + json!({ "operation": "services/schema", "input": { "name": "admin/secret" } }), + ); + let (status, body) = send(router, req).await; + assert_eq!( + status, + StatusCode::UNAUTHORIZED, + "FORBIDDEN with no identity maps to 401 (gateway error mapping): {body}" + ); + assert_eq!(body.get("code"), Some(&json!("FORBIDDEN"))); + } + + #[tokio::test] + async fn call_services_schema_with_acl_restricted_name_returns_403_for_unauthorized_identity() { + let discovery = registry_with_discovery_and_ops(vec![HandlerRegistration::new( + external_spec( + "admin/secret", + AccessControl { + required_scopes: vec!["admin".to_string()], + ..Default::default() + }, + ), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )]); + let provider: Arc = Arc::new( + StaticIdentityProvider::new() + .with_token("user-tok", identity_with_scopes("user", &["user"])), + ); + let router = build_router(discovery, provider); + let (k, v) = auth_header("user-tok"); + let req = Request::builder() + .method("POST") + .uri("/call") + .header("content-type", "application/json") + .header(k, v) + .body(Body::from( + serde_json::to_vec( + &json!({ "operation": "services/schema", "input": { "name": "admin/secret" } }), + ) + .unwrap(), + )) + .unwrap(); + let (status, body) = send(router, req).await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "an identity the GET /schema route would deny must be denied identically: {body}" + ); + assert_eq!(body.get("code"), Some(&json!("FORBIDDEN"))); + } + + #[tokio::test] + async fn call_services_schema_authorized_name_still_round_trips() { + let ops = vec![HandlerRegistration::new( + external_spec("echo/run", AccessControl::default()), + HandlerKind::Once(echo_handler()), + OperationProvenance::Local, + None, + None, + Capabilities::new(), + )]; + let discovery = registry_with_discovery_and_ops(ops); + let router = build_router(discovery, unused_provider()); + let req = json_request( + "POST", + "/call", + json!({ "operation": "services/schema", "input": { "name": "echo/run" } }), + ); + let (status, body) = send(router, req).await; + assert_eq!(status, StatusCode::OK); + let output = body + .get("output") + .and_then(|o| o.get("name")) + .cloned() + .unwrap_or(Value::Null); + assert_eq!(output, json!("echo/run")); + } + + #[tokio::test] + async fn batch_services_schema_with_internal_name_yields_not_found_entry() { + let discovery = registry_with_discovery_and_internal_op(); + let router = build_router(discovery, unused_provider()); + let req = json_request( + "POST", + "/batch", + json!([ + { "operation": "services/schema", "input": { "name": "secret/op" } }, + { "operation": "services/schema", "input": { "name": "echo/run" } }, + ]), + ); + let (status, body) = send(router, req).await; + assert_eq!(status, StatusCode::OK); + let results = body + .get("results") + .and_then(|r| r.as_array()) + .expect("results array"); + assert_eq!(results.len(), 2); + assert_eq!(results[0].get("result"), Some(&json!("error"))); + assert_eq!( + results[0] + .get("error") + .and_then(|e| e.get("code")) + .cloned() + .unwrap_or(Value::Null), + json!("NOT_FOUND") + ); + assert_eq!(results[1].get("result"), Some(&json!("ok"))); + } + + #[tokio::test] + async fn subscribe_on_services_schema_internal_name_emits_not_found_event() { + let discovery = registry_with_discovery_and_internal_op(); + let router = build_router(discovery, unused_provider()); + let req = json_request( + "POST", + "/subscribe", + json!({ "operation": "services/schema", "input": { "name": "secret/op" } }), + ); + let resp = router.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = resp.into_body().collect().await.unwrap().to_bytes(); + let body = String::from_utf8_lossy(&bytes); + assert!( + body.contains("event:error") || body.contains("event: error"), + "expected the guard's error event on the streaming path, got: {body}" + ); + assert!( + body.contains("NOT_FOUND"), + "expected NOT_FOUND from the op-path guard, got: {body}" + ); + assert!( + !body.contains("input_schema"), + "the inner op's spec must not leak through the streaming path: {body}" + ); + } + #[tokio::test] async fn batch_returns_array_of_results_in_order() { let router = build_router(registry_with_echo(), unused_provider());