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

fix(adapters): consumer adapter hygiene (CON-01, CON-03..CON-13)

Browse Source 
- CON-01: from_mcp discovery follows tools/list pagination
  (rmcp list_all_tools); three-page paginating-server test
- CON-03: from_wss refuses ws:// with a Bearer token unless
  FromWss::allow_plaintext() is called explicitly (tests: refusal,
  opt-in, token-less passthrough)
- CON-04: audio variant of content_block_union_schema requires
  ["type","data","mimeType"]; jsonschema-validated audio block
- CON-05/07: import-time credential documented on both adapters;
  dead per-call capability read removed
- CON-06: 401 classification typed-first (downcast to rmcp
  StreamableHttpError<reqwest::Error>; AuthRequired/InsufficientScope/
  Client with status 401); a :40101 URL no longer misclassifies (tested)
- CON-11: transport tools/call failures declare MCP_TRANSPORT_ERROR;
  rmcp JSON-RPC errors preserve code (MCP_JRPC_<code>) and data
- CON-12: tool names validated at import (/, whitespace, empty →
  SchemaParse); unit + integration tests
- CON-13: tokens held as alkcall Secret<String> (zeroize, redacted Debug)
- CON-08/09: no close handles; explicit-limitation notes in from_mcp
  module docs, from_wss module docs, and ADR-070
- CON-10: full_surface [[test]] required-features = ["mcp","test-support"];
  cargo test --features mcp now compiles and passes

Verified: cargo test; cargo test --features mcp; cargo test --all-features;
cargo clippy (--all-features) --all-targets -- -D warnings; cargo fmt --check
This commit is contained in:
glm-5.3-flash
2026-08-29 10:54:33 +00:00
parent 314472012d
commit 8700ed0fea
7 changed files with 837 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapters/from_mcp/mod.rs
+147 -55
View File
@@ -4,13 +4,25 @@
//! projection; this is the import direction).
//!
//! Streamable HTTP only (ADR-037 — stdio is not built). Feature-gated behind
//! `mcp`. The forwarding handler reads the bearer token from
//! `OperationContext.capabilities` (ADR-014 no-env-vars), not `std::env::var`.
//! Provenance is `FromMCP` (leaf — `composition_authority: None`,
//! `scoped_env: None`, `Internal` by default — ADR-015/022).
//! `mcp`. **The credential is captured at import time:** the token given to
//! [`FromMCP::with_auth_token`] pins the rmcp transport's auth header for the
//! session's lifetime; per-call credentials handed to
//! `OperationContext.capabilities` are not observed by the forwarding
//! handler (the no-env-vars invariant, ADR-014, still holds — the token
//! comes from config, never `std::env::var`). Provenance is `FromMCP`
//! (leaf — `composition_authority: None`, `scoped_env: None`, `Internal`
//! by default — ADR-015/022).
//!
//! Session teardown (explicit limitation, review-001 CON-08): `import()`
//! detaches the rmcp client session fire-and-forget; there is no close
//! handle on [`FromMCP`], so each import leaves the remote server-side
//! session and its GET SSE stream open until the remote times it out.
//! Repeated imports accumulate sessions — reconnecting importers should
//! import once per process (or wait for a teardown handle in a later
//! version).
use alkcall::client::{AdapterError, OperationAdapter};
use alkcall::core::types::Capabilities;
use alkcall::core::types::{Capabilities, Secret};
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::context::OperationContext;
use alkcall::registry::registration::{
@@ -25,16 +37,22 @@ use rmcp::model::{
};
use rmcp::service::RoleClient;
use rmcp::transport::{
streamable_http_client::StreamableHttpClientTransportConfig, StreamableHttpClientTransport,
streamable_http_client::{StreamableHttpClientTransportConfig, StreamableHttpError},
DynamicTransportError, StreamableHttpClientTransport,
};
use rmcp::{Peer, ServiceExt};
use rmcp::{Peer, ServiceError, ServiceExt};
use serde_json::{Map, Value};
const MCP_CAPABILITY_KEY: &str = "mcp";
/// Declared error for transport-level `tools/call` failures (review-001
/// CON-11): the remote is unreachable, the transport closed, or the call
/// timed out. Payload is the array-of-content-blocks shape; retryable.
const MCP_TRANSPORT_ERROR: &str = "MCP_TRANSPORT_ERROR";
pub struct FromMCP {
endpoint: String,
auth_token: Option<String>,
auth_token: Option<Secret<String>>,
namespace: String,
}
@@ -48,7 +66,7 @@ impl FromMCP {
}
pub fn with_auth_token(mut self, token: impl Into<String>) -> Self {
self.auth_token = Some(token.into());
self.auth_token = Some(Secret::new(token.into()));
self
}
@@ -60,8 +78,8 @@ impl FromMCP {
&self.namespace
}
pub fn auth_token(&self) -> Option<&str> {
self.auth_token.as_deref()
pub fn auth_token(&self) -> Option<&Secret<String>> {
self.auth_token.as_ref()
}
}
@@ -70,7 +88,7 @@ impl OperationAdapter for FromMCP {
async fn import(&self) -> Result<Vec<HandlerRegistration>, AdapterError> {
let mut config = StreamableHttpClientTransportConfig::with_uri(self.endpoint.clone());
if let Some(token) = &self.auth_token {
config = config.auth_header(token.clone());
config = config.auth_header(token.expose_secret().clone());
}
let transport = StreamableHttpClientTransport::from_config(config);
let client_info = ClientInfo::new(
@@ -83,17 +101,17 @@ impl OperationAdapter for FromMCP {
.map_err(|e| classify_init_error(&e))?;
let peer: Peer<RoleClient> = running.peer().clone();
let tools = peer.list_tools(Default::default()).await.map_err(|e| {
AdapterError::DiscoveryFailed {
let tools = peer
.list_all_tools()
.await
.map_err(|e| AdapterError::DiscoveryFailed {
message: format!("tools/list failed: {e}"),
}
})?;
})?;
let bundles = tools
.tools
.into_iter()
.map(|tool| build_registration(&peer, &self.namespace, self.auth_token.clone(), tool))
.collect::<Vec<_>>();
.collect::<Result<Vec<_>, _>>()?;
std::mem::forget(running);
Ok(bundles)
@@ -104,15 +122,11 @@ fn classify_init_error(e: &rmcp::service::ClientInitializeError) -> AdapterError
use rmcp::service::ClientInitializeError as E;
match e {
E::TransportError { error, .. } => {
let msg = format!("{error:?}");
if msg.contains("401")
|| msg.contains("Unauthorized")
|| msg.contains("AuthRequired")
|| msg.contains("AuthRequired(")
{
AdapterError::Unauthorized { message: msg }
let message = error.to_string();
if is_unauthorized_transport(error) || auth_error_message(&message) {
AdapterError::Unauthorized { message }
} else {
AdapterError::DiscoveryFailed { message: msg }
AdapterError::DiscoveryFailed { message }
}
}
other => AdapterError::DiscoveryFailed {
@@ -121,13 +135,34 @@ fn classify_init_error(e: &rmcp::service::ClientInitializeError) -> AdapterError
}
}
fn is_unauthorized_transport(error: &DynamicTransportError) -> bool {
match error
.error
.downcast_ref::<StreamableHttpError<reqwest::Error>>()
{
Some(StreamableHttpError::AuthRequired(_))
| Some(StreamableHttpError::InsufficientScope(_)) => true,
Some(StreamableHttpError::Client(e)) => {
e.status() == Some(reqwest::StatusCode::UNAUTHORIZED)
}
_ => false,
}
}
fn auth_error_message(message: &str) -> bool {
message.contains("AuthRequired")
|| message.contains("InsufficientScope")
|| message.contains("www-authenticate")
|| message.to_ascii_lowercase().contains("unauthorized")
}
fn build_registration(
peer: &Peer<RoleClient>,
namespace: &str,
auth_token: Option<String>,
auth_token: Option<Secret<String>>,
tool: Tool,
) -> HandlerRegistration {
let spec = build_spec(&tool, namespace);
) -> Result<HandlerRegistration, AdapterError> {
let spec = build_spec(&tool, namespace)?;
let caps = capabilities_for(auth_token);
let tool_name = tool.name.to_string();
@@ -137,18 +172,12 @@ fn build_registration(
let tool_name = tool_name.clone();
async move {
let request_id = context.request_id.clone();
let _token_present = context
.capabilities
.get(MCP_CAPABILITY_KEY)
.map(|s| s.expose_secret().len());
let arguments = value_to_json_object(input);
let params = CallToolRequestParams::new(tool_name.clone()).with_arguments(arguments);
let result = match peer.call_tool(params).await {
Ok(r) => r,
Err(e) => {
let message = format!("tools/call failed: {e}");
return ResponseEnvelope::error(request_id, CallError::internal(message));
return ResponseEnvelope::error(request_id, transport_call_tool_error(&e))
}
};
@@ -156,23 +185,69 @@ fn build_registration(
}
});
HandlerRegistration::new(
Ok(HandlerRegistration::new(
spec,
HandlerKind::Once(handler),
OperationProvenance::FromMCP,
None,
None,
caps,
)
))
}
pub(crate) fn build_spec(tool: &Tool, namespace: &str) -> OperationSpec {
let tool_name = tool.name.to_string();
/// Transport-level `tools/call` failure (remote unreachable, timeout,
/// transport closed), mapped to the declared [`MCP_TRANSPORT_ERROR`]
/// schema — never the undeclared `INTERNAL` (review-001 CON-11).
fn transport_call_tool_error(error: &ServiceError) -> CallError {
let message = format!("tools/call failed: {error}");
match error {
ServiceError::McpError(e) => {
let code = format!("MCP_JRPC_{:+06}", e.code.0);
let mut err = CallError::new(code, e.message.to_string(), true);
if let Some(data) = &e.data {
err = err.with_details(data.clone());
}
err
}
ServiceError::Timeout { .. }
| ServiceError::TransportClosed
| ServiceError::Cancelled { .. }
| ServiceError::TransportSend(_)
| ServiceError::UnexpectedResponse => CallError::new(MCP_TRANSPORT_ERROR, message, true),
_ => CallError::new(MCP_TRANSPORT_ERROR, message, true),
}
}
fn sanitize_tool_name(tool_name: &str) -> Result<String, AdapterError> {
let name = tool_name.trim();
if name.is_empty() {
return Err(AdapterError::SchemaParse {
message: "MCP tool name is empty".to_string(),
});
}
if name.contains('/') {
return Err(AdapterError::SchemaParse {
message: format!(
"MCP tool name `{name}` contains `/` — the two-segment ns/op op-name convention \
(review-001 CON-12) requires flat tool names; refusing import"
),
});
}
if name.chars().any(|c| c.is_whitespace()) {
return Err(AdapterError::SchemaParse {
message: format!("MCP tool name `{name}` contains whitespace"),
});
}
Ok(name.to_string())
}
pub(crate) fn build_spec(tool: &Tool, namespace: &str) -> Result<OperationSpec, AdapterError> {
let tool_name = sanitize_tool_name(&tool.name)?;
let op_name = format!("{namespace}/{tool_name}");
let input_schema = json_object_to_value(tool.input_schema.as_ref().clone());
let output_schema = output_schema_for(tool);
let error_schemas = error_schemas_for(tool);
OperationSpec::new(
Ok(OperationSpec::new(
op_name,
OperationType::Mutation,
Visibility::Internal,
@@ -181,7 +256,7 @@ pub(crate) fn build_spec(tool: &Tool, namespace: &str) -> OperationSpec {
error_schemas,
AccessControl::default(),
None,
)
))
}
pub(crate) fn map_call_tool_result(result: CallToolResult, request_id: String) -> ResponseEnvelope {
@@ -245,7 +320,7 @@ pub(crate) fn content_block_union_schema() -> Value {
"data": { "type": "string" },
"mimeType": { "type": "string" }
},
"required": ["type", "audio", "mimeType"]
"required": ["type", "data", "mimeType"]
},
{
"type": "object",
@@ -278,21 +353,38 @@ pub(crate) fn content_blocks_to_value(blocks: &[Content]) -> Value {
}
fn error_schemas_for(tool: &Tool) -> Vec<ErrorDefinition> {
vec![ErrorDefinition {
code: "MCP_TOOL_ERROR".to_string(),
description: format!("MCP tool '{}' reported an error (isError)", tool.name),
schema: serde_json::json!({
"type": "array",
"description": "MCP error content blocks",
"items": content_block_union_schema()
}),
http_status: None,
}]
vec![
ErrorDefinition {
code: "MCP_TOOL_ERROR".to_string(),
description: format!("MCP tool '{}' reported an error (isError)", tool.name),
schema: serde_json::json!({
"type": "array",
"description": "MCP error content blocks",
"items": content_block_union_schema()
}),
http_status: None,
},
ErrorDefinition {
code: MCP_TRANSPORT_ERROR.to_string(),
description: format!(
"the transport failed while calling MCP tool '{}' (remote unreachable, \
connection closed, or call timed out); retryable",
tool.name
),
schema: serde_json::json!({
"type": "null",
"description": "transport failures carry no payload"
}),
http_status: None,
},
]
}
fn capabilities_for(auth_token: Option<String>) -> Capabilities {
fn capabilities_for(auth_token: Option<Secret<String>>) -> Capabilities {
match auth_token {
Some(token) => Capabilities::new().with_http_token(MCP_CAPABILITY_KEY, token),
Some(token) => {
Capabilities::new().with_http_token(MCP_CAPABILITY_KEY, token.expose_secret().clone())
}
None => Capabilities::new(),
}
}
src/adapters/from_mcp/tests.rs
+205 -8
View File
@@ -1,6 +1,7 @@
use super::*;
use alkcall::registry::spec::Visibility;
use rmcp::model::{CallToolResult, Content, Tool};
use rmcp::transport::streamable_http_client::{AuthRequiredError, InsufficientScopeError};
fn make_tool(name: &str, input: Value, output: Option<Value>) -> Tool {
let input_map = match input {
@@ -40,10 +41,58 @@ fn struct_holds_endpoint_auth_token_namespace() {
let adapter = FromMCP::new("http://localhost:8000/mcp", "weather");
assert_eq!(adapter.endpoint(), "http://localhost:8000/mcp");
assert_eq!(adapter.namespace(), "weather");
assert_eq!(adapter.auth_token(), None);
assert!(adapter.auth_token().is_none());
let with_token = adapter.with_auth_token("sekrit");
assert_eq!(with_token.auth_token(), Some("sekrit"));
assert_eq!(
with_token.auth_token().map(|s| s.expose_secret().as_str()),
Some("sekrit")
);
}
#[test]
fn token_debug_output_is_redacted() {
let adapter = FromMCP::new("http://localhost:8000/mcp", "ns").with_auth_token("sekrit");
let debug = format!("{:?}", adapter.auth_token().expect("token present"));
assert_eq!(debug, "[REDACTED]");
assert!(!debug.contains("sekrit"));
}
#[test]
fn sanitize_tool_name_rejects_slash_and_whitespace() {
let err = sanitize_tool_name("a/b")
.map(|_| ())
.map_err(|e| e.to_string());
assert!(
err.unwrap_err().contains('/'),
"slash tool name must be refused (two-segment convention)"
);
assert!(sanitize_tool_name("a b").is_err());
assert!(sanitize_tool_name("").is_err());
assert_eq!(sanitize_tool_name(" padded ").unwrap(), "padded");
assert_eq!(sanitize_tool_name("get_weather").unwrap(), "get_weather");
assert_eq!(sanitize_tool_name("ns:op.v2").unwrap(), "ns:op.v2");
}
#[test]
fn build_spec_rejects_tool_name_with_slash() {
let tool = make_tool("bad/name", serde_json::json!({}), None);
match build_spec(&tool, "ns") {
Ok(_) => panic!("expected Err for tool name containing `/`"),
Err(alkcall::client::AdapterError::SchemaParse { message }) => {
assert!(message.contains('/'));
}
Err(other) => panic!("expected SchemaParse, got {other}"),
}
}
#[test]
fn build_spec_trims_and_keeps_flat_name() {
let tool = make_tool(" get_weather ", serde_json::json!({}), None);
let spec = build_spec(&tool, "ns").expect("flat name accepted");
assert_eq!(spec.name, "ns/get_weather");
assert_eq!(spec.namespace, "ns");
assert_eq!(spec.name.split('/').count(), 2);
}
#[test]
@@ -80,6 +129,36 @@ fn content_block_union_schema_has_all_five_variants() {
assert!(variants.contains(&"resource_link"));
}
#[test]
fn content_block_union_schema_audio_variant_satisfied_by_valid_block() {
let schema = content_block_union_schema();
let validator = jsonschema::options()
.build(&schema)
.expect("schema compiles");
let audio_result = serde_json::json!([
{ "type": "audio", "data": "QkFTRTY0", "mimeType": "audio/wav" }
]);
assert!(
validator.is_valid(&audio_result),
"valid audio block array must satisfy the union (CON-04); errors: {:?}",
validator
.iter_errors(&audio_result)
.map(|e| e.to_string())
.collect::<Vec<_>>()
);
assert!(
validator.is_valid(&serde_json::json!([{ "type": "text", "text": "hi" }])),
"text variant still valid"
);
let bogus = serde_json::json!([{ "type": "audio", "data": "x" }]);
assert!(!validator.is_valid(&bogus), "missing mimeType must fail");
let pre_fix = serde_json::json!([{ "type": "audio", "audio": "x", "mimeType": "audio/wav" }]);
assert!(
!validator.is_valid(&pre_fix),
"the pre-fix `audio` property shape must still fail (the old bug)"
);
}
#[test]
fn map_structured_content_present_used_as_result() {
let result = CallToolResult::structured(serde_json::json!({ "temperature": 22.5 }));
@@ -183,16 +262,18 @@ fn map_structured_content_preferred_over_content_blocks() {
fn error_schemas_for_tool_yields_mcp_tool_error() {
let tool = make_tool("weather", serde_json::json!({}), None);
let errors = error_schemas_for(&tool);
assert_eq!(errors.len(), 1);
assert_eq!(errors.len(), 2);
assert_eq!(errors[0].code, "MCP_TOOL_ERROR");
assert!(errors[0].description.contains("weather"));
assert!(errors[0].description.contains("isError"));
assert!(errors[0].schema["type"] == "array");
assert_eq!(errors[1].code, "MCP_TRANSPORT_ERROR");
assert!(errors[1].description.contains("transport"));
}
#[test]
fn capabilities_for_token_injects_http_token() {
let caps = capabilities_for(Some("tok-123".to_string()));
let caps = capabilities_for(Some(Secret::new("tok-123".to_string())));
let secret = caps.get(MCP_CAPABILITY_KEY).expect("token present");
assert_eq!(secret.expose_secret(), "tok-123");
}
@@ -212,7 +293,7 @@ fn build_spec_output_schema_present_shape() {
serde_json::json!({ "type": "object", "properties": { "temperature": { "type": "number" } } }),
),
);
let spec = build_spec(&tool, "weather");
let spec = build_spec(&tool, "weather").expect("flat tool name");
assert_eq!(spec.name, "weather/get_weather");
assert_eq!(spec.namespace, "weather");
assert_eq!(spec.op_type, OperationType::Mutation);
@@ -224,22 +305,23 @@ fn build_spec_output_schema_present_shape() {
spec.output_schema["properties"]["temperature"]["type"],
"number"
);
assert_eq!(spec.error_schemas.len(), 1);
assert_eq!(spec.error_schemas.len(), 2);
assert_eq!(spec.error_schemas[0].code, "MCP_TOOL_ERROR");
assert_eq!(spec.error_schemas[1].code, "MCP_TRANSPORT_ERROR");
assert!(spec.access_control == AccessControl::default());
}
#[test]
fn build_spec_output_schema_absent_uses_union() {
let tool = make_tool("legacy", serde_json::json!({}), None);
let spec = build_spec(&tool, "legacy");
let spec = build_spec(&tool, "legacy").expect("flat tool name");
assert_eq!(spec.output_schema, content_block_union_schema());
}
#[test]
fn build_spec_name_with_prefix_when_namespace_set() {
let tool = make_tool("search", serde_json::json!({}), None);
let spec = build_spec(&tool, "tools");
let spec = build_spec(&tool, "tools").expect("flat tool name");
assert_eq!(spec.name, "tools/search");
assert_eq!(spec.namespace, "tools");
}
@@ -255,3 +337,118 @@ async fn forwarding_handler_reads_capabilities_not_env_vars() {
let _ = adapter.auth_token();
assert!(adapter.auth_token().is_none());
}
#[test]
fn auth_error_message_ignores_url_port_4010() {
assert!(!auth_error_message(
"error sending request for url (http://host:4010/mcp)"
));
assert!(!auth_error_message("connection refused"));
assert!(!auth_error_message("HTTP 500: boom"));
}
#[test]
fn auth_error_message_matches_auth_wording_not_digits() {
assert!(auth_error_message(
"HTTP 401 Unauthorized: invalid bearer token (www-authenticate: Bearer)"
));
assert!(auth_error_message(
"AuthRequired(www_authenticate_header: Bearer realm=\"x\")"
));
assert!(auth_error_message(
"InsufficientScope(required_scope: admin)"
));
assert!(!auth_error_message(
"error sending request for url (http://127.0.0.1:40101/mcp)"
));
}
fn dyn_transport_error(error: StreamableHttpError<reqwest::Error>) -> DynamicTransportError {
DynamicTransportError::from_parts(
"streamable_http_client",
std::any::TypeId::of::<StreamableHttpClientTransport<reqwest::Client>>(),
Box::new(error),
)
}
#[test]
fn unauthorized_transport_matched_by_typed_variants() {
let auth_required = StreamableHttpError::<reqwest::Error>::AuthRequired(
AuthRequiredError::new("Bearer realm=\"test\"".to_string()),
);
assert!(
is_unauthorized_transport(&dyn_transport_error(auth_required)),
"rmcp's typed AuthRequired must classify as Unauthorized"
);
let insufficient_scope =
StreamableHttpError::<reqwest::Error>::InsufficientScope(InsufficientScopeError::new(
"Bearer scope=\"admin\"".to_string(),
Some("admin".to_string()),
));
assert!(
is_unauthorized_transport(&dyn_transport_error(insufficient_scope)),
"rmcp's typed InsufficientScope must classify as Unauthorized"
);
}
#[tokio::test]
async fn unauthorized_transport_matched_on_real_401_status() {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let Ok((mut sock, _)) = listener.accept().await else {
break;
};
tokio::spawn(async move {
let mut buf = [0u8; 4096];
let _ = tokio::io::AsyncReadExt::read(&mut sock, &mut buf).await;
let _ = tokio::io::AsyncWriteExt::write_all(
&mut sock,
b"HTTP/1.1 401 Unauthorized\r\ncontent-length: 0\r\n\r\n".as_slice(),
)
.await;
});
}
});
let resp = reqwest::get(format!("http://{addr}/mcp")).await.unwrap();
let status_error = resp.error_for_status().expect_err("401 -> status error");
assert_eq!(
status_error.status(),
Some(reqwest::StatusCode::UNAUTHORIZED)
);
let err = StreamableHttpError::<reqwest::Error>::Client(status_error);
assert!(
is_unauthorized_transport(&dyn_transport_error(err)),
"a real 401 (typed reqwest status error) must classify as Unauthorized"
);
}
#[tokio::test]
async fn url_with_port_401xx_is_not_unauthorized() {
let dead_url = "http://127.0.0.1:40101/mcp";
let err = reqwest::Client::new()
.get(dead_url)
.timeout(std::time::Duration::from_millis(500))
.send()
.await
.expect_err("dead endpoint -> connect error");
assert!(err.status().is_none(), "connect error carries no status");
let message = err.to_string();
assert!(
message.contains("401"),
"precondition: the URL's :40101 must appear in the message, got: {message}"
);
let transport_err = StreamableHttpError::<reqwest::Error>::Client(err);
assert!(
!is_unauthorized_transport(&dyn_transport_error(transport_err)),
"a connect error whose URL contains :40101 must NOT classify as Unauthorized (CON-06)"
);
assert!(
!auth_error_message(&message),
"the string fallback must not resurface the :40101 misclassification"
);
}
src/adapters/from_wss.rs
+152 -21
View File
@@ -3,15 +3,24 @@
//! import the remote node's operations as forwarding handlers — the
//! same-protocol importer (`from_call` pattern), with WSS as the transport.
//!
//! Feature-gated behind `wss` (tokio-tungstenite). The dial carries
//! `Authorization: Bearer <token>` when constructed with an auth token; at
//! call time the imported handlers read the per-call credential from
//! `OperationContext.capabilities` (the no-env-vars path, ADR-014), never
//! from `std::env::var`. Provenance is `FromCall` (leaf,
//! Feature-gated behind `wss` (tokio-tungstenite). **The credential is
//! captured at dial time:** the token given to
//! [`FromWss::with_auth_token`] rides the one `Authorization: Bearer`
//! header on the WS upgrade request; the imported handlers carry no
//! capabilities at all, so per-call credentials handed to
//! `OperationContext.capabilities` are never observed (the no-env-vars
//! invariant, ADR-014, still holds — the token comes from config, never
//! `std::env::var`). Provenance is `FromCall` (leaf,
//! `composition_authority: None`, `scoped_env: None`, `Internal` by
//! default — ADR-015/022), because the imported session IS the call
//! protocol.
//!
//! Plaintext `ws://` is refused by default: a dial that would carry a
//! Bearer token over an unencrypted connection is rejected at
//! [`WssSession::connect`] with a clear error unless
//! [`FromWss::allow_plaintext`] was called explicitly (review-001
//! CON-03).
//!
//! Import flow (ADR-070): dial WSS → adapt the tungstenite stream with the
//! shared WS↔byte-stream seam
//! ([`crate::websocket::split_tungstenite_to_bytes`]) →
@@ -36,6 +45,15 @@
//! (CON-02). Subsequent handler calls fail on write; reconnect policy is
//! the assembly layer's job.
//!
//! Session teardown (explicit limitation, review-001 CON-09): `import()`
//! detaches the session fire-and-forget — there is no close/shutdown
//! handle on [`FromWss`], and a reconnecting assembly layer that calls
//! `import()` again stacks a second full session over the first
//! (duplicate op names, the original session untorn-down). v1 accepts
//! this: import once per process, or tear the whole registry down when
//! the session drops. A teardown handle is future work (ADR-070 §Not in
//! scope — no reconnect layer in v1).
//!
//! [ADR-070]: https://docs.rs/alkhttp (docs/architecture/decisions)
use std::sync::Arc;
@@ -44,7 +62,7 @@ use alkcall::channels::client::ChannelClient;
use alkcall::client::{
from_call as import_from_call, AdapterError, FromCallConfig, OperationAdapter,
};
use alkcall::core::types::Connection;
use alkcall::core::types::{Connection, Secret};
use alkcall::protocol::connection::CallConnection;
use alkcall::protocol::wire::CallError;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
@@ -62,8 +80,9 @@ fn connection_closed_error() -> CallError {
pub struct FromWss {
endpoint: String,
auth_token: Option<String>,
auth_token: Option<Secret<String>>,
namespace: Option<String>,
allow_plaintext: bool,
}
impl FromWss {
@@ -72,11 +91,12 @@ impl FromWss {
endpoint: endpoint.into(),
auth_token: None,
namespace: None,
allow_plaintext: false,
}
}
pub fn with_auth_token(mut self, token: impl Into<String>) -> Self {
self.auth_token = Some(token.into());
self.auth_token = Some(Secret::new(token.into()));
self
}
@@ -85,6 +105,15 @@ impl FromWss {
self
}
/// Explicitly allow dialing a plaintext `ws://` endpoint. Without
/// this, a `ws://` endpoint is refused when a token is present
/// (review-001 CON-03): a Bearer credential must not ride an
/// unencrypted connection without an explicit opt-in.
pub fn allow_plaintext(mut self) -> Self {
self.allow_plaintext = true;
self
}
pub fn endpoint(&self) -> &str {
&self.endpoint
}
@@ -93,8 +122,8 @@ impl FromWss {
self.namespace.as_deref()
}
pub fn auth_token(&self) -> Option<&str> {
self.auth_token.as_deref()
pub fn auth_token(&self) -> Option<&Secret<String>> {
self.auth_token.as_ref()
}
}
@@ -128,7 +157,30 @@ impl WssSession {
/// Exposed for the assembly layer and tests: hold the session for as
/// long as imported ops should stay callable; on drop, in-flight
/// calls fail retryable (no hang until the 30s sweeper deadline).
pub async fn connect(endpoint: &str, auth_token: Option<&str>) -> Result<Self, AdapterError> {
///
/// A plaintext `ws://` endpoint is refused when `auth_token` is
/// present unless `allow_plaintext` is set — a Bearer token must not
/// ride an unencrypted connection without an explicit opt-in (CON-03).
pub async fn connect(
endpoint: &str,
auth_token: Option<&str>,
allow_plaintext: bool,
) -> Result<Self, AdapterError> {
if let Some(_token) = auth_token {
if !allow_plaintext {
let scheme = endpoint.split("://").next().unwrap_or_default();
if scheme.eq_ignore_ascii_case("ws") {
return Err(AdapterError::Transport {
message: format!(
"refusing plaintext `ws://` endpoint `{endpoint}` with a Bearer token: \
the credential would ride an unencrypted connection (CON-03); \
use `wss://` or call `FromWss::allow_plaintext` explicitly"
),
});
}
}
}
let mut request = endpoint
.into_client_request()
.map_err(|e| AdapterError::Transport {
@@ -214,7 +266,12 @@ impl OperationAdapter for FromWss {
async fn import(
&self,
) -> Result<Vec<alkcall::registry::registration::HandlerRegistration>, AdapterError> {
let session = WssSession::connect(&self.endpoint, self.auth_token.as_deref()).await?;
let session = WssSession::connect(
&self.endpoint,
self.auth_token.as_ref().map(|s| s.expose_secret().as_str()),
self.allow_plaintext,
)
.await?;
let config = match &self.namespace {
Some(ns) => FromCallConfig::new().with_namespace_prefix(ns),
None => FromCallConfig::new(),
@@ -565,11 +622,78 @@ mod tests {
let adapter = FromWss::new("ws://localhost:9000/alk/channels");
assert_eq!(adapter.endpoint(), "ws://localhost:9000/alk/channels");
assert_eq!(adapter.namespace(), None);
assert_eq!(adapter.auth_token(), None);
assert!(adapter.auth_token().is_none());
let with_all = adapter.with_auth_token("tok").with_namespace("remote");
assert_eq!(with_all.auth_token(), Some("tok"));
assert_eq!(
with_all.auth_token().map(|s| s.expose_secret().as_str()),
Some("tok")
);
assert_eq!(with_all.namespace(), Some("remote"));
assert!(!with_all.allow_plaintext, "ws:// refused by default");
let opt_in = with_all.allow_plaintext();
assert!(opt_in.allow_plaintext);
}
#[test]
fn token_debug_output_is_redacted() {
let adapter = FromWss::new("wss://localhost/alk/channels").with_auth_token("sekrit");
let debug = format!("{:?}", adapter.auth_token().expect("token present"));
assert_eq!(debug, "[REDACTED]");
assert!(!debug.contains("sekrit"));
}
#[tokio::test]
async fn plaintext_ws_with_token_refused_by_default() {
let endpoint = spawn_producer(
producer_registry(),
provider_with(vec![("tok-1", identity("alice", &[]))]),
)
.await;
let adapter = FromWss::new(&endpoint).with_auth_token("tok-1");
match adapter.import().await {
Ok(_) => panic!("ws:// + token must be refused without an explicit opt-in (CON-03)"),
Err(AdapterError::Transport { message }) => {
assert!(
message.contains("CON-03"),
"error explains the refusal: {message}"
);
assert!(message.contains("ws://"));
}
Err(other) => panic!("expected Transport error, got {other}"),
}
}
#[tokio::test]
async fn plaintext_ws_with_token_allowed_when_explicitly_enabled() {
let endpoint = spawn_producer(
producer_registry(),
provider_with(vec![("tok-1", identity("alice", &["user", "admin"]))]),
)
.await;
let adapter = FromWss::new(&endpoint)
.with_auth_token("tok-1")
.allow_plaintext();
let bundles = adapter.import().await.expect("explicit opt-in dials ws://");
assert!(!bundles.is_empty());
}
#[tokio::test]
async fn plaintext_ws_without_token_is_not_refused_by_the_adapter() {
let endpoint = spawn_producer(producer_registry(), provider_with(vec![])).await;
let adapter = FromWss::new(&endpoint);
match adapter.import().await {
Ok(bundles) => assert!(!bundles.is_empty()),
Err(AdapterError::Transport { message }) => {
assert!(
!message.contains("CON-03"),
"without a token the adapter must not refuse ws://, got: {message}"
);
}
Err(other) => panic!("expected Transport error, got {other}"),
}
}
#[tokio::test]
@@ -580,7 +704,9 @@ mod tests {
)
.await;
let adapter = FromWss::new(&endpoint).with_auth_token("tok-1");
let adapter = FromWss::new(&endpoint)
.with_auth_token("tok-1")
.allow_plaintext();
let bundles = adapter.import().await.expect("import succeeds");
let mut names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect();
names.sort();
@@ -602,7 +728,9 @@ mod tests {
)
.await;
let adapter = FromWss::new(&endpoint).with_auth_token("tok-1");
let adapter = FromWss::new(&endpoint)
.with_auth_token("tok-1")
.allow_plaintext();
let bundles = adapter.import().await.expect("import succeeds");
let echo = bundles
.into_iter()
@@ -630,7 +758,9 @@ mod tests {
)
.await;
let adapter = FromWss::new(&endpoint).with_auth_token("tok-alice");
let adapter = FromWss::new(&endpoint)
.with_auth_token("tok-alice")
.allow_plaintext();
let bundles = adapter.import().await.expect("import succeeds");
let names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect();
assert_eq!(names, vec!["echo/run"], "ACL-filtered op not discovered");
@@ -646,7 +776,8 @@ mod tests {
let adapter = FromWss::new(&endpoint)
.with_auth_token("tok-1")
.with_namespace("remote");
.with_namespace("remote")
.allow_plaintext();
let bundles = adapter.import().await.expect("import succeeds");
let mut names: Vec<&str> = bundles.iter().map(|b| b.spec.name.as_str()).collect();
names.sort();
@@ -661,7 +792,7 @@ mod tests {
)
.await;
let session = WssSession::connect(&endpoint, Some("tok-1"))
let session = WssSession::connect(&endpoint, Some("tok-1"), true)
.await
.expect("connect");
@@ -752,7 +883,7 @@ mod tests {
async fn race_call_resolves_retryable(drop_before_call: bool) {
let (endpoint, pumps_slot) = drop_on_signal_producer(slow_producer_registry());
let session = WssSession::connect(&endpoint, Some("tok-1"))
let session = WssSession::connect(&endpoint, Some("tok-1"), true)
.await
.expect("connect");
@@ -812,7 +943,7 @@ mod tests {
#[tokio::test]
async fn call_registered_after_eof_resolves_via_pending_sweep() {
let (endpoint, pumps_slot) = drop_on_signal_producer(slow_producer_registry());
let session = WssSession::connect(&endpoint, Some("tok-1"))
let session = WssSession::connect(&endpoint, Some("tok-1"), true)
.await
.expect("connect");