feat(websocket): WS-29/30/32 — op/register ACL surface, SessionSlots, builder-path gate
Review 007 Unit 2 (the two "implement" decisions taken during remediation, plus the coverage gap): - WS-29: the `op/register` override surface review-006 UP-02 and ADR-048 recorded as landed is now implemented. The hook threads an `op_register_acl` `AccessControl` into `op_register_spec` (the permissive `AccessControl::default()` remains the default everywhere); the built-in surface sets it via `HttpAdapter::with_ws_op_register_acl`, bare-registry/custom routes via the `OpRegisterAcl` request extension (mirroring `ChannelsPolicy`/`WsTimeouts`/`OpenableAlpns`). A peer whose identity does not satisfy the ACL gets `FORBIDDEN` on the announce. Gates: builder path (`FORBIDDEN` scope-less / announce-ok scoped) + extension path. - WS-30: the bare-registry `SessionState` is built by `FromRef` per request, so its default-cap semaphore bounds nothing across requests (corrects review-002 WS-17's "bounded at 64 sessions" claim). New `SessionSlots` request extension carries the shared semaphore for routes that need an effective cap; the upgrade handler prefers it over the state value. Doc comments corrected (`SessionState`, `WsTimeouts`, `ws_upgrade_handler`). Gate: cap-1 route → 503 over cap → slot freed on session end. - WS-32: the built-in openables threading (`with_ws_openable_alpns` → `RouterState` → `SessionState` → hook) gets its first gate — every Unit-3 gate rode the `OpenableAlpns` extension fallback. `builder_path_openables_serve_the_data_channel_ surface` discovers the openable via `services/list`, opens the channel, and round-trips bytes through the builder-built router. Verification: cargo test 454 passed / 0 failed; cargo test --all-features 587 passed / 0 failed (+5 gates); clippy (both configs) clean; fmt clean. Review: docs/reviews/007-ws-data-channel-surface-review.md
This commit is contained in:
@@ -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<Identity> {
|
||||
None
|
||||
}
|
||||
fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
|
||||
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<dyn IdentityProvider>,
|
||||
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<Identity> {
|
||||
None
|
||||
}
|
||||
fn resolve_from_token(&self, token: &alkcall::core::auth::AuthToken) -> Option<Identity> {
|
||||
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<String> = 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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user