diff --git a/src/gateway/routes.rs b/src/gateway/routes.rs index 0259d3b..769ed1d 100644 --- a/src/gateway/routes.rs +++ b/src/gateway/routes.rs @@ -1138,6 +1138,7 @@ mod tests { )), ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), ws_openable_alpns: None, + ws_op_register_acl: alkcall::registry::spec::AccessControl::default(), }; let auth_state = Arc::clone(&provider); gateway_router() diff --git a/src/server/adapter.rs b/src/server/adapter.rs index 7a2842d..91cd56c 100644 --- a/src/server/adapter.rs +++ b/src/server/adapter.rs @@ -99,6 +99,7 @@ pub struct HttpAdapter { ws_session_slots: Arc, ws_idle_timeout: Option, ws_openable_alpns: Option>, + ws_op_register_acl: alkcall::registry::spec::AccessControl, } impl HttpAdapter { @@ -138,6 +139,7 @@ impl HttpAdapter { ws_session_slots: Arc::clone(&ws_session_slots), ws_idle_timeout, ws_openable_alpns: None, + ws_op_register_acl: alkcall::registry::spec::AccessControl::default(), }; let router = build_router(state, None); Self { @@ -153,6 +155,7 @@ impl HttpAdapter { ws_session_slots, ws_idle_timeout, ws_openable_alpns: None, + ws_op_register_acl: alkcall::registry::spec::AccessControl::default(), } } @@ -169,6 +172,7 @@ impl HttpAdapter { ws_session_slots: Arc::clone(&self.ws_session_slots), ws_idle_timeout: self.ws_idle_timeout, ws_openable_alpns: self.ws_openable_alpns.clone(), + ws_op_register_acl: self.ws_op_register_acl.clone(), }; // `extra_routes` is borrowed, not consumed (SRV-05): a builder // call after `with_extra_routes` must keep the custom routes in @@ -191,6 +195,7 @@ impl HttpAdapter { ws_session_slots: Arc::clone(&self.ws_session_slots), ws_idle_timeout: self.ws_idle_timeout, ws_openable_alpns: self.ws_openable_alpns.clone(), + ws_op_register_acl: self.ws_op_register_acl.clone(), }; self.router = build_router(state, Some(routes.clone())); self.extra_routes = Some(routes); @@ -215,6 +220,7 @@ impl HttpAdapter { ws_session_slots: Self::rebuild_session_slots(max_sessions), ws_idle_timeout: self.ws_idle_timeout, ws_openable_alpns: self.ws_openable_alpns.clone(), + ws_op_register_acl: self.ws_op_register_acl.clone(), }; self.router = build_router(state, self.extra_routes.clone()); self @@ -247,6 +253,7 @@ impl HttpAdapter { ws_session_slots: Arc::clone(&self.ws_session_slots), ws_idle_timeout: self.ws_idle_timeout, ws_openable_alpns: self.ws_openable_alpns.clone(), + ws_op_register_acl: self.ws_op_register_acl.clone(), }; self.router = build_router(state, self.extra_routes.clone()); self @@ -293,6 +300,38 @@ impl HttpAdapter { ws_session_slots: Arc::clone(&self.ws_session_slots), ws_idle_timeout: self.ws_idle_timeout, ws_openable_alpns: self.ws_openable_alpns.clone(), + ws_op_register_acl: self.ws_op_register_acl.clone(), + }; + self.router = build_router(state, self.extra_routes.clone()); + self + } + + /// The `op/register` surface's `AccessControl` for WS sessions + /// (review 007 WS-29, implementing review-006 UP-02's recorded + /// posture): the per-session announce op is registered with this + /// ACL on each session's fork, so a peer whose identity does not + /// satisfy it gets `FORBIDDEN` on the announce. This is the + /// assembly layer's override surface for restricting which + /// authenticated peers may announce ops on WS sessions. + /// + /// Default: `AccessControl::default()` — any authenticated peer + /// may announce (the SRV-10 permissive-crate-default precedent). + /// Bare-registry / custom upgrade routes pass their own ACL via + /// the + /// [`OpRegisterAcl`](crate::websocket::OpRegisterAcl) request + /// extension instead. + pub fn with_ws_op_register_acl(mut self, acl: alkcall::registry::spec::AccessControl) -> Self { + self.ws_op_register_acl = acl; + let state = RouterState { + registry: Arc::clone(&self.registry), + identity_provider: Arc::clone(&self.identity_provider), + decoy: self.decoy.clone(), + openapi_doc: self.openapi_doc.clone(), + ws_sessions: Arc::clone(&self.ws_sessions), + ws_session_slots: Arc::clone(&self.ws_session_slots), + ws_idle_timeout: self.ws_idle_timeout, + ws_openable_alpns: self.ws_openable_alpns.clone(), + ws_op_register_acl: self.ws_op_register_acl.clone(), }; self.router = build_router(state, self.extra_routes.clone()); self @@ -1199,6 +1238,7 @@ mod tests { )), ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), ws_openable_alpns: None, + ws_op_register_acl: alkcall::registry::spec::AccessControl::default(), } } diff --git a/src/server/state.rs b/src/server/state.rs index 8b7a2ac..787a58a 100644 --- a/src/server/state.rs +++ b/src/server/state.rs @@ -53,6 +53,11 @@ pub(crate) struct RouterState { /// `None` (the default) declares no openables — a WS session then /// carries no data-channel open ops (the pre-Unit-2 shape). pub(crate) ws_openable_alpns: Option>, + /// The `op/register` surface's `AccessControl` for WS sessions + /// (review 007 WS-29): the per-session announce op is registered + /// with this ACL. Default: `AccessControl::default()` (the + /// permissive crate default). + pub(crate) ws_op_register_acl: alkcall::registry::spec::AccessControl, } impl axum::extract::FromRef for crate::websocket::SessionState { @@ -63,6 +68,7 @@ impl axum::extract::FromRef for crate::websocket::SessionState { Arc::clone(&state.ws_session_slots), state.ws_idle_timeout, state.ws_openable_alpns.clone(), + state.ws_op_register_acl.clone(), ) } } @@ -103,6 +109,7 @@ mod tests { )), ws_idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), ws_openable_alpns: None, + ws_op_register_acl: alkcall::registry::spec::AccessControl::default(), }; let extracted: DecoyConfig = axum::extract::FromRef::from_ref(&state); assert!(matches!(extracted, DecoyConfig::Redirect { .. })); diff --git a/src/websocket/mod.rs b/src/websocket/mod.rs index 778a0a9..1ff483f 100644 --- a/src/websocket/mod.rs +++ b/src/websocket/mod.rs @@ -38,8 +38,9 @@ pub use byte_adapter::split_tungstenite_to_bytes; pub(crate) use upgrade::adapter_install_channel_zero; #[cfg(feature = "server")] pub use upgrade::{ - run_channels_session, ws_bearer_auth, ws_upgrade_handler, ChannelsPolicy, OpenableAlpn, - OpenableAlpns, SessionState, WsSessions, WsTimeouts, DEFAULT_WS_MAX_SESSIONS, + run_channels_session, ws_bearer_auth, ws_upgrade_handler, ChannelsPolicy, OpRegisterAcl, + OpenableAlpn, OpenableAlpns, SessionSlots, SessionState, WsSessions, WsTimeouts, + DEFAULT_WS_MAX_SESSIONS, }; #[cfg(all(any(test, feature = "test-support"), feature = "server"))] diff --git a/src/websocket/upgrade.rs b/src/websocket/upgrade.rs index 9263e07..c9b4cf1 100644 --- a/src/websocket/upgrade.rs +++ b/src/websocket/upgrade.rs @@ -91,8 +91,12 @@ pub const DEFAULT_WS_MAX_SESSIONS: usize = 64; /// the adapter's `RouterState` (carrying the shared [`WsSessions`] /// instance and the configured session cap) or a bare /// `Arc` (custom upgrade routes / integration tests -/// get a handler-private registry and the default cap; eviction still -/// works in-crate, just not shared). +/// get a handler-private registry, a **per-request** session-cap +/// semaphore and the default knobs; eviction still works in-crate, +/// just not shared — and the per-request semaphore bounds nothing +/// across requests, review 007 WS-30: a route that needs an effective +/// shared cap inserts a [`SessionSlots`] extension; `FromRef` +/// extraction is per request with no caching). #[derive(Clone)] pub struct SessionState { registry: Arc, @@ -104,13 +108,19 @@ pub struct SessionState { idle_timeout: Option, /// The openable-ALPN set (WS-22): `None` declares no openables. openable_alpns: Option>, + /// The `op/register` surface's `AccessControl` (review 007 WS-29). + op_register_acl: alkcall::registry::spec::AccessControl, } impl SessionState { /// From the plain registry state (custom upgrade routes / /// integration tests): sessions default to a fresh [`WsSessions`] /// private to the handler (eviction still functional in-crate but - /// not shared with the assembly layer) and the default session cap. + /// not shared with the assembly layer) and a **per-request** + /// default session-cap semaphore — built by `FromRef` per request + /// (no caching), so it bounds nothing across requests (review 007 + /// WS-30); a route needing an effective shared cap inserts a + /// [`SessionSlots`] extension. pub(crate) fn from_registry(registry: &Arc) -> Self { Self { registry: Arc::clone(registry), @@ -118,19 +128,22 @@ impl SessionState { session_slots: Arc::new(tokio::sync::Semaphore::new(DEFAULT_WS_MAX_SESSIONS)), idle_timeout: Some(crate::websocket::DEFAULT_WS_IDLE_TIMEOUT), openable_alpns: None, + op_register_acl: alkcall::registry::spec::AccessControl::default(), } } /// From the adapter's router state: the shared [`WsSessions`] /// instance the assembly layer can hold for eviction, the /// pre-built session-cap semaphore (one per `HttpAdapter`, shared - /// across requests), and the deployment's openable-ALPN set. + /// across requests), the deployment's openable-ALPN set, and the + /// `op/register` ACL. pub(crate) fn new( registry: Arc, sessions: Arc, session_slots: Arc, idle_timeout: Option, openable_alpns: Option>, + op_register_acl: alkcall::registry::spec::AccessControl, ) -> Self { Self { registry, @@ -138,6 +151,7 @@ impl SessionState { session_slots, idle_timeout, openable_alpns, + op_register_acl, } } @@ -149,6 +163,10 @@ impl SessionState { &self.sessions } + pub(crate) fn session_slots(&self) -> &Arc { + &self.session_slots + } + pub(crate) fn idle_timeout(&self) -> Option { self.idle_timeout } @@ -156,6 +174,10 @@ impl SessionState { pub(crate) fn openable_alpns(&self) -> Option> { self.openable_alpns.clone() } + + pub(crate) fn op_register_acl(&self) -> alkcall::registry::spec::AccessControl { + self.op_register_acl.clone() + } } /// `FromRef` chain: a bare `Arc` router state lifts @@ -267,6 +289,10 @@ impl OpenableAlpn { /// `None` = the crate default window, `Some(d)` a deployment-set /// window. `openable_alpns` (WS-22) is the deployment's openable-ALPN /// set registered per session (see [`install_channel_zero`]). +/// `op_register_acl` (review 007 WS-29) is the `op/register` +/// surface's `AccessControl` — `AccessControl::default()` (the +/// permissive crate default) unless a deployment restricts which +/// authenticated peers may announce ops. #[allow(clippy::too_many_arguments)] pub async fn run_channels_session( socket: axum::extract::ws::WebSocket, @@ -277,6 +303,7 @@ pub async fn run_channels_session( idle_timeout: Option, write_timeout: Option, openable_alpns: Option>, + op_register_acl: alkcall::registry::spec::AccessControl, ) { let (byte_stream, pumps) = split_ws_to_bytes_idle_with_write(socket, idle_timeout, write_timeout); @@ -294,7 +321,13 @@ pub async fn run_channels_session( let _ = conn.set_identity(identity.clone()); let adapter = ChannelsAdapter::new( - install_channel_zero(registry, sessions, Arc::clone(&policy), openable_alpns), + install_channel_zero( + registry, + sessions, + Arc::clone(&policy), + openable_alpns, + op_register_acl, + ), policy, ); let auth = AuthContext { @@ -359,6 +392,16 @@ impl Drop for ConnectionGuard { /// overlay and can hub→session-call announced or imported ops. The /// handle is removed when the channel-0 task ends (any teardown path). /// +/// `op_register_acl` (review 007 WS-29) is the `AccessControl` the +/// per-session `op/register` op is registered with — the announce +/// surface's gate. The default is `AccessControl::default()` (the +/// SRV-10 permissive-crate-default precedent, review-006 UP-02's +/// recorded posture); a deployment restricting which authenticated +/// peers may announce ops threads a stricter value — +/// `HttpAdapter::with_ws_op_register_acl` on the built-in surface or +/// the [`OpRegisterAcl`] request extension on bare-registry/custom +/// upgrade routes (both mirroring the openables/policy threading). +/// /// The dispatcher's token resolver is a no-op: the WS identity is /// attached to the connection at upgrade time and /// `Dispatcher::resolve_identity` falls back to it when the payload @@ -369,12 +412,14 @@ fn install_channel_zero( sessions: Option, policy: Arc, openable_alpns: Option>, + op_register_acl: alkcall::registry::spec::AccessControl, ) -> alkcall::channels::adapter::InstallChannelZero { Arc::new(move |manager, channel0_conn, auth| { let registry = Arc::clone(®istry); let sessions = sessions.clone(); let policy = Arc::clone(&policy); let openable_alpns = openable_alpns.clone(); + let op_register_acl = op_register_acl.clone(); tokio::spawn(async move { // The WS identity rides the upgrade request; propagate it to // channel 0's `CallConnection` so the dispatcher's @@ -427,9 +472,7 @@ fn install_channel_zero( } alkcall::registry::discovery::install_bootstrap_discovery(&fork)?; fork.register(alkcall::registry::registration::HandlerRegistration::new( - alkcall::registry::op_register::op_register_spec( - alkcall::registry::spec::AccessControl::default(), - ), + alkcall::registry::op_register::op_register_spec(op_register_acl), alkcall::registry::registration::HandlerKind::Once( alkcall::registry::op_register::op_register_handler( Arc::clone(&call_connection), @@ -477,7 +520,13 @@ fn install_channel_zero( pub(crate) fn adapter_install_channel_zero( registry: Arc, ) -> alkcall::channels::adapter::InstallChannelZero { - install_channel_zero(registry, None, Arc::new(NoCap), None) + install_channel_zero( + registry, + None, + Arc::new(NoCap), + None, + alkcall::registry::spec::AccessControl::default(), + ) } struct NoopProvider; @@ -516,8 +565,9 @@ pub struct ChannelsPolicy(pub Arc); /// [`crate::websocket::DEFAULT_WS_WRITE_TIMEOUT`] for the write side, /// which is also what bare-registry routes (custom upgrade routes on /// a plain `Arc` state) get: 60 s idle + 60 s -/// write windows, 64-session semaphore, handler-private -/// [`WsSessions`]. +/// write windows, a handler-private [`WsSessions`], and a per-request +/// (no effective) session cap — see [`SessionSlots`] for the +/// shared-cap surface. #[derive(Clone, Copy, Debug)] pub struct WsTimeouts { /// Idle-read window (WS-01); `None` disables the eviction. @@ -537,6 +587,39 @@ pub struct WsTimeouts { #[derive(Clone)] pub struct OpenableAlpns(pub Arc<[OpenableAlpn]>); +/// Per-request `op/register` ACL (review 007 WS-29): a deployment +/// inserts `OpRegisterAcl` into the request extensions (a route layer +/// on the WS route, mirroring [`ChannelsPolicy`] / [`WsTimeouts`] / +/// [`OpenableAlpns`]) to gate the announce surface per route — the +/// per-session `op/register` op is registered with this +/// `AccessControl`, so a peer whose identity does not satisfy it gets +/// `FORBIDDEN` on the announce. Without the extension the +/// [`SessionState`] value applies — +/// [`HttpAdapter::with_ws_op_register_acl`](crate::server::HttpAdapter::with_ws_op_register_acl) +/// for the built-in surface; the default everywhere is +/// `AccessControl::default()` (any authenticated peer may announce). +#[derive(Clone, Debug)] +pub struct OpRegisterAcl(pub alkcall::registry::spec::AccessControl); + +/// Per-request session-cap slots (review 007 WS-30): a deployment +/// inserts `SessionSlots` into the request extensions (a route layer on +/// the WS route, mirroring [`ChannelsPolicy`] / [`WsTimeouts`] / +/// [`OpenableAlpns`] / [`OpRegisterAcl`]) to bound the route's +/// concurrent WS sessions with a shared `Arc` — one permit +/// per upgrade, held for the session's lifetime; over-cap callers are +/// rejected with 503 (WS-09). Without the extension the +/// [`SessionState`] value applies — the built-in surface's +/// adapter-wide semaphore +/// (`HttpAdapter::with_ws_max_sessions`, one per `HttpAdapter`, +/// shared across requests). A bare-registry route's state is built per +/// request (`FromRef`, no caching), so its semaphore bounds nothing +/// across requests; such a route inserts this extension to get an +/// effective shared cap. Wrap the semaphore in +/// `SessionSlots(Arc::new(tokio::sync::Semaphore::new(n)))` at the +/// route layer. +#[derive(Clone)] +pub struct SessionSlots(pub Arc); + /// The upgrade handler. Requires the resolved identity in request /// extensions (stashed by [`ws_bearer_auth`]) — a WS session without /// an identity cannot run `AccessControl::check`. @@ -563,12 +646,23 @@ pub struct OpenableAlpns(pub Arc<[OpenableAlpn]>); /// (`HttpAdapter::with_ws_openable_alpns`); the default is no /// openables (channel 0 only). /// +/// The `op/register` ACL (review 007 WS-29) comes from the +/// [`OpRegisterAcl`] request extension when present, else from the +/// router state (`HttpAdapter::with_ws_op_register_acl`); the default +/// is `AccessControl::default()` (any authenticated peer may announce). +/// /// Session cap (WS-09): one semaphore permit is acquired per upgrade, /// post-auth and pre-upgrade; when the configured cap /// (`HttpAdapter::with_ws_max_sessions`, default /// [`DEFAULT_WS_MAX_SESSIONS`]) is exhausted the upgrade is rejected /// with **503 Service Unavailable** — holding the permit for the -/// session's lifetime, so ended sessions free their slot. +/// session's lifetime, so ended sessions free their slot. The permit +/// source is the [`SessionSlots`] request extension when present, else +/// the router state's semaphore; a bare-registry route's state is +/// built per request, so its semaphore bounds nothing across requests +/// (review 007 WS-30) — a route that needs an effective shared cap +/// inserts the `SessionSlots` extension. +#[allow(clippy::too_many_arguments)] pub async fn ws_upgrade_handler( sessions: Option>, axum::extract::State(state): axum::extract::State, @@ -576,9 +670,14 @@ pub async fn ws_upgrade_handler( policy: Option>, timeouts: Option>, openables: Option>, + op_register_acl: Option>, + session_slots: Option>, ws_upgrade: WebSocketUpgrade, ) -> Response { - let Ok(permit) = state.session_slots.clone().try_acquire_owned() else { + let session_slots = session_slots + .map(|axum::Extension(s)| s.0) + .unwrap_or_else(|| Arc::clone(state.session_slots())); + let Ok(permit) = session_slots.try_acquire_owned() else { return ( StatusCode::SERVICE_UNAVAILABLE, "503 Service Unavailable: WS session cap reached", @@ -602,6 +701,9 @@ pub async fn ws_upgrade_handler( let openable_alpns = openables .map(|axum::Extension(o)| o.0) .or_else(|| state.openable_alpns()); + let op_register_acl = op_register_acl + .map(|axum::Extension(a)| a.0) + .unwrap_or_else(|| state.op_register_acl()); ws_upgrade .max_frame_size(INBOUND_WS_FRAME_CAP) .max_message_size(INBOUND_WS_MESSAGE_CAP) @@ -616,6 +718,7 @@ pub async fn ws_upgrade_handler( idle_timeout, Some(write_timeout.unwrap_or(crate::websocket::DEFAULT_WS_WRITE_TIMEOUT)), openable_alpns, + op_register_acl, ) .await }) diff --git a/tests/ws_upgrade_session.rs b/tests/ws_upgrade_session.rs index ad19b0e..839d570 100644 --- a/tests/ws_upgrade_session.rs +++ b/tests/ws_upgrade_session.rs @@ -1602,3 +1602,263 @@ async fn too_large_data_channel_chunk_survives_demux_resync() { assert_eq!(echoed, payload, "demux resynced after the TooLarge skip"); ws.close().await; } + +// --- Review 007 remediation gates (WS-29/30/32) --- + +/// The announce frame the `op/register` gates send: one +/// `OpRegisterRequest` wrapped as a channel-0 chunk. +async fn announce_op(ws: &mut WsClient, request_id: &str, name: &str) -> EventEnvelope { + use alkcall::registry::op_register::OpRegisterRequest; + + let spec = OperationSpec::new( + name, + OperationType::Query, + Visibility::External, + serde_json::json!({}), + serde_json::json!({}), + vec![], + AccessControl::default(), + None, + ); + let request = OpRegisterRequest { + spec, + replace: false, + }; + let frame = EventEnvelope::requested( + request_id, + serde_json::json!({ + "operationId": "op/register", + "input": request.to_json(), + }), + ); + ws.send_binary(frame_channel0_chunk(&frame)).await; + await_envelope(ws, &frame.id).await +} + +/// Review 007 WS-29 acceptance (the builder path): a deployment's +/// `with_ws_op_register_acl` value is the `op/register` surface's ACL — +/// a peer whose identity lacks the required scope gets `FORBIDDEN` on +/// the announce, a peer with it announces cleanly. The permissive +/// default (any authenticated peer may announce) is pinned by the +/// Unit-3 gate above, whose registry route carries no override. +#[tokio::test] +async fn op_register_acl_builder_gates_the_announce_surface() { + use alkhttp::server::HttpAdapter; + + let registry = echo_registry(); + struct StaticTok; + impl IdentityProvider for StaticTok { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let s = String::from_utf8_lossy(&token.raw).to_string(); + match s.as_str() { + "tok-1" => Some(identity("alice", &[])), + "tok-2" => Some(identity("bobb", &["announcer"])), + _ => None, + } + } + } + + let adapter = HttpAdapter::new( + Arc::new(StaticTok) as Arc, + Arc::clone(®istry), + ) + .with_ws_op_register_acl(AccessControl { + required_scopes: vec!["announcer".to_string()], + ..Default::default() + }); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + let app: axum::Router = adapter.router().clone(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let url = format!("{addr}/alk/channels"); + + // Scope-less peer: the announce is denied before the handler runs. + let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap(); + let env = call_and_await(&mut ws, "req-deny", "echo/run", serde_json::json!({})).await; + assert_eq!(env.r#type, EVENT_RESPONDED, "session established"); + let env = announce_op(&mut ws, "req-announce-deny", "consumer/denied").await; + assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type); + assert_eq!( + env.payload["code"], "FORBIDDEN", + "ACL without the scope is denied: {}", + env.payload + ); + + // A peer with the scope announces cleanly. + let mut ws2 = WsClient::connect_authorized(&url, "tok-2").await.unwrap(); + let env = announce_op(&mut ws2, "req-announce-ok", "consumer/allowed").await; + assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type); + assert_eq!(env.payload["output"]["registered"], true); + ws.close().await; + ws2.close().await; +} + +/// Review 007 WS-29 acceptance (the extension path): the +/// `OpRegisterAcl` request extension on a bare-registry route carries +/// the same gate — `FORBIDDEN` without the scope, announce-ok with it. +#[tokio::test] +async fn op_register_acl_extension_gates_the_announce_surface() { + let app = axum::Router::new() + .route( + "/alk/channels", + axum::routing::get(alkhttp::websocket::ws_upgrade_handler), + ) + .layer(axum::middleware::from_fn_with_state( + provider_with(vec![ + ("tok-1", identity("alice", &[])), + ("tok-2", identity("bobb", &["announcer"])), + ]), + alkhttp::websocket::ws_bearer_auth, + )) + .layer(axum::Extension(alkhttp::websocket::OpRegisterAcl( + AccessControl { + required_scopes: vec!["announcer".to_string()], + ..Default::default() + }, + ))) + .with_state(echo_registry()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let url = format!("{addr}/alk/channels"); + + let mut ws = WsClient::connect_authorized(&url, "tok-1").await.unwrap(); + let env = announce_op(&mut ws, "req-announce-deny", "consumer/denied").await; + assert_eq!(env.r#type, EVENT_ERROR, "got {}", env.r#type); + assert_eq!( + env.payload["code"], "FORBIDDEN", + "extension ACL without the scope is denied: {}", + env.payload + ); + + let mut ws2 = WsClient::connect_authorized(&url, "tok-2").await.unwrap(); + let env = announce_op(&mut ws2, "req-announce-ok", "consumer/allowed").await; + assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type); + assert_eq!(env.payload["output"]["registered"], true); + ws.close().await; + ws2.close().await; +} + +/// Review 007 WS-30 acceptance: a bare-registry route's +/// `SessionState` semaphore is built per request (`FromRef`) and +/// bounds nothing; the `SessionSlots` extension carries the shared +/// semaphore — two sessions over a cap of 1 reject with 503, and a +/// slot frees on session end (the WS-09 shape on the extension path). +#[tokio::test] +async fn session_slots_extension_bounds_a_bare_registry_route() { + let slots = Arc::new(tokio::sync::Semaphore::new(1)); + let app = axum::Router::new() + .route( + "/alk/channels", + axum::routing::get(alkhttp::websocket::ws_upgrade_handler), + ) + .layer(axum::middleware::from_fn_with_state( + provider_with(vec![("tok-1", identity("alice", &[]))]), + alkhttp::websocket::ws_bearer_auth, + )) + .layer(axum::Extension(alkhttp::websocket::SessionSlots( + Arc::clone(&slots), + ))) + .with_state(echo_registry()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let url = format!("{addr}/alk/channels"); + + let mut first = WsClient::connect_authorized(&url, "tok-1").await.unwrap(); + let env = call_and_await(&mut first, "req-1", "echo/run", serde_json::json!({})).await; + assert_eq!(env.r#type, EVENT_RESPONDED, "first session admitted"); + + let status = WsClient::connect_status(&url, Some("tok-1")).await; + assert_eq!(status, Some(503), "second session rejected over the cap"); + + first.close().await; + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + let mut third = WsClient::connect_authorized(&url, "tok-1").await.unwrap(); + let env = call_and_await(&mut third, "req-2", "echo/run", serde_json::json!({})).await; + assert_eq!(env.r#type, EVENT_RESPONDED, "slot freed after session end"); + third.close().await; +} + +/// Review 007 WS-32 acceptance: the built-in openables threading +/// (`HttpAdapter::with_ws_openable_alpns` → `RouterState` → +/// `SessionState` → hook) serves the data-channel surface — a session +/// on the built-in router discovers the openable via `services/list`, +/// opens the channel, and round-trips bytes on it. Every Unit-3 gate +/// rides the `OpenableAlpns` request-extension fallback; this is the +/// only gate through the builder path a real `HttpAdapter` deployment +/// uses. +#[tokio::test] +async fn builder_path_openables_serve_the_data_channel_surface() { + use alkhttp::server::HttpAdapter; + + struct StaticTok; + impl IdentityProvider for StaticTok { + fn resolve_from_fingerprint(&self, _: &str) -> Option { + None + } + fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option { + let s = String::from_utf8_lossy(&token.raw).to_string(); + (s == "tok-1").then(|| identity("alice", &[])) + } + } + + let adapter = + HttpAdapter::new(Arc::new(StaticTok), echo_registry()).with_ws_openable_alpns(vec![ + alkhttp::websocket::OpenableAlpn::new(echo_open_spec(), echo_open_handler()), + ]); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = format!("ws://{}", listener.local_addr().unwrap()); + let app: axum::Router = adapter.router().clone(); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let mut ws = WsClient::connect_authorized(&format!("{addr}/alk/channels"), "tok-1") + .await + .unwrap(); + + let env = call_and_await(&mut ws, "req-list", "services/list", serde_json::json!({})).await; + assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type); + let names: Vec = env.payload["output"]["operations"] + .as_array() + .expect("operations array") + .iter() + .filter_map(|o| o["name"].as_str().map(String::from)) + .collect(); + assert!( + names.contains(&"channels/echo/sub".to_string()), + "builder-path openable discoverable: {names:?}" + ); + + let env = call_and_await( + &mut ws, + "req-open", + "channels/echo/sub", + serde_json::json!({}), + ) + .await; + assert_eq!(env.r#type, EVENT_RESPONDED, "got {}", env.r#type); + let channel_id = env.payload["output"]["channel_id"] + .as_u64() + .expect("channel_id"); + + let payload = b"builder-path-bytes"; + ws.send_binary(frame_data_channel(channel_id as u32, payload)) + .await; + let echoed = read_data_channel_block(&mut ws, channel_id as u32).await; + assert_eq!( + echoed, payload, + "data channel round trip on the built-in surface" + ); + ws.close().await; +}