alkdev/alkhttp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(gateway): block services/schema spec disclosure via the op path (review-002 PRJ-16)

Browse Source 
Internal/ACL-restricted op specs were readable through
POST /call {"operation":"services/schema","input":{"name":...}}
(the MCP call/batch tools identically): the outer-name pre-checks pass
(services/schema is External) and alkcall's services_schema_handler
projects any registered spec with no visibility/ACL check of its own
(alkcall CF-004 is the complete fix there).

- GatewayDispatch.invoke/invoke_streaming now apply the GET /schema
  route's is-internal + access-control checks to the meta-op's inner
  name input before dispatch (404 Internal / FORBIDDEN ACL), one
  interception point covering /call, /batch, /subscribe and the MCP
  call/batch tools; /publish cannot reach the Query-typed meta-op
- the visibility+ACL check is one shared fn (schema_disclosure_denial)
  used by the HTTP /schema route, the dispatch guard, and the MCP
  schema tool, so transports cannot drift
- when CF-004 lands, this guard remains as defense-in-depth (ADR-071)

Tests: dispatch-spine guard unit tests; /call 404 + 401/403 matrix,
/batch NOT_FOUND entry, /subscribe error event; MCP call/batch tools
via services/schema with an Internal inner name (mcp feature).

Verify: cargo test (405), --all-features (523), clippy default and
--all-features --all-targets -D warnings, fmt --check — all pass.
This commit is contained in:
glm-5.3-flash
2026-08-31 01:32:59 +00:00
parent a7f10ed04c
commit 207bca4f23
6 changed files with 744 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapters/to_mcp.rs
+173 -13
View File
@@ -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<CallError> {
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![(
src/gateway/dispatch.rs
+246
View File
@@ -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<CallError> {
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<CallError> {
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<OperationSpec>) -> Arc<OperationRegistry> {
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<ResponseEnvelope> = 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:?}"),
}
}
}
src/gateway/mod.rs
+2
View File
@@ -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;
src/gateway/routes.rs
+247
View File
@@ -1087,6 +1087,73 @@ mod tests {
Arc::new(registry)
}
fn registry_with_discovery_and_internal_op() -> Arc<OperationRegistry> {
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<dyn IdentityProvider> {
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<dyn IdentityProvider> = 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());