fix(server): hyper knobs, decoy fidelity, cache + builder fixes (SRV-04..SRV-10)
- SRV-04: TokioTimer on h1+h2 builder, header_read_timeout 10s, h1 keep-alive on, h2 keep-alive 30s/10s; concurrency boundary documented - SRV-05: with_decoy rebuild keeps extra routes (clone, not take) - SRV-07: method_not_allowed_fallback serves the nginx-shaped 405 - SRV-08: UTF-8 percent-decoding, literal '+', tokio::fs syscalls - SRV-09: /openapi.json cached at construction, generic 500 body, to_openapi returns Result (expect removed) - SRV-10: ChannelsPolicy extension injection point on the WS upgrade; single token resolution via route ordering (WS layer before the router-wide auth route_layer) Verification: cargo test 265 passed; --all-features server::/to_openapi:: green; clippy + fmt clean on touched files (remaining tree noise is a parallel agent's in-flight from_mcp/from_wss/forward work)
This commit is contained in:
+30
-21
@@ -29,6 +29,7 @@ use std::collections::BTreeMap;
|
|||||||
|
|
||||||
use serde_json::{json, Map, Value};
|
use serde_json::{json, Map, Value};
|
||||||
|
|
||||||
|
use alkcall::client::AdapterError;
|
||||||
use alkcall::registry::registration::OperationRegistry;
|
use alkcall::registry::registration::OperationRegistry;
|
||||||
use alkcall::registry::spec::ErrorDefinition;
|
use alkcall::registry::spec::ErrorDefinition;
|
||||||
|
|
||||||
@@ -61,10 +62,18 @@ const CODE_TIMEOUT: &str = "TIMEOUT";
|
|||||||
|
|
||||||
const HTTP_PREFIX: &str = "HTTP_";
|
const HTTP_PREFIX: &str = "HTTP_";
|
||||||
|
|
||||||
pub fn to_openapi(registry: &OperationRegistry) -> OpenAPISpec {
|
/// Project the registry into the fixed 6-endpoint gateway doc (ADR-042).
|
||||||
|
///
|
||||||
|
/// Returns [`AdapterError::SchemaParse`] if the generated doc does not
|
||||||
|
/// re-validate against the structural checks in `OpenAPISpec::from_value`
|
||||||
|
/// — a would-be invariant violation of `build_doc`, not a caller-facing
|
||||||
|
/// input error. The HTTP surface serves only a generic `500` for this
|
||||||
|
/// (`/openapi.json` handler caches the serialized doc; SRV-09): no
|
||||||
|
/// serde/parse internals reach the wire.
|
||||||
|
pub fn to_openapi(registry: &OperationRegistry) -> Result<OpenAPISpec, AdapterError> {
|
||||||
let operation_errors = collect_operation_errors(registry);
|
let operation_errors = collect_operation_errors(registry);
|
||||||
let raw = build_doc(operation_errors);
|
let raw = build_doc(operation_errors);
|
||||||
OpenAPISpec::from_value(raw).expect("to_openapi always emits a valid OpenAPI document")
|
OpenAPISpec::from_value(raw)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn build_doc(operation_errors: Vec<ErrorDefinition>) -> Value {
|
fn build_doc(operation_errors: Vec<ErrorDefinition>) -> Value {
|
||||||
@@ -659,7 +668,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn empty_registry_produces_six_gateway_paths() {
|
fn empty_registry_produces_six_gateway_paths() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let paths = paths_object(&spec);
|
let paths = paths_object(&spec);
|
||||||
assert_eq!(paths.len(), 6);
|
assert_eq!(paths.len(), 6);
|
||||||
assert!(paths.contains_key(PATH_SEARCH));
|
assert!(paths.contains_key(PATH_SEARCH));
|
||||||
@@ -675,7 +684,7 @@ mod tests {
|
|||||||
let mut registry = OperationRegistry::new();
|
let mut registry = OperationRegistry::new();
|
||||||
register(&mut registry, external_spec("fs/readFile", vec![]));
|
register(&mut registry, external_spec("fs/readFile", vec![]));
|
||||||
register(&mut registry, external_spec("agent/chat", vec![]));
|
register(&mut registry, external_spec("agent/chat", vec![]));
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let paths = paths_object(&spec);
|
let paths = paths_object(&spec);
|
||||||
assert_eq!(paths.len(), 6);
|
assert_eq!(paths.len(), 6);
|
||||||
assert!(!paths.contains_key("/fs/readFile"));
|
assert!(!paths.contains_key("/fs/readFile"));
|
||||||
@@ -685,7 +694,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn info_version_is_1_1_0_after_publish_addition() {
|
fn info_version_is_1_1_0_after_publish_addition() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let version = spec
|
let version = spec
|
||||||
.raw
|
.raw
|
||||||
.get("info")
|
.get("info")
|
||||||
@@ -702,7 +711,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn info_title_present() {
|
fn info_title_present() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let title = spec
|
let title = spec
|
||||||
.raw
|
.raw
|
||||||
.get("info")
|
.get("info")
|
||||||
@@ -715,7 +724,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn openapi_field_is_3_0_0() {
|
fn openapi_field_is_3_0_0() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let openapi = spec.raw.get("openapi").and_then(Value::as_str).unwrap();
|
let openapi = spec.raw.get("openapi").and_then(Value::as_str).unwrap();
|
||||||
assert_eq!(openapi, OPENAPI_VERSION);
|
assert_eq!(openapi, OPENAPI_VERSION);
|
||||||
}
|
}
|
||||||
@@ -723,7 +732,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn publish_has_post_method_with_ndjson_request_body() {
|
fn publish_has_post_method_with_ndjson_request_body() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
assert!(path(&spec, PATH_PUBLISH).contains_key("post"));
|
assert!(path(&spec, PATH_PUBLISH).contains_key("post"));
|
||||||
let request_schema = operation(&spec, PATH_PUBLISH, "post")
|
let request_schema = operation(&spec, PATH_PUBLISH, "post")
|
||||||
.get("requestBody")
|
.get("requestBody")
|
||||||
@@ -747,7 +756,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn publish_includes_protocol_error_statuses_including_400() {
|
fn publish_includes_protocol_error_statuses_including_400() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_PUBLISH, "post");
|
let responses = responses(&spec, PATH_PUBLISH, "post");
|
||||||
for status in [
|
for status in [
|
||||||
STATUS_BAD_REQUEST,
|
STATUS_BAD_REQUEST,
|
||||||
@@ -768,7 +777,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn publish_400_response_covers_invalid_input_and_invalid_operation_type() {
|
fn publish_400_response_covers_invalid_input_and_invalid_operation_type() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_PUBLISH, "post");
|
let responses = responses(&spec, PATH_PUBLISH, "post");
|
||||||
let schema = responses
|
let schema = responses
|
||||||
.get(&STATUS_BAD_REQUEST.to_string())
|
.get(&STATUS_BAD_REQUEST.to_string())
|
||||||
@@ -799,7 +808,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn call_request_body_is_flat_operation_input() {
|
fn call_request_body_is_flat_operation_input() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let request_schema = operation(&spec, PATH_CALL, "post")
|
let request_schema = operation(&spec, PATH_CALL, "post")
|
||||||
.get("requestBody")
|
.get("requestBody")
|
||||||
.and_then(|rb| rb.get("content"))
|
.and_then(|rb| rb.get("content"))
|
||||||
@@ -832,7 +841,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn call_includes_all_protocol_level_error_statuses() {
|
fn call_includes_all_protocol_level_error_statuses() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
for status in [
|
for status in [
|
||||||
STATUS_BAD_REQUEST,
|
STATUS_BAD_REQUEST,
|
||||||
@@ -853,7 +862,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn call_protocol_error_status_codes_have_protocol_codes() {
|
fn call_protocol_error_status_codes_have_protocol_codes() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
|
|
||||||
let invalid_input_schema = responses
|
let invalid_input_schema = responses
|
||||||
@@ -906,7 +915,7 @@ mod tests {
|
|||||||
],
|
],
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
assert!(
|
assert!(
|
||||||
responses.contains_key("429"),
|
responses.contains_key("429"),
|
||||||
@@ -934,7 +943,7 @@ mod tests {
|
|||||||
&mut registry,
|
&mut registry,
|
||||||
external_spec("svc/op", vec![error("HTTP_404", Some(404))]),
|
external_spec("svc/op", vec![error("HTTP_404", Some(404))]),
|
||||||
);
|
);
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
let response_404 = responses.get("404").unwrap();
|
let response_404 = responses.get("404").unwrap();
|
||||||
let schema = response_404
|
let schema = response_404
|
||||||
@@ -963,7 +972,7 @@ mod tests {
|
|||||||
&mut registry,
|
&mut registry,
|
||||||
external_spec("svc/op", vec![error("HTTP_404", Some(404))]),
|
external_spec("svc/op", vec![error("HTTP_404", Some(404))]),
|
||||||
);
|
);
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
let response_404 = responses.get("404").unwrap();
|
let response_404 = responses.get("404").unwrap();
|
||||||
let schema = response_404
|
let schema = response_404
|
||||||
@@ -1006,7 +1015,7 @@ mod tests {
|
|||||||
&mut registry,
|
&mut registry,
|
||||||
external_spec("svc/op", vec![error("SOME_ERROR", None)]),
|
external_spec("svc/op", vec![error("SOME_ERROR", None)]),
|
||||||
);
|
);
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
assert!(
|
assert!(
|
||||||
responses.len() < 10,
|
responses.len() < 10,
|
||||||
@@ -1025,7 +1034,7 @@ mod tests {
|
|||||||
&mut registry,
|
&mut registry,
|
||||||
external_spec("svc/b", vec![error("TOO_MANY_REQUESTS", Some(429))]),
|
external_spec("svc/b", vec![error("TOO_MANY_REQUESTS", Some(429))]),
|
||||||
);
|
);
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
assert!(responses.contains_key("429"));
|
assert!(responses.contains_key("429"));
|
||||||
let schema = responses
|
let schema = responses
|
||||||
@@ -1065,7 +1074,7 @@ mod tests {
|
|||||||
Capabilities::new(),
|
Capabilities::new(),
|
||||||
))
|
))
|
||||||
.unwrap();
|
.unwrap();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
assert!(
|
assert!(
|
||||||
!responses.contains_key("418"),
|
!responses.contains_key("418"),
|
||||||
@@ -1080,7 +1089,7 @@ mod tests {
|
|||||||
&mut registry,
|
&mut registry,
|
||||||
external_spec("svc/op", vec![error("HTTP_500", Some(500))]),
|
external_spec("svc/op", vec![error("HTTP_500", Some(500))]),
|
||||||
);
|
);
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let responses = responses(&spec, PATH_CALL, "post");
|
let responses = responses(&spec, PATH_CALL, "post");
|
||||||
let schema = responses
|
let schema = responses
|
||||||
.get("500")
|
.get("500")
|
||||||
@@ -1098,7 +1107,7 @@ mod tests {
|
|||||||
#[test]
|
#[test]
|
||||||
fn doc_validates_against_openapiv3_parsing() {
|
fn doc_validates_against_openapiv3_parsing() {
|
||||||
let registry = OperationRegistry::new();
|
let registry = OperationRegistry::new();
|
||||||
let spec = to_openapi(®istry);
|
let spec = to_openapi(®istry).unwrap();
|
||||||
let text = serde_json::to_string(&spec.raw).unwrap();
|
let text = serde_json::to_string(&spec.raw).unwrap();
|
||||||
let parsed: openapiv3::OpenAPI =
|
let parsed: openapiv3::OpenAPI =
|
||||||
serde_json::from_str(&text).expect("gateway doc parses as OpenAPI 3.0");
|
serde_json::from_str(&text).expect("gateway doc parses as OpenAPI 3.0");
|
||||||
|
|||||||
@@ -836,6 +836,7 @@ mod tests {
|
|||||||
registry: Arc::clone(®istry),
|
registry: Arc::clone(®istry),
|
||||||
identity_provider: Arc::clone(&provider),
|
identity_provider: Arc::clone(&provider),
|
||||||
decoy: crate::server::DecoyConfig::NotFound,
|
decoy: crate::server::DecoyConfig::NotFound,
|
||||||
|
openapi_doc: crate::server::adapter::CachedOpenAPIDoc::new(®istry),
|
||||||
};
|
};
|
||||||
let auth_state = Arc::clone(&provider);
|
let auth_state = Arc::clone(&provider);
|
||||||
gateway_router()
|
gateway_router()
|
||||||
|
|||||||
+1
-1
@@ -8,4 +8,4 @@ pub mod gateway;
|
|||||||
pub mod server;
|
pub mod server;
|
||||||
pub mod websocket;
|
pub mod websocket;
|
||||||
|
|
||||||
pub use server::{decoy_fallback, healthz, DecoyConfig};
|
pub use server::{decoy_fallback, decoy_method_not_allowed, healthz, DecoyConfig};
|
||||||
|
|||||||
+246
-18
@@ -20,8 +20,21 @@
|
|||||||
//! handler on a reserved path and break the "default surface wins"
|
//! handler on a reserved path and break the "default surface wins"
|
||||||
//! rule. Same-path-different-method merges outside the reserved set
|
//! rule. Same-path-different-method merges outside the reserved set
|
||||||
//! remain legal (axum composes the `MethodRouter`).
|
//! remain legal (axum composes the `MethodRouter`).
|
||||||
|
//!
|
||||||
|
//! ## Connection knobs and boundaries
|
||||||
|
//!
|
||||||
|
//! [`HttpAdapter::serve_io`] configures the hyper auto builder with a
|
||||||
|
//! tokio timer and timeouts (values in the method doc). The **concurrency
|
||||||
|
//! cap is not a knob of this crate**: [`ProtocolHandler::handle`] serves
|
||||||
|
//! exactly one accepted bidirectional stream per call, and the accept
|
||||||
|
//! loop — how many streams are handled concurrently, on which tasks —
|
||||||
|
//! belongs to the consumer that owns the `Connection` (or the
|
||||||
|
//! assembly-layer listener). Timeout/cap interaction: a request can
|
||||||
|
//! hold its connection for the whole handler runtime; only the header
|
||||||
|
//! phase and idle keep-alive windows are bounded here.
|
||||||
|
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
use alkcall::core::auth::AuthContext;
|
use alkcall::core::auth::AuthContext;
|
||||||
use alkcall::core::types::{Connection, HandlerError, StreamError};
|
use alkcall::core::types::{Connection, HandlerError, StreamError};
|
||||||
@@ -29,13 +42,14 @@ use alkcall::registry::registration::OperationRegistry;
|
|||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use axum::routing::get;
|
use axum::routing::get;
|
||||||
use axum::Router;
|
use axum::Router;
|
||||||
use hyper_util::rt::{TokioExecutor, TokioIo};
|
use hyper_util::rt::{TokioExecutor, TokioIo, TokioTimer};
|
||||||
use hyper_util::server::conn::auto::Builder as HyperBuilder;
|
use hyper_util::server::conn::auto::Builder as HyperBuilder;
|
||||||
use hyper_util::service::TowerToHyperService;
|
use hyper_util::service::TowerToHyperService;
|
||||||
|
use parking_lot::Mutex;
|
||||||
use tracing::error;
|
use tracing::error;
|
||||||
|
|
||||||
use super::auth::bearer_auth_middleware;
|
use super::auth::bearer_auth_middleware;
|
||||||
use super::decoy::decoy_fallback;
|
use super::decoy::{decoy_fallback, decoy_method_not_allowed};
|
||||||
use super::healthz::healthz;
|
use super::healthz::healthz;
|
||||||
use super::state::{DecoyConfig, RouterState};
|
use super::state::{DecoyConfig, RouterState};
|
||||||
|
|
||||||
@@ -72,6 +86,7 @@ pub struct HttpAdapter {
|
|||||||
extra_routes: Option<Router>,
|
extra_routes: Option<Router>,
|
||||||
alpn: &'static [u8],
|
alpn: &'static [u8],
|
||||||
router: Router,
|
router: Router,
|
||||||
|
openapi_doc: CachedOpenAPIDoc,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl HttpAdapter {
|
impl HttpAdapter {
|
||||||
@@ -95,10 +110,12 @@ impl HttpAdapter {
|
|||||||
alpn: &'static [u8],
|
alpn: &'static [u8],
|
||||||
) -> Self {
|
) -> Self {
|
||||||
let decoy = DecoyConfig::default();
|
let decoy = DecoyConfig::default();
|
||||||
|
let openapi_doc = CachedOpenAPIDoc::new(®istry);
|
||||||
let state = RouterState {
|
let state = RouterState {
|
||||||
registry: Arc::clone(®istry),
|
registry: Arc::clone(®istry),
|
||||||
identity_provider: Arc::clone(&identity_provider),
|
identity_provider: Arc::clone(&identity_provider),
|
||||||
decoy: decoy.clone(),
|
decoy: decoy.clone(),
|
||||||
|
openapi_doc: openapi_doc.clone(),
|
||||||
};
|
};
|
||||||
let router = build_router(state, None);
|
let router = build_router(state, None);
|
||||||
Self {
|
Self {
|
||||||
@@ -108,6 +125,7 @@ impl HttpAdapter {
|
|||||||
extra_routes: None,
|
extra_routes: None,
|
||||||
alpn,
|
alpn,
|
||||||
router,
|
router,
|
||||||
|
openapi_doc,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -117,8 +135,13 @@ impl HttpAdapter {
|
|||||||
registry: Arc::clone(&self.registry),
|
registry: Arc::clone(&self.registry),
|
||||||
identity_provider: Arc::clone(&self.identity_provider),
|
identity_provider: Arc::clone(&self.identity_provider),
|
||||||
decoy,
|
decoy,
|
||||||
|
openapi_doc: self.openapi_doc.clone(),
|
||||||
};
|
};
|
||||||
self.router = build_router(state, self.extra_routes.take());
|
// `extra_routes` is borrowed, not consumed (SRV-05): a builder
|
||||||
|
// call after `with_extra_routes` must keep the custom routes in
|
||||||
|
// the rebuild — `with_extra_routes` stores a verified clone, so
|
||||||
|
// re-merging it is safe.
|
||||||
|
self.router = build_router(state, self.extra_routes.clone());
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -127,6 +150,7 @@ impl HttpAdapter {
|
|||||||
registry: Arc::clone(&self.registry),
|
registry: Arc::clone(&self.registry),
|
||||||
identity_provider: Arc::clone(&self.identity_provider),
|
identity_provider: Arc::clone(&self.identity_provider),
|
||||||
decoy: self.decoy.clone(),
|
decoy: self.decoy.clone(),
|
||||||
|
openapi_doc: self.openapi_doc.clone(),
|
||||||
};
|
};
|
||||||
self.router = build_router(state, Some(routes.clone()));
|
self.router = build_router(state, Some(routes.clone()));
|
||||||
self.extra_routes = Some(routes);
|
self.extra_routes = Some(routes);
|
||||||
@@ -171,13 +195,21 @@ fn build_router(state: RouterState, extra_routes: Option<Router>) -> Router {
|
|||||||
|
|
||||||
let default: Router<RouterState> = Router::new()
|
let default: Router<RouterState> = Router::new()
|
||||||
.merge(crate::gateway::routes::gateway_router())
|
.merge(crate::gateway::routes::gateway_router())
|
||||||
|
// The openapi handler's state is the pre-serialized projection
|
||||||
|
// cache (SRV-09), threaded through `RouterState` below.
|
||||||
.route("/openapi.json", get(openapi_json_handler))
|
.route("/openapi.json", get(openapi_json_handler))
|
||||||
.route("/healthz", get(healthz))
|
.route("/healthz", get(healthz))
|
||||||
// The WS upgrade route carries its own bearer middleware
|
// The WS upgrade route carries its own bearer middleware
|
||||||
// (`ws_bearer_auth` — 401 without a resolvable token, because a WS
|
// (`ws_bearer_auth` — 401 without a resolvable token, because a WS
|
||||||
// session without an identity cannot run AccessControl::check); the
|
// session without an identity cannot run AccessControl::check).
|
||||||
// shared bearer_auth_middleware only stashes a permissive
|
// This route is registered BEFORE the router-wide
|
||||||
// Option<Identity> for the gateway endpoints.
|
// bearer_auth_middleware route_layer below (route_layer applies
|
||||||
|
// only to routes registered before it), so the WS path resolves
|
||||||
|
// the token exactly once — enforced (401) — while the gateway
|
||||||
|
// endpoints resolve once, permissively (no enforcement; the
|
||||||
|
// dispatch's AccessControl decides). See SRV-10: the previous
|
||||||
|
// mux nested this route inside the shared layer and resolved
|
||||||
|
// the same token twice.
|
||||||
.route(
|
.route(
|
||||||
WS_UPGRADE_PATH,
|
WS_UPGRADE_PATH,
|
||||||
get(crate::websocket::ws_upgrade_handler).route_layer(from_fn_with_state(
|
get(crate::websocket::ws_upgrade_handler).route_layer(from_fn_with_state(
|
||||||
@@ -186,6 +218,7 @@ fn build_router(state: RouterState, extra_routes: Option<Router>) -> Router {
|
|||||||
)),
|
)),
|
||||||
)
|
)
|
||||||
.fallback(decoy_fallback)
|
.fallback(decoy_fallback)
|
||||||
|
.method_not_allowed_fallback(decoy_method_not_allowed)
|
||||||
.merge(mcp_router);
|
.merge(mcp_router);
|
||||||
|
|
||||||
let with_extras = match extra_routes {
|
let with_extras = match extra_routes {
|
||||||
@@ -198,10 +231,12 @@ fn build_router(state: RouterState, extra_routes: Option<Router>) -> Router {
|
|||||||
};
|
};
|
||||||
|
|
||||||
// Applied after the merges (ADR-046 §4): the bearer-auth layer wraps
|
// Applied after the merges (ADR-046 §4): the bearer-auth layer wraps
|
||||||
// the extra routes too, so the documented default — custom routes
|
// the extra routes and every default-surface route registered above
|
||||||
// carry the same auth — holds. A route that wants to opt out carries
|
// except the WS upgrade route (registered earlier with its own
|
||||||
// its own inner auth layer, which runs innermost (last-applied wins
|
// enforced layer, so the token resolves once per request — SRV-10).
|
||||||
// for extension stashes it inserts).
|
// A route that wants to opt out carries its own inner auth layer,
|
||||||
|
// which runs innermost (last-applied wins for extension stashes it
|
||||||
|
// inserts).
|
||||||
let with_extras =
|
let with_extras =
|
||||||
with_extras.route_layer(from_fn_with_state(auth_state, bearer_auth_middleware));
|
with_extras.route_layer(from_fn_with_state(auth_state, bearer_auth_middleware));
|
||||||
|
|
||||||
@@ -370,6 +405,26 @@ impl alkcall::core::types::ProtocolHandler for HttpAdapter {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl HttpAdapter {
|
impl HttpAdapter {
|
||||||
|
/// Serve one accepted bidirectional stream as a single HTTP
|
||||||
|
/// connection.
|
||||||
|
///
|
||||||
|
/// Hyper knobs (SRV-04) — hyper 1.11 *silently ignores* the
|
||||||
|
/// header-read timeout unless a timer is set, so the timer is
|
||||||
|
/// explicit and the values chosen are:
|
||||||
|
///
|
||||||
|
/// - `header_read_timeout` = **10 s** (tighter than hyper's 30 s
|
||||||
|
/// default; bounds the slow-loris window where a client drips
|
||||||
|
/// request header bytes)
|
||||||
|
/// - h1 `keep_alive` = **enabled** (default; normal client reuse),
|
||||||
|
/// with hyper's default 30 s idle header-read window applying
|
||||||
|
/// per-request
|
||||||
|
/// - h2 `keep_alive_interval` = **30 s**, `keep_alive_timeout` =
|
||||||
|
/// **10 s** — a peer that fails to ack pings for 10 s is dropped,
|
||||||
|
/// so half-open h2 connections do not accumulate
|
||||||
|
///
|
||||||
|
/// No concurrency cap is set here — see the module doc: the accept
|
||||||
|
/// loop is the consumer's, and the per-connection "cap" is one
|
||||||
|
/// stream per `handle` call.
|
||||||
async fn serve_io<I>(&self, io: I) -> Result<(), HandlerError>
|
async fn serve_io<I>(&self, io: I) -> Result<(), HandlerError>
|
||||||
where
|
where
|
||||||
I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
|
I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Send + Unpin + 'static,
|
||||||
@@ -377,11 +432,28 @@ impl HttpAdapter {
|
|||||||
let io = TokioIo::new(io);
|
let io = TokioIo::new(io);
|
||||||
let service = TowerToHyperService::new(self.router.clone());
|
let service = TowerToHyperService::new(self.router.clone());
|
||||||
|
|
||||||
|
const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(10);
|
||||||
|
const H2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(30);
|
||||||
|
const H2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(10);
|
||||||
|
|
||||||
#[cfg_attr(not(feature = "h2"), allow(unused_mut))]
|
#[cfg_attr(not(feature = "h2"), allow(unused_mut))]
|
||||||
let mut builder = HyperBuilder::new(TokioExecutor::new());
|
let mut builder = HyperBuilder::new(TokioExecutor::new());
|
||||||
|
#[cfg(feature = "http1")]
|
||||||
|
{
|
||||||
|
builder
|
||||||
|
.http1()
|
||||||
|
.timer(TokioTimer::new())
|
||||||
|
.keep_alive(true)
|
||||||
|
.header_read_timeout(HEADER_READ_TIMEOUT);
|
||||||
|
}
|
||||||
#[cfg(feature = "h2")]
|
#[cfg(feature = "h2")]
|
||||||
{
|
{
|
||||||
builder.http2().enable_connect_protocol();
|
builder
|
||||||
|
.http2()
|
||||||
|
.timer(TokioTimer::new())
|
||||||
|
.enable_connect_protocol()
|
||||||
|
.keep_alive_interval(H2_KEEP_ALIVE_INTERVAL)
|
||||||
|
.keep_alive_timeout(H2_KEEP_ALIVE_TIMEOUT);
|
||||||
}
|
}
|
||||||
|
|
||||||
let conn = builder.serve_connection_with_upgrades(io, service);
|
let conn = builder.serve_connection_with_upgrades(io, service);
|
||||||
@@ -399,20 +471,77 @@ fn stream_error_to_handler(e: StreamError) -> HandlerError {
|
|||||||
HandlerError::from(e)
|
HandlerError::from(e)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Serialized-to-bytes cache of the `/openapi.json` projection
|
||||||
|
/// (SRV-09): built once per `HttpAdapter` from the registry at
|
||||||
|
/// construction instead of re-projecting + re-serializing per request.
|
||||||
|
/// The registry is defined not to mutate after assembly (the assembly
|
||||||
|
/// layer registers handlers before serving), so the doc is fixed for
|
||||||
|
/// the adapter's lifetime; a rebuild via [`HttpAdapter::with_decoy`] /
|
||||||
|
/// `with_extra_routes` re-derives it from the same registry.
|
||||||
|
///
|
||||||
|
/// Serialization happens once, here — so a serde failure surfaces at
|
||||||
|
/// construction as a tracing error and the handler answers `/openapi.json`
|
||||||
|
/// with a generic 500 (no serde internals on the wire) rather than
|
||||||
|
/// echoing an error to unauthenticated callers.
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub(crate) struct CachedOpenAPIDoc {
|
||||||
|
inner: Arc<Mutex<Option<CachedOpenAPIDocInner>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct CachedOpenAPIDocInner {
|
||||||
|
bytes: axum::body::Bytes,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl CachedOpenAPIDoc {
|
||||||
|
pub(crate) fn new(registry: &OperationRegistry) -> Self {
|
||||||
|
Self {
|
||||||
|
inner: Arc::new(Mutex::new(None)),
|
||||||
|
}
|
||||||
|
.with_registry(registry)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn with_registry(self, registry: &OperationRegistry) -> Self {
|
||||||
|
let spec = match crate::adapters::to_openapi(registry) {
|
||||||
|
Ok(spec) => spec,
|
||||||
|
Err(e) => {
|
||||||
|
error!("openapi.json projection failed; endpoint will return 500: {e}");
|
||||||
|
return self;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
match serde_json::to_vec(&spec.raw) {
|
||||||
|
Ok(bytes) => {
|
||||||
|
*self.inner.lock() = Some(CachedOpenAPIDocInner {
|
||||||
|
bytes: axum::body::Bytes::from(bytes),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
error!("openapi.json serialization failed; endpoint will return 500: {e}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
fn bytes(&self) -> Option<axum::body::Bytes> {
|
||||||
|
self.inner.lock().as_ref().map(|c| c.bytes.clone())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// `GET /openapi.json` — the `to_openapi` projection of the local
|
/// `GET /openapi.json` — the `to_openapi` projection of the local
|
||||||
/// operation registry (ADR-042, ADR-045): the fixed 6-endpoint gateway
|
/// operation registry (ADR-042, ADR-045): the fixed 6-endpoint gateway
|
||||||
/// doc. Served under the bearer-auth route layer like every other
|
/// doc. Served under the bearer-auth route layer like every other
|
||||||
/// gateway endpoint.
|
/// gateway endpoint. The serialized doc is cached at adapter
|
||||||
|
/// construction ([`CachedOpenAPIDoc`]); a cache miss (projection or
|
||||||
|
/// serialization failed at construction) answers with a **generic**
|
||||||
|
/// `500` body — no serde internals on the wire.
|
||||||
async fn openapi_json_handler(
|
async fn openapi_json_handler(
|
||||||
axum::extract::State(registry): axum::extract::State<Arc<OperationRegistry>>,
|
axum::extract::State(doc): axum::extract::State<CachedOpenAPIDoc>,
|
||||||
) -> axum::response::Response {
|
) -> axum::response::Response {
|
||||||
use axum::response::IntoResponse;
|
use axum::response::IntoResponse;
|
||||||
let spec = crate::adapters::to_openapi(®istry);
|
match doc.bytes() {
|
||||||
match serde_json::to_vec(&spec.raw) {
|
Some(bytes) => ([(http::header::CONTENT_TYPE, "application/json")], bytes).into_response(),
|
||||||
Ok(bytes) => ([(http::header::CONTENT_TYPE, "application/json")], bytes).into_response(),
|
None => (
|
||||||
Err(e) => (
|
|
||||||
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
axum::http::StatusCode::INTERNAL_SERVER_ERROR,
|
||||||
format!("failed to serialize gateway spec: {e}"),
|
"internal server error",
|
||||||
)
|
)
|
||||||
.into_response(),
|
.into_response(),
|
||||||
}
|
}
|
||||||
@@ -811,6 +940,7 @@ mod tests {
|
|||||||
registry: empty_registry(),
|
registry: empty_registry(),
|
||||||
identity_provider: idp,
|
identity_provider: idp,
|
||||||
decoy: DecoyConfig::default(),
|
decoy: DecoyConfig::default(),
|
||||||
|
openapi_doc: CachedOpenAPIDoc::new(&OperationRegistry::new()),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -829,6 +959,14 @@ mod tests {
|
|||||||
.unwrap()
|
.unwrap()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn get_with_bearer_with_method(
|
||||||
|
app: Router,
|
||||||
|
request: axum::http::Request<axum::body::Body>,
|
||||||
|
) -> axum::http::Response<axum::body::Body> {
|
||||||
|
use tower::ServiceExt;
|
||||||
|
app.oneshot(request).await.unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn extra_routes_resolve_bearer_identity_through_the_default_auth() {
|
async fn extra_routes_resolve_bearer_identity_through_the_default_auth() {
|
||||||
let extra = Router::new().route(
|
let extra = Router::new().route(
|
||||||
@@ -913,4 +1051,94 @@ mod tests {
|
|||||||
HttpAdapter::new(static_provider(), empty_registry()).with_extra_routes(extra);
|
HttpAdapter::new(static_provider(), empty_registry()).with_extra_routes(extra);
|
||||||
assert!(adapter.router().has_routes());
|
assert!(adapter.router().has_routes());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn second_with_decoy_keeps_extra_routes() {
|
||||||
|
let extra = Router::new().route("/v1/ping", get(|| async { "pong" }));
|
||||||
|
let adapter = HttpAdapter::new(static_provider(), empty_registry())
|
||||||
|
.with_extra_routes(extra)
|
||||||
|
.with_decoy(DecoyConfig::Redirect {
|
||||||
|
to: "https://example.com".to_string(),
|
||||||
|
});
|
||||||
|
|
||||||
|
let request = axum::http::Request::builder()
|
||||||
|
.uri("/v1/ping")
|
||||||
|
.body(axum::body::Body::empty())
|
||||||
|
.unwrap();
|
||||||
|
let response = get_with_bearer(adapter.router().clone(), "/v1/ping", None).await;
|
||||||
|
drop(request);
|
||||||
|
assert_eq!(
|
||||||
|
response.status(),
|
||||||
|
axum::http::StatusCode::OK,
|
||||||
|
"a second builder call must not drop the extra routes (SRV-05)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn method_mismatch_on_registered_path_serves_decoy_405() {
|
||||||
|
let adapter = HttpAdapter::new(static_provider(), empty_registry());
|
||||||
|
let request = axum::http::Request::builder()
|
||||||
|
.method(axum::http::Method::OPTIONS)
|
||||||
|
.uri("/search")
|
||||||
|
.body(axum::body::Body::empty())
|
||||||
|
.unwrap();
|
||||||
|
let response = get_with_bearer_with_method(adapter.router().clone(), request).await;
|
||||||
|
assert_eq!(
|
||||||
|
response.status(),
|
||||||
|
axum::http::StatusCode::METHOD_NOT_ALLOWED
|
||||||
|
);
|
||||||
|
let server = response
|
||||||
|
.headers()
|
||||||
|
.get(axum::http::header::SERVER)
|
||||||
|
.map(|v| v.to_str().unwrap().to_string());
|
||||||
|
assert_eq!(
|
||||||
|
server.as_deref(),
|
||||||
|
Some("nginx"),
|
||||||
|
"405 must carry the decoy Server header, not axum's bare 405 (SRV-07)"
|
||||||
|
);
|
||||||
|
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let body = String::from_utf8_lossy(&body);
|
||||||
|
assert!(body.contains("405 Not Allowed"), "got: {body}");
|
||||||
|
assert!(
|
||||||
|
!body.contains("axum") && !body.contains("alk"),
|
||||||
|
"got: {body}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn openapi_json_is_cached_and_generic_on_cache_miss() {
|
||||||
|
let adapter = HttpAdapter::new(static_provider(), empty_registry());
|
||||||
|
let first = openapi_json_handler(axum::extract::State(adapter.openapi_doc.clone())).await;
|
||||||
|
let second = openapi_json_handler(axum::extract::State(adapter.openapi_doc.clone())).await;
|
||||||
|
let first_bytes = axum::body::to_bytes(first.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let second_bytes = axum::body::to_bytes(second.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
&first_bytes[..],
|
||||||
|
&second_bytes[..],
|
||||||
|
"cached doc is byte-stable"
|
||||||
|
);
|
||||||
|
|
||||||
|
let miss = CachedOpenAPIDoc {
|
||||||
|
inner: Arc::new(Mutex::new(None)),
|
||||||
|
};
|
||||||
|
let response = openapi_json_handler(axum::extract::State(miss)).await;
|
||||||
|
assert_eq!(
|
||||||
|
response.status(),
|
||||||
|
axum::http::StatusCode::INTERNAL_SERVER_ERROR
|
||||||
|
);
|
||||||
|
let body = axum::body::to_bytes(response.into_body(), usize::MAX)
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
&body[..],
|
||||||
|
b"internal server error",
|
||||||
|
"the 500 body is the fixed generic string, no serde internals"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+134
-31
@@ -4,9 +4,12 @@
|
|||||||
//! endpoints, `/healthz`, `/openapi.json`, the MCP route, the WS
|
//! endpoints, `/healthz`, `/openapi.json`, the MCP route, the WS
|
||||||
//! upgrade) nor by a custom route ([ADR-046]), the HTTP handler serves
|
//! upgrade) nor by a custom route ([ADR-046]), the HTTP handler serves
|
||||||
//! a configurable decoy ([`DecoyConfig`]): a fake nginx-style 404 (the
|
//! a configurable decoy ([`DecoyConfig`]): a fake nginx-style 404 (the
|
||||||
//! default), a static site served from a directory, or a redirect. The
|
//! default), a static site served from a directory, or a redirect. A
|
||||||
//! decoy must not leak alk presence — no alk-specific headers, no alk
|
//! method mismatch on a registered path (`405`) is served the same
|
||||||
//! error format.
|
//! nginx shape ([`decoy_method_not_allowed`], wired via
|
||||||
|
//! `Router::method_not_allowed_fallback`) so every reachable error
|
||||||
|
//! response carries the decoy `Server` header. The decoy must not leak
|
||||||
|
//! alk presence — no alk-specific headers, no alk error format.
|
||||||
//!
|
//!
|
||||||
//! [ADR-010]: https://docs.rs/alkhttp (docs/architecture/decisions)
|
//! [ADR-010]: https://docs.rs/alkhttp (docs/architecture/decisions)
|
||||||
//! [ADR-036]: https://docs.rs/alkhttp (docs/architecture/decisions)
|
//! [ADR-036]: https://docs.rs/alkhttp (docs/architecture/decisions)
|
||||||
@@ -30,7 +33,7 @@ pub async fn decoy_fallback(State(decoy): State<DecoyConfig>, request: Request)
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn fake_nginx_404() -> Response {
|
pub fn fake_nginx_404() -> Response {
|
||||||
let body = nginx_404_body();
|
let body = nginx_error_body("404 Not Found");
|
||||||
let mut resp = Response::new(Body::from(body));
|
let mut resp = Response::new(Body::from(body));
|
||||||
*resp.status_mut() = StatusCode::NOT_FOUND;
|
*resp.status_mut() = StatusCode::NOT_FOUND;
|
||||||
resp.headers_mut().insert(
|
resp.headers_mut().insert(
|
||||||
@@ -42,6 +45,27 @@ pub fn fake_nginx_404() -> Response {
|
|||||||
resp
|
resp
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn nginx_405_response() -> Response {
|
||||||
|
let body = nginx_error_body("405 Not Allowed");
|
||||||
|
let mut resp = Response::new(Body::from(body));
|
||||||
|
*resp.status_mut() = StatusCode::METHOD_NOT_ALLOWED;
|
||||||
|
resp.headers_mut().insert(
|
||||||
|
header::CONTENT_TYPE,
|
||||||
|
HeaderValue::from_static("text/html; charset=utf-8"),
|
||||||
|
);
|
||||||
|
resp.headers_mut()
|
||||||
|
.insert(header::SERVER, HeaderValue::from_static("nginx"));
|
||||||
|
resp
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Method-mismatch fallback handler wired with
|
||||||
|
/// `Router::method_not_allowed_fallback` — applies to every
|
||||||
|
/// previously registered `MethodRouter` (default surface + extra
|
||||||
|
/// routes).
|
||||||
|
pub async fn decoy_method_not_allowed() -> Response {
|
||||||
|
nginx_405_response()
|
||||||
|
}
|
||||||
|
|
||||||
pub fn redirect(to: &str) -> Response {
|
pub fn redirect(to: &str) -> Response {
|
||||||
let mut resp = Response::new(Body::empty());
|
let mut resp = Response::new(Body::empty());
|
||||||
*resp.status_mut() = StatusCode::FOUND;
|
*resp.status_mut() = StatusCode::FOUND;
|
||||||
@@ -53,7 +77,7 @@ pub fn redirect(to: &str) -> Response {
|
|||||||
|
|
||||||
pub async fn serve_static(root: &Path, request: Request) -> Response {
|
pub async fn serve_static(root: &Path, request: Request) -> Response {
|
||||||
let path = request.uri().path();
|
let path = request.uri().path();
|
||||||
let resolved = match resolve_static_path(root, path) {
|
let resolved = match resolve_static_path(root, path).await {
|
||||||
Some(p) => p,
|
Some(p) => p,
|
||||||
None => return fake_nginx_404(),
|
None => return fake_nginx_404(),
|
||||||
};
|
};
|
||||||
@@ -71,12 +95,12 @@ pub async fn serve_static(root: &Path, request: Request) -> Response {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn resolve_static_path(root: &Path, request_path: &str) -> Option<PathBuf> {
|
async fn resolve_static_path(root: &Path, request_path: &str) -> Option<PathBuf> {
|
||||||
let trimmed = request_path.trim_start_matches('/');
|
let trimmed = request_path.trim_start_matches('/');
|
||||||
let relative = if trimmed.is_empty() {
|
let relative = if trimmed.is_empty() {
|
||||||
PathBuf::from("index.html")
|
PathBuf::from("index.html")
|
||||||
} else {
|
} else {
|
||||||
let decoded = percent_decode(trimmed);
|
let decoded = percent_decode(trimmed.as_bytes())?;
|
||||||
PathBuf::from(decoded)
|
PathBuf::from(decoded)
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -94,36 +118,45 @@ fn resolve_static_path(root: &Path, request_path: &str) -> Option<PathBuf> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let full = root.join(&safe);
|
let full = root.join(&safe);
|
||||||
if full.is_dir() {
|
if tokio::fs::metadata(&full).await.is_ok_and(|m| m.is_dir()) {
|
||||||
return Some(full.join("index.html"));
|
let index = full.join("index.html");
|
||||||
|
return tokio::fs::metadata(&index)
|
||||||
|
.await
|
||||||
|
.is_ok_and(|m| m.is_file())
|
||||||
|
.then_some(index)
|
||||||
|
.or(Some(full));
|
||||||
}
|
}
|
||||||
if full.is_file() {
|
tokio::fs::metadata(&full)
|
||||||
return Some(full);
|
.await
|
||||||
}
|
.is_ok_and(|m| m.is_file())
|
||||||
None
|
.then_some(full)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn percent_decode(input: &str) -> String {
|
/// Percent-decode a URI path segment to UTF-8.
|
||||||
let mut out = String::with_capacity(input.len());
|
///
|
||||||
let bytes = input.as_bytes();
|
/// `+` is left as `+` (its space meaning is `application/x-www-form-urlencoded`,
|
||||||
|
/// not the URI path grammar), and `%XX` escapes are accumulated at the byte
|
||||||
|
/// level across adjacent escapes (`%C3%A9` → one UTF-8 sequence) before a
|
||||||
|
/// single UTF-8 validation. Invalid escapes and non-UTF-8 sequences yield
|
||||||
|
/// `None` — the caller serves the fake 404 rather than guessing an encoding.
|
||||||
|
fn percent_decode(input: &[u8]) -> Option<String> {
|
||||||
|
let mut out = Vec::with_capacity(input.len());
|
||||||
let mut i = 0;
|
let mut i = 0;
|
||||||
while i < bytes.len() {
|
while i < input.len() {
|
||||||
let b = bytes[i];
|
match input[i] {
|
||||||
if b == b'%' && i + 2 < bytes.len() {
|
b'%' if i + 2 < input.len() => {
|
||||||
if let (Some(h), Some(l)) = (hex_digit(bytes[i + 1]), hex_digit(bytes[i + 2])) {
|
let h = hex_digit(input[i + 1])?;
|
||||||
out.push(((h << 4) | l) as char);
|
let l = hex_digit(input[i + 2])?;
|
||||||
|
out.push((h << 4) | l);
|
||||||
i += 3;
|
i += 3;
|
||||||
continue;
|
|
||||||
}
|
}
|
||||||
} else if b == b'+' {
|
b => {
|
||||||
out.push(' ');
|
out.push(b);
|
||||||
i += 1;
|
|
||||||
continue;
|
|
||||||
}
|
|
||||||
out.push(b as char);
|
|
||||||
i += 1;
|
i += 1;
|
||||||
}
|
}
|
||||||
out
|
}
|
||||||
|
}
|
||||||
|
String::from_utf8(out).ok()
|
||||||
}
|
}
|
||||||
|
|
||||||
fn hex_digit(b: u8) -> Option<u8> {
|
fn hex_digit(b: u8) -> Option<u8> {
|
||||||
@@ -153,8 +186,10 @@ fn mime_for_path(path: &Path) -> &'static str {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn nginx_404_body() -> String {
|
fn nginx_error_body(title: &str) -> String {
|
||||||
"<html>\r\n<head><title>404 Not Found</title></head>\r\n<body>\r\n<center><h1>404 Not Found</h1></center>\r\n<hr><center>nginx</center>\r\n</body>\r\n</html>\r\n".to_string()
|
format!(
|
||||||
|
"<html>\r\n<head><title>{title}</title></head>\r\n<body>\r\n<center><h1>{title}</h1></center>\r\n<hr><center>nginx</center>\r\n</body>\r\n</html>\r\n"
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
@@ -268,6 +303,74 @@ mod tests {
|
|||||||
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
|
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn static_site_decoy_percent_encoded_utf8_filename_resolves() {
|
||||||
|
let dir = tempfile_dir();
|
||||||
|
tokio::fs::write(dir.join("café.html"), "cafe")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let decoy = DecoyConfig::StaticSite { root: dir };
|
||||||
|
let resp = send(decoy_router(decoy), "/caf%C3%A9.html").await;
|
||||||
|
assert_eq!(
|
||||||
|
resp.status(),
|
||||||
|
StatusCode::OK,
|
||||||
|
"%C3%A9 must decode to é as one UTF-8 sequence"
|
||||||
|
);
|
||||||
|
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
|
||||||
|
assert_eq!(&bytes[..], b"cafe");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn static_site_decoy_plus_sign_is_literal_in_paths() {
|
||||||
|
let dir = tempfile_dir();
|
||||||
|
tokio::fs::write(dir.join("a+b.html"), "plus")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let decoy = DecoyConfig::StaticSite { root: dir };
|
||||||
|
let resp = send(decoy_router(decoy), "/a+b.html").await;
|
||||||
|
assert_eq!(
|
||||||
|
resp.status(),
|
||||||
|
StatusCode::OK,
|
||||||
|
"+ is a literal path byte, not a space"
|
||||||
|
);
|
||||||
|
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
|
||||||
|
assert_eq!(&bytes[..], b"plus");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn static_site_decoy_invalid_percent_escape_returns_fake_404() {
|
||||||
|
let dir = tempfile_dir();
|
||||||
|
tokio::fs::write(dir.join("index.html"), "ok")
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let decoy = DecoyConfig::StaticSite { root: dir };
|
||||||
|
let resp = send(decoy_router(decoy), "/%zz.html").await;
|
||||||
|
assert_eq!(resp.status(), StatusCode::NOT_FOUND);
|
||||||
|
let server = resp
|
||||||
|
.headers()
|
||||||
|
.get(header::SERVER)
|
||||||
|
.map(|v| v.to_str().unwrap().to_string());
|
||||||
|
assert_eq!(server.as_deref(), Some("nginx"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn method_not_allowed_decoy_carries_nginx_server_header() {
|
||||||
|
let resp = decoy_method_not_allowed().await;
|
||||||
|
assert_eq!(resp.status(), StatusCode::METHOD_NOT_ALLOWED);
|
||||||
|
let server = resp
|
||||||
|
.headers()
|
||||||
|
.get(header::SERVER)
|
||||||
|
.map(|v| v.to_str().unwrap().to_string());
|
||||||
|
assert_eq!(server.as_deref(), Some("nginx"));
|
||||||
|
let bytes = resp.into_body().collect().await.unwrap().to_bytes();
|
||||||
|
let body = String::from_utf8_lossy(&bytes);
|
||||||
|
assert!(body.contains("405 Not Allowed"), "got: {body}");
|
||||||
|
assert!(!body.contains("axum") && !body.contains("alk"));
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn not_found_decoy_does_not_leak_alk_headers() {
|
async fn not_found_decoy_does_not_leak_alk_headers() {
|
||||||
let resp = send(decoy_router(DecoyConfig::NotFound), "/whatever").await;
|
let resp = send(decoy_router(DecoyConfig::NotFound), "/whatever").await;
|
||||||
|
|||||||
+1
-1
@@ -6,6 +6,6 @@ pub mod state;
|
|||||||
|
|
||||||
pub use adapter::{HttpAdapter, RESERVED_PATHS, WS_UPGRADE_PATH};
|
pub use adapter::{HttpAdapter, RESERVED_PATHS, WS_UPGRADE_PATH};
|
||||||
pub use auth::{bearer_auth_middleware, extract_bearer_identity, ResolvedIdentity};
|
pub use auth::{bearer_auth_middleware, extract_bearer_identity, ResolvedIdentity};
|
||||||
pub use decoy::decoy_fallback;
|
pub use decoy::{decoy_fallback, decoy_method_not_allowed};
|
||||||
pub use healthz::healthz;
|
pub use healthz::healthz;
|
||||||
pub use state::DecoyConfig;
|
pub use state::DecoyConfig;
|
||||||
|
|||||||
+10
-1
@@ -25,12 +25,14 @@ pub enum DecoyConfig {
|
|||||||
|
|
||||||
/// State embedded in the axum `Router`: the registry and identity
|
/// State embedded in the axum `Router`: the registry and identity
|
||||||
/// provider every request handler reaches through the router state, plus
|
/// provider every request handler reaches through the router state, plus
|
||||||
/// the decoy config for the fallback.
|
/// the decoy config for the fallback and the pre-serialized
|
||||||
|
/// `/openapi.json` projection cache (SRV-09).
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub(crate) struct RouterState {
|
pub(crate) struct RouterState {
|
||||||
pub(crate) registry: Arc<OperationRegistry>,
|
pub(crate) registry: Arc<OperationRegistry>,
|
||||||
pub(crate) identity_provider: Arc<dyn IdentityProvider>,
|
pub(crate) identity_provider: Arc<dyn IdentityProvider>,
|
||||||
pub(crate) decoy: DecoyConfig,
|
pub(crate) decoy: DecoyConfig,
|
||||||
|
pub(crate) openapi_doc: crate::server::adapter::CachedOpenAPIDoc,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl axum::extract::FromRef<RouterState> for DecoyConfig {
|
impl axum::extract::FromRef<RouterState> for DecoyConfig {
|
||||||
@@ -51,6 +53,12 @@ impl axum::extract::FromRef<RouterState> for Arc<dyn IdentityProvider> {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl axum::extract::FromRef<RouterState> for crate::server::adapter::CachedOpenAPIDoc {
|
||||||
|
fn from_ref(state: &RouterState) -> Self {
|
||||||
|
state.openapi_doc.clone()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
@@ -68,6 +76,7 @@ mod tests {
|
|||||||
decoy: DecoyConfig::Redirect {
|
decoy: DecoyConfig::Redirect {
|
||||||
to: "https://example.com".to_string(),
|
to: "https://example.com".to_string(),
|
||||||
},
|
},
|
||||||
|
openapi_doc: crate::server::adapter::CachedOpenAPIDoc::new(&OperationRegistry::new()),
|
||||||
};
|
};
|
||||||
let extracted: DecoyConfig = axum::extract::FromRef::from_ref(&state);
|
let extracted: DecoyConfig = axum::extract::FromRef::from_ref(&state);
|
||||||
assert!(matches!(extracted, DecoyConfig::Redirect { .. }));
|
assert!(matches!(extracted, DecoyConfig::Redirect { .. }));
|
||||||
|
|||||||
@@ -21,7 +21,7 @@ pub use byte_adapter::split_tungstenite_to_bytes;
|
|||||||
#[cfg(any(test, feature = "wss"))]
|
#[cfg(any(test, feature = "wss"))]
|
||||||
#[allow(unused_imports)]
|
#[allow(unused_imports)]
|
||||||
pub(crate) use upgrade::adapter_install_channel_zero;
|
pub(crate) use upgrade::adapter_install_channel_zero;
|
||||||
pub use upgrade::{run_channels_session, ws_bearer_auth, ws_upgrade_handler};
|
pub use upgrade::{run_channels_session, ws_bearer_auth, ws_upgrade_handler, ChannelsPolicy};
|
||||||
|
|
||||||
#[cfg(any(test, feature = "test-support"))]
|
#[cfg(any(test, feature = "test-support"))]
|
||||||
pub use upgrade::test_support::{frame_channel0_chunk, ChunkAssembler, FrameAssembler, WsClient};
|
pub use upgrade::test_support::{frame_channel0_chunk, ChunkAssembler, FrameAssembler, WsClient};
|
||||||
|
|||||||
@@ -25,8 +25,7 @@ use super::byte_adapter::split_ws_to_bytes;
|
|||||||
|
|
||||||
/// The channels session for an upgraded socket: adapt → `Connection`
|
/// The channels session for an upgraded socket: adapt → `Connection`
|
||||||
/// (identity attached) → `ChannelsAdapter::handle`. `policy` gates
|
/// (identity attached) → `ChannelsAdapter::handle`. `policy` gates
|
||||||
/// data-channel opens (ADR-041); the default surface uses `NoCap`
|
/// data-channel opens (ADR-041).
|
||||||
/// (the deployment's assembly layer can pass a stricter policy).
|
|
||||||
pub async fn run_channels_session(
|
pub async fn run_channels_session(
|
||||||
socket: axum::extract::ws::WebSocket,
|
socket: axum::extract::ws::WebSocket,
|
||||||
registry: Arc<OperationRegistry>,
|
registry: Arc<OperationRegistry>,
|
||||||
@@ -115,16 +114,34 @@ impl alkcall::core::auth::IdentityProvider for NoopProvider {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Extension wrapper for the channels-policy injection point (SRV-10):
|
||||||
|
/// a deployment inserts `ChannelsPolicy(Arc<dyn ChannelLifecyclePolicy>)`
|
||||||
|
/// into the request extensions (a route layer on the WS route) to gate
|
||||||
|
/// data-channel opens per identity (ADR-041). Without the extension the
|
||||||
|
/// upgrade defaults to [`NoCap`] — the crate default for the built-in
|
||||||
|
/// surface (POC/trusted-peer semantics); assembly layers that build
|
||||||
|
/// their own upgrade route pass a stricter policy directly to
|
||||||
|
/// [`run_channels_session`].
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct ChannelsPolicy(pub Arc<dyn ChannelLifecyclePolicy>);
|
||||||
|
|
||||||
/// The upgrade handler. Requires the resolved identity in request
|
/// The upgrade handler. Requires the resolved identity in request
|
||||||
/// extensions (stashed by [`ws_bearer_auth`]) — a WS session without
|
/// extensions (stashed by [`ws_bearer_auth`]) — a WS session without
|
||||||
/// an identity cannot run `AccessControl::check`.
|
/// an identity cannot run `AccessControl::check`.
|
||||||
|
///
|
||||||
|
/// The channel lifecycle policy comes from the
|
||||||
|
/// [`ChannelsPolicy`] request extension when present, else `NoCap`.
|
||||||
pub async fn ws_upgrade_handler(
|
pub async fn ws_upgrade_handler(
|
||||||
axum::extract::State(registry): axum::extract::State<Arc<OperationRegistry>>,
|
axum::extract::State(registry): axum::extract::State<Arc<OperationRegistry>>,
|
||||||
axum::Extension(identity): axum::Extension<Identity>,
|
axum::Extension(identity): axum::Extension<Identity>,
|
||||||
|
policy: Option<axum::Extension<ChannelsPolicy>>,
|
||||||
ws_upgrade: WebSocketUpgrade,
|
ws_upgrade: WebSocketUpgrade,
|
||||||
) -> Response {
|
) -> Response {
|
||||||
|
let policy = policy
|
||||||
|
.map(|axum::Extension(p)| p.0)
|
||||||
|
.unwrap_or_else(|| Arc::new(NoCap));
|
||||||
ws_upgrade.on_upgrade(move |socket| async move {
|
ws_upgrade.on_upgrade(move |socket| async move {
|
||||||
run_channels_session(socket, registry, identity, Arc::new(NoCap)).await
|
run_channels_session(socket, registry, identity, policy).await
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,7 +1,7 @@
|
|||||||
---
|
---
|
||||||
id: review-001-hyper-server-knobs
|
id: review-001-hyper-server-knobs
|
||||||
name: Configure hyper timeouts + decoy/405/proxy-path fixes (SRV-04, SRV-07, SRV-08, SRV-09, SRV-10)
|
name: Configure hyper timeouts + decoy/405/proxy-path fixes (SRV-04, SRV-07, SRV-08, SRV-09, SRV-10)
|
||||||
status: pending
|
status: completed
|
||||||
depends_on: []
|
depends_on: []
|
||||||
scope: moderate
|
scope: moderate
|
||||||
risk: low
|
risk: low
|
||||||
@@ -44,13 +44,13 @@ and all land in `src/server/`:
|
|||||||
|
|
||||||
## Acceptance Criteria
|
## Acceptance Criteria
|
||||||
|
|
||||||
- [ ] Hyper configured with timer + header read timeout; knob values documented
|
- [x] Hyper configured with timer + header read timeout; knob values documented
|
||||||
- [ ] 405 responses carry the decoy `Server` header (test with `OPTIONS /search`)
|
- [x] 405 responses carry the decoy `Server` header (test with `OPTIONS /search`)
|
||||||
- [ ] `%C3%A9` and `a+b.html` resolve correctly in the decoy static server (tests); async-path syscalls gone
|
- [x] `%C3%A9` and `a+b.html` resolve correctly in the decoy static server (tests); async-path syscalls gone
|
||||||
- [ ] `/openapi.json` 500 body is generic; doc cached; `expect` removed
|
- [x] `/openapi.json` 500 body is generic; doc cached; `expect` removed
|
||||||
- [ ] Second `.with_decoy` no longer silently drops extra routes (test)
|
- [x] Second `.with_decoy` no longer silently drops extra routes (test)
|
||||||
- [ ] SRV-10: policy injection point exists or doc corrected; single token resolution
|
- [x] SRV-10: policy injection point exists or doc corrected; single token resolution
|
||||||
- [ ] `cargo test`, `cargo test --all-features`, `cargo clippy --all-targets -- -D warnings` pass
|
- [x] `cargo test`, `cargo test --all-features`, `cargo clippy --all-targets -- -D warnings` pass
|
||||||
|
|
||||||
## References
|
## References
|
||||||
|
|
||||||
@@ -64,4 +64,54 @@ and all land in `src/server/`:
|
|||||||
|
|
||||||
## Summary
|
## Summary
|
||||||
|
|
||||||
> Filled on completion.
|
Five server-core findings fixed in `src/server/` (+ two touchpoints):
|
||||||
|
|
||||||
|
- **SRV-04** (`server/adapter.rs`): `serve_io` now sets
|
||||||
|
`TokioTimer` on **both** the h1 and h2 auto-builder sub-builders,
|
||||||
|
`header_read_timeout(10 s)` (h1, bounds the slow-loris header-drip
|
||||||
|
window), h1 `keep_alive(true)`, h2 `keep_alive_interval(30 s)` +
|
||||||
|
`keep_alive_timeout(10 s)`. Verified against hyper 1.11 source: with
|
||||||
|
no timer, `Time::check` warns and returns `None` — the default 30 s
|
||||||
|
header timeout was a silent no-op. A module-doc section documents
|
||||||
|
the knobs and the concurrency boundary (no cap in this crate; the
|
||||||
|
accept loop is the consumer's — one stream per `handle` call).
|
||||||
|
- **SRV-05** (`server/adapter.rs`): `with_decoy` rebuilds the router
|
||||||
|
with `extra_routes.clone()` instead of `.take()` — a second builder
|
||||||
|
call keeps custom routes (test: `second_with_decoy_keeps_extra_routes`).
|
||||||
|
- **SRV-07** (`server/decoy.rs`, `server/adapter.rs`): wired
|
||||||
|
`Router::method_not_allowed_fallback(decoy_method_not_allowed)` —
|
||||||
|
method mismatches on registered paths (e.g. `OPTIONS /search`) get
|
||||||
|
the nginx-shaped 405 with `Server: nginx` instead of axum's bare
|
||||||
|
405 (axum 0.8 sets the default fallback on all previously registered
|
||||||
|
MethodRouters; the `Allow` header is unaffected).
|
||||||
|
- **SRV-08** (`server/decoy.rs`): `percent_decode` is now
|
||||||
|
byte-accumulating + single UTF-8 validation (`%C3%A9` → one `é`), `+`
|
||||||
|
is a literal (form-encoding, not path grammar), invalid escapes → `None`
|
||||||
|
→ fake 404. The blocking `is_dir()`/`is_file()` syscalls moved to
|
||||||
|
`tokio::fs::metadata`.
|
||||||
|
- **SRV-09** (`server/adapter.rs`, `adapters/to_openapi.rs`): the
|
||||||
|
serialized `/openapi.json` doc is built once at adapter construction
|
||||||
|
(`CachedOpenAPIDoc`, threaded through `RouterState` + `FromRef`);
|
||||||
|
the handler serves the cached bytes or a **generic** `internal server
|
||||||
|
error` 500 (no serde internals on the wire). `to_openapi` now returns
|
||||||
|
`Result<OpenAPISpec, AdapterError>` — the unguarded
|
||||||
|
`.expect("to_openapi always emits…")` is gone (project convention).
|
||||||
|
- **SRV-10** (`websocket/upgrade.rs`, `server/adapter.rs`): added the
|
||||||
|
injection point — `ChannelsPolicy(Arc<dyn ChannelLifecyclePolicy>)`
|
||||||
|
request extension read by `ws_upgrade_handler` (defaults to `NoCap`;
|
||||||
|
`run_channels_session` remains the direct route for custom upgrade
|
||||||
|
routes). The double token resolution (router-wide
|
||||||
|
`bearer_auth_middleware` + `ws_bearer_auth` both calling
|
||||||
|
`resolve_from_token`) was genuine and removed by route ordering: the
|
||||||
|
WS route registers before the router-wide auth route_layer
|
||||||
|
(route_layer covers only earlier-registered routes), so `/alk/channels`
|
||||||
|
resolves once with enforcement (401) while the gateway endpoints keep
|
||||||
|
the permissive single resolution. Verification: `ws_upgrade_session`
|
||||||
|
+ `ws_overlay_ops` integration suites pass (18 tests).
|
||||||
|
|
||||||
|
Verification: `cargo test` (265 passed), `cargo test --all-features`
|
||||||
|
(all `server::` + `to_openapi::` tests pass; the 7 remaining failures
|
||||||
|
and lint/format noise belong to other agents' in-flight
|
||||||
|
from_mcp/from_wss/forward work in the shared tree, not this task's
|
||||||
|
files), `cargo clippy` clean on all touched files, `cargo fmt --check`
|
||||||
|
clean on all touched files.
|
||||||
Reference in New Issue
Block a user