commit 973a90f5f4bd8504b55e4abe49a916a1f19c5acf
parent 5242297bbd44383215223201cdee3c261ae160c0
Author: triesap <tyson@radroots.org>
Date: Tue, 11 Aug 2026 03:53:36 +0000
service-host: add bounded admin server
- serve HTTP/1.1 only over the owned Unix socket binding
- enforce strict mutation envelopes and bounded parameter routes
- cap request concurrency deadlines and response serialization
- preserve safe correlation and redaction across every failure path
Diffstat:
7 files changed, 2082 insertions(+), 4 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
@@ -2069,6 +2069,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
[[package]]
+name = "httpdate"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
+
+[[package]]
name = "hybrid-array"
version = "0.4.12"
source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -2092,6 +2098,7 @@ dependencies = [
"http",
"http-body",
"httparse",
+ "httpdate",
"itoa",
"pin-project-lite",
"smallvec",
@@ -3879,8 +3886,13 @@ dependencies = [
name = "radroots_service_host"
version = "0.1.0-alpha"
dependencies = [
+ "bytes",
"fs2",
"getrandom 0.2.17",
+ "http",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
"radroots_runtime_paths",
"rustix 1.1.4",
"serde",
diff --git a/Cargo.toml b/Cargo.toml
@@ -191,6 +191,7 @@ aes-gcm = { version = "0.10.3", default-features = false, features = [
async-wsocket = { version = "0.13.2" }
base64 = { version = "0.22", default-features = false, features = ["alloc"] }
bincode = { version = "1.3.3" }
+bytes = { version = "1" }
chacha20poly1305 = { version = "0.10.1", default-features = false, features = [
"alloc",
] }
@@ -207,6 +208,10 @@ fs2 = { version = "0.4" }
getrandom = { version = "0.2", default-features = false }
hkdf = { version = "0.12", default-features = false }
hmac = { version = "0.12", default-features = false }
+http = { version = "1" }
+http-body-util = { version = "0.1" }
+hyper = { version = "1", features = ["http1", "server"] }
+hyper-util = { version = "0.1", features = ["tokio"] }
hex = { version = "0.4" }
jiff-tzdb = { version = "=0.1.8", default-features = false }
jsonschema = { version = "0.48.1", default-features = false }
diff --git a/crates/service_host/Cargo.toml b/crates/service_host/Cargo.toml
@@ -12,18 +12,23 @@ homepage.workspace = true
readme = "README.md"
[dependencies]
+bytes = { workspace = true }
fs2 = { workspace = true }
getrandom = { workspace = true }
+http = { workspace = true }
+http-body-util = { workspace = true }
+hyper = { workspace = true }
+hyper-util = { workspace = true }
radroots_runtime_paths = { workspace = true }
rustix = { workspace = true }
serde = { workspace = true, features = ["derive", "std"] }
-serde_json = { workspace = true, features = ["std"] }
+serde_json = { workspace = true, features = ["raw_value", "std"] }
tokio = { workspace = true, features = ["macros", "net", "rt", "sync", "time"] }
tokio-util = { workspace = true }
[dev-dependencies]
tempfile = { workspace = true }
-tokio = { workspace = true, features = ["macros", "rt", "sync", "test-util", "time"] }
+tokio = { workspace = true, features = ["io-util", "macros", "rt", "sync", "test-util", "time"] }
[lints]
workspace = true
diff --git a/crates/service_host/src/admin/mod.rs b/crates/service_host/src/admin/mod.rs
@@ -3,6 +3,8 @@
mod limits;
mod model;
#[cfg(any(target_os = "linux", target_os = "macos"))]
+mod server;
+#[cfg(any(target_os = "linux", target_os = "macos"))]
mod unix;
pub use limits::{
@@ -18,6 +20,15 @@ pub use model::{
AdminSuccessResponse,
};
#[cfg(any(target_os = "linux", target_os = "macos"))]
+pub use server::{
+ ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES, ADMIN_ROUTE_PARAMETER_NAME_MAX_UTF8_BYTES,
+ ADMIN_ROUTE_PARAMETER_VALUE_MAX_UTF8_BYTES, ADMIN_ROUTE_PATH_MAX_UTF8_BYTES, AdminHttpMethod,
+ AdminRequest, AdminRequestDecodeError, AdminRouteFailure, AdminRouteFailureStatus,
+ AdminRouteOutcome, AdminRouteOutcomeError, AdminRoutePath, AdminRoutePathError,
+ AdminRouteRegistrationError, AdminRouter, AdminServer, AdminServerConfigError,
+ AdminServerError,
+};
+#[cfg(any(target_os = "linux", target_os = "macos"))]
pub use unix::{
UNIX_ADMIN_ACTIVE_PROBE_TIMEOUT, UNIX_ADMIN_OWNER_DIRECTORY_MODE, UNIX_ADMIN_OWNER_SOCKET_MODE,
UnixAdminSocketBinding, UnixAdminSocketError, UnixAdminSocketWriterAuthority,
diff --git a/crates/service_host/src/admin/server.rs b/crates/service_host/src/admin/server.rs
@@ -0,0 +1,2033 @@
+//! Bounded HTTP/1.1 administration server over an owned Unix listener.
+
+use core::fmt;
+use serde::{Deserialize, Deserializer, Serialize, de, de::DeserializeOwned};
+use serde_json::{Value, value::RawValue};
+use std::collections::{BTreeMap, BTreeSet};
+use std::convert::Infallible;
+use std::error::Error;
+use std::future::Future;
+use std::io;
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+
+use bytes::Bytes;
+use http::header::{CONTENT_TYPE, HeaderMap, HeaderValue};
+use http::{Method, Request, Response, StatusCode, Version};
+use http_body_util::{BodyExt, Full, LengthLimitError, Limited};
+use hyper::body::Incoming;
+use hyper::server::conn::http1;
+use hyper::service::service_fn;
+use hyper_util::rt::{TokioIo, TokioTimer};
+use tokio::sync::Semaphore;
+use tokio::task::JoinSet;
+
+use super::{
+ ADMIN_CONTRACT_VERSION, AdminCorrelationId, AdminError, AdminErrorCode, AdminErrorMessage,
+ AdminFailureResponse, AdminMutationRequest, AdminTransportLimits, UnixAdminSocketBinding,
+};
+use crate::{CancellationToken, EntropySource, SystemEntropy};
+
+/// Smallest configured response cap that can always carry a safe v1 failure envelope.
+pub const ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES: u32 = 512;
+
+pub const ADMIN_ROUTE_PATH_MAX_UTF8_BYTES: usize = 256;
+pub const ADMIN_ROUTE_PARAMETER_NAME_MAX_UTF8_BYTES: usize = 64;
+pub const ADMIN_ROUTE_PARAMETER_VALUE_MAX_UTF8_BYTES: usize = 128;
+const HTTP_MINIMUM_MAX_BUFFER_SIZE: usize = 8 * 1024;
+const JSON_CONTENT_TYPE: &str = "application/json";
+const FALLBACK_CORRELATION_ID: &str = "correlation-unavailable";
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub enum AdminHttpMethod {
+ Get,
+ Post,
+}
+
+impl AdminHttpMethod {
+ fn from_http(method: &Method) -> Option<Self> {
+ if method == Method::GET {
+ Some(Self::Get)
+ } else if method == Method::POST {
+ Some(Self::Post)
+ } else {
+ None
+ }
+ }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AdminRoutePathError {
+ Empty,
+ TooLong,
+ WrongVersionPrefix,
+ InvalidCharacter,
+ EmptySegment,
+ InvalidParameter,
+ DuplicateParameter,
+}
+
+impl fmt::Display for AdminRoutePathError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("admin route path is not a canonical v1 route")
+ }
+}
+
+impl Error for AdminRoutePathError {}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AdminRoutePath {
+ canonical: String,
+ segments: Vec<AdminRouteSegment>,
+ literal_count: usize,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum AdminRouteSegment {
+ Literal(String),
+ Parameter(String),
+}
+
+impl AdminRoutePath {
+ pub fn new(value: impl Into<String>) -> Result<Self, AdminRoutePathError> {
+ let value = value.into();
+ if value.is_empty() {
+ return Err(AdminRoutePathError::Empty);
+ }
+ if value.len() > ADMIN_ROUTE_PATH_MAX_UTF8_BYTES {
+ return Err(AdminRoutePathError::TooLong);
+ }
+ if !value.starts_with("/v1/") || value.ends_with('/') {
+ return Err(AdminRoutePathError::WrongVersionPrefix);
+ }
+ let mut parameter_names = BTreeSet::new();
+ let mut literal_count = 0;
+ let mut segments = Vec::new();
+ for segment in value[1..].split('/') {
+ if segment.is_empty() {
+ return Err(AdminRoutePathError::EmptySegment);
+ }
+ if let Some(parameter) = segment
+ .strip_prefix('{')
+ .and_then(|segment| segment.strip_suffix('}'))
+ {
+ if !valid_parameter_name(parameter) {
+ return Err(AdminRoutePathError::InvalidParameter);
+ }
+ if !parameter_names.insert(parameter.to_owned()) {
+ return Err(AdminRoutePathError::DuplicateParameter);
+ }
+ segments.push(AdminRouteSegment::Parameter(parameter.to_owned()));
+ } else {
+ if !segment
+ .bytes()
+ .all(|byte| matches!(byte, b'a'..=b'z' | b'0'..=b'9' | b'_' | b'-'))
+ {
+ return Err(AdminRoutePathError::InvalidCharacter);
+ }
+ literal_count += 1;
+ segments.push(AdminRouteSegment::Literal(segment.to_owned()));
+ }
+ }
+ Ok(Self {
+ canonical: value,
+ segments,
+ literal_count,
+ })
+ }
+
+ #[must_use]
+ pub fn as_str(&self) -> &str {
+ &self.canonical
+ }
+
+ fn overlaps(&self, other: &Self) -> bool {
+ self.segments.len() == other.segments.len()
+ && self
+ .segments
+ .iter()
+ .zip(&other.segments)
+ .all(|(left, right)| match (left, right) {
+ (AdminRouteSegment::Literal(left), AdminRouteSegment::Literal(right)) => {
+ left == right
+ }
+ _ => true,
+ })
+ }
+
+ fn match_path(&self, path: &str) -> Option<BTreeMap<String, String>> {
+ if !path.starts_with('/') || path.ends_with('/') || path.contains("//") {
+ return None;
+ }
+ let raw_segments = path[1..].split('/').collect::<Vec<_>>();
+ if raw_segments.len() != self.segments.len() {
+ return None;
+ }
+ let mut parameters = BTreeMap::new();
+ for (pattern, raw) in self.segments.iter().zip(raw_segments) {
+ match pattern {
+ AdminRouteSegment::Literal(literal) if literal != raw => return None,
+ AdminRouteSegment::Literal(_) => {}
+ AdminRouteSegment::Parameter(name) => {
+ parameters.insert(name.clone(), decode_route_parameter(raw)?);
+ }
+ }
+ }
+ Some(parameters)
+ }
+}
+
+fn valid_parameter_name(value: &str) -> bool {
+ value.len() <= ADMIN_ROUTE_PARAMETER_NAME_MAX_UTF8_BYTES
+ && matches!(value.as_bytes().first(), Some(b'a'..=b'z'))
+ && value
+ .bytes()
+ .all(|byte| matches!(byte, b'a'..=b'z' | b'0'..=b'9' | b'_'))
+}
+
+fn decode_route_parameter(raw: &str) -> Option<String> {
+ if raw.is_empty() {
+ return None;
+ }
+ let raw = raw.as_bytes();
+ let mut decoded = Vec::with_capacity(raw.len().min(ADMIN_ROUTE_PARAMETER_VALUE_MAX_UTF8_BYTES));
+ let mut index = 0;
+ while index < raw.len() {
+ let byte = if raw[index] == b'%' {
+ let high = *raw.get(index + 1)?;
+ let low = *raw.get(index + 2)?;
+ index += 3;
+ (hex_nibble(high)? << 4) | hex_nibble(low)?
+ } else {
+ let byte = raw[index];
+ index += 1;
+ byte
+ };
+ if byte == b'/' || byte == b'\\' || byte == 0 || byte.is_ascii_control() {
+ return None;
+ }
+ if decoded.len() == ADMIN_ROUTE_PARAMETER_VALUE_MAX_UTF8_BYTES {
+ return None;
+ }
+ decoded.push(byte);
+ }
+ String::from_utf8(decoded)
+ .ok()
+ .filter(|value| !value.is_empty())
+}
+
+fn hex_nibble(value: u8) -> Option<u8> {
+ match value {
+ b'0'..=b'9' => Some(value - b'0'),
+ b'a'..=b'f' => Some(value - b'a' + 10),
+ b'A'..=b'F' => Some(value - b'A' + 10),
+ _ => None,
+ }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AdminRouteRegistrationError {
+ InvalidPath(AdminRoutePathError),
+ Duplicate,
+ Ambiguous,
+}
+
+impl fmt::Display for AdminRouteRegistrationError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("admin route registration is invalid")
+ }
+}
+
+impl Error for AdminRouteRegistrationError {}
+
+pub enum AdminRequestDecodeError {
+ Empty,
+ Malformed(serde_json::Error),
+}
+
+impl fmt::Debug for AdminRequestDecodeError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(match self {
+ Self::Empty => "AdminRequestDecodeError::Empty",
+ Self::Malformed(_) => "AdminRequestDecodeError::Malformed(<redacted>)",
+ })
+ }
+}
+
+impl fmt::Display for AdminRequestDecodeError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("admin request body is not valid JSON")
+ }
+}
+
+impl Error for AdminRequestDecodeError {
+ fn source(&self) -> Option<&(dyn Error + 'static)> {
+ match self {
+ Self::Empty => None,
+ Self::Malformed(error) => Some(error),
+ }
+ }
+}
+
+/// Validated request metadata and a bounded raw JSON body.
+pub struct AdminRequest {
+ method: AdminHttpMethod,
+ path: AdminRoutePath,
+ query: Option<String>,
+ correlation_id: AdminCorrelationId,
+ parameters: BTreeMap<String, String>,
+ body: Bytes,
+ response_body_limit: usize,
+}
+
+impl AdminRequest {
+ #[must_use]
+ pub const fn method(&self) -> AdminHttpMethod {
+ self.method
+ }
+
+ #[must_use]
+ pub const fn path(&self) -> &AdminRoutePath {
+ &self.path
+ }
+
+ #[must_use]
+ pub fn query(&self) -> Option<&str> {
+ self.query.as_deref()
+ }
+
+ #[must_use]
+ pub const fn correlation_id(&self) -> &AdminCorrelationId {
+ &self.correlation_id
+ }
+
+ #[must_use]
+ pub fn parameter(&self, name: &str) -> Option<&str> {
+ self.parameters.get(name).map(String::as_str)
+ }
+
+ #[must_use]
+ pub fn body(&self) -> &[u8] {
+ &self.body
+ }
+
+ pub fn decode_json<T>(&self) -> Result<T, AdminRequestDecodeError>
+ where
+ T: DeserializeOwned,
+ {
+ if self.body.is_empty() {
+ return Err(AdminRequestDecodeError::Empty);
+ }
+ serde_json::from_slice(&self.body).map_err(AdminRequestDecodeError::Malformed)
+ }
+
+ pub fn success<T>(&self, result: &T) -> Result<AdminRouteOutcome, AdminRouteOutcomeError>
+ where
+ T: Serialize,
+ {
+ let encoded =
+ encode_bounded(result, self.response_body_limit).map_err(|error| match error {
+ BoundedEncodingError::Limit => AdminRouteOutcomeError::ResponseLimit,
+ BoundedEncodingError::Encoding(error) => AdminRouteOutcomeError::Encoding(error),
+ })?;
+ let _: StrictJsonValue =
+ serde_json::from_slice(&encoded).map_err(AdminRouteOutcomeError::InvalidPayload)?;
+ let encoded =
+ String::from_utf8(encoded).expect("serde_json output must always be valid UTF-8");
+ let result = RawValue::from_string(encoded).map_err(AdminRouteOutcomeError::Encoding)?;
+ Ok(AdminRouteOutcome(AdminRouteOutcomeKind::Success(result)))
+ }
+}
+
+impl fmt::Debug for AdminRequest {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("AdminRequest")
+ .field("method", &self.method)
+ .field("path", &self.path)
+ .field("query", &self.query.as_ref().map(|_| "[redacted]"))
+ .field("correlation_id", &self.correlation_id)
+ .field(
+ "parameter_names",
+ &self.parameters.keys().collect::<Vec<_>>(),
+ )
+ .field("body", &"[redacted]")
+ .finish()
+ }
+}
+
+pub enum AdminRouteOutcomeError {
+ Encoding(serde_json::Error),
+ InvalidPayload(serde_json::Error),
+ ResponseLimit,
+}
+
+impl fmt::Debug for AdminRouteOutcomeError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(match self {
+ Self::Encoding(_) => "AdminRouteOutcomeError::Encoding(<redacted>)",
+ Self::InvalidPayload(_) => "AdminRouteOutcomeError::InvalidPayload(<redacted>)",
+ Self::ResponseLimit => "AdminRouteOutcomeError::ResponseLimit",
+ })
+ }
+}
+
+impl fmt::Display for AdminRouteOutcomeError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str(match self {
+ Self::Encoding(_) => "admin route result could not be encoded",
+ Self::InvalidPayload(_) => "admin route result violates the JSON payload contract",
+ Self::ResponseLimit => "admin route result exceeds the response limit",
+ })
+ }
+}
+
+impl Error for AdminRouteOutcomeError {
+ fn source(&self) -> Option<&(dyn Error + 'static)> {
+ match self {
+ Self::Encoding(error) | Self::InvalidPayload(error) => Some(error),
+ Self::ResponseLimit => None,
+ }
+ }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AdminRouteFailureStatus {
+ BadRequest,
+ NotFound,
+ Conflict,
+ Unavailable,
+ Internal,
+}
+
+impl AdminRouteFailureStatus {
+ const fn http_status(self) -> StatusCode {
+ match self {
+ Self::BadRequest => StatusCode::BAD_REQUEST,
+ Self::NotFound => StatusCode::NOT_FOUND,
+ Self::Conflict => StatusCode::CONFLICT,
+ Self::Unavailable => StatusCode::SERVICE_UNAVAILABLE,
+ Self::Internal => StatusCode::INTERNAL_SERVER_ERROR,
+ }
+ }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct AdminRouteFailure {
+ status: AdminRouteFailureStatus,
+ error: AdminError,
+}
+
+impl AdminRouteFailure {
+ #[must_use]
+ pub const fn new(status: AdminRouteFailureStatus, error: AdminError) -> Self {
+ Self { status, error }
+ }
+
+ #[must_use]
+ pub const fn status(&self) -> AdminRouteFailureStatus {
+ self.status
+ }
+
+ #[must_use]
+ pub const fn error(&self) -> &AdminError {
+ &self.error
+ }
+}
+
+enum AdminRouteOutcomeKind {
+ Success(Box<RawValue>),
+ Failure(AdminRouteFailure),
+}
+
+pub struct AdminRouteOutcome(AdminRouteOutcomeKind);
+
+impl AdminRouteOutcome {
+ #[must_use]
+ pub const fn failure(failure: AdminRouteFailure) -> Self {
+ Self(AdminRouteOutcomeKind::Failure(failure))
+ }
+}
+
+impl fmt::Debug for AdminRouteOutcome {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ match &self.0 {
+ AdminRouteOutcomeKind::Success(_) => {
+ formatter.write_str("AdminRouteOutcome::Success(<redacted>)")
+ }
+ AdminRouteOutcomeKind::Failure(failure) => {
+ formatter.debug_tuple("Failure").field(failure).finish()
+ }
+ }
+ }
+}
+
+#[derive(Serialize)]
+struct ServerSuccessEnvelope<'a> {
+ contract_version: u32,
+ ok: bool,
+ correlation_id: &'a AdminCorrelationId,
+ result: &'a RawValue,
+}
+
+struct StrictJsonValue(Value);
+
+impl Serialize for StrictJsonValue {
+ fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+ where
+ S: serde::Serializer,
+ {
+ self.0.serialize(serializer)
+ }
+}
+
+impl<'de> Deserialize<'de> for StrictJsonValue {
+ fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+ where
+ D: Deserializer<'de>,
+ {
+ deserializer.deserialize_any(StrictJsonVisitor)
+ }
+}
+
+struct StrictJsonVisitor;
+
+impl<'de> de::Visitor<'de> for StrictJsonVisitor {
+ type Value = StrictJsonValue;
+
+ fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("JSON without duplicate object keys or null values")
+ }
+
+ fn visit_bool<E>(self, value: bool) -> Result<Self::Value, E> {
+ Ok(StrictJsonValue(Value::Bool(value)))
+ }
+
+ fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> {
+ Ok(StrictJsonValue(Value::Number(value.into())))
+ }
+
+ fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
+ Ok(StrictJsonValue(Value::Number(value.into())))
+ }
+
+ fn visit_f64<E>(self, value: f64) -> Result<Self::Value, E>
+ where
+ E: de::Error,
+ {
+ serde_json::Number::from_f64(value)
+ .map(Value::Number)
+ .map(StrictJsonValue)
+ .ok_or_else(|| E::custom("non-finite JSON number"))
+ }
+
+ fn visit_str<E>(self, value: &str) -> Result<Self::Value, E> {
+ Ok(StrictJsonValue(Value::String(value.to_owned())))
+ }
+
+ fn visit_borrowed_str<E>(self, value: &'de str) -> Result<Self::Value, E>
+ where
+ E: de::Error,
+ {
+ self.visit_str(value)
+ }
+
+ fn visit_string<E>(self, value: String) -> Result<Self::Value, E> {
+ Ok(StrictJsonValue(Value::String(value)))
+ }
+
+ fn visit_none<E>(self) -> Result<Self::Value, E>
+ where
+ E: de::Error,
+ {
+ Err(E::custom("JSON null is forbidden"))
+ }
+
+ fn visit_unit<E>(self) -> Result<Self::Value, E>
+ where
+ E: de::Error,
+ {
+ Err(E::custom("JSON null is forbidden"))
+ }
+
+ fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
+ where
+ A: de::SeqAccess<'de>,
+ {
+ let mut values = Vec::with_capacity(sequence.size_hint().unwrap_or(0));
+ while let Some(value) = sequence.next_element::<StrictJsonValue>()? {
+ values.push(value.0);
+ }
+ Ok(StrictJsonValue(Value::Array(values)))
+ }
+
+ fn visit_map<A>(self, mut map: A) -> Result<Self::Value, A::Error>
+ where
+ A: de::MapAccess<'de>,
+ {
+ let mut values = serde_json::Map::with_capacity(map.size_hint().unwrap_or(0));
+ while let Some(key) = map.next_key::<String>()? {
+ if values.contains_key(&key) {
+ return Err(de::Error::custom("duplicate JSON object key"));
+ }
+ let value = map.next_value::<StrictJsonValue>()?;
+ values.insert(key, value.0);
+ }
+ Ok(StrictJsonValue(Value::Object(values)))
+ }
+}
+
+type AdminRouteFuture = Pin<Box<dyn Future<Output = AdminRouteOutcome> + Send + 'static>>;
+
+trait AdminRouteHandler: Send + Sync {
+ fn handle(&self, request: AdminRequest) -> AdminRouteFuture;
+}
+
+struct FunctionRouteHandler<F>(F);
+
+impl<F, Fut> AdminRouteHandler for FunctionRouteHandler<F>
+where
+ F: Fn(AdminRequest) -> Fut + Send + Sync + 'static,
+ Fut: Future<Output = AdminRouteOutcome> + Send + 'static,
+{
+ fn handle(&self, request: AdminRequest) -> AdminRouteFuture {
+ Box::pin((self.0)(request))
+ }
+}
+
+struct RegisteredRoute {
+ method: AdminHttpMethod,
+ path: AdminRoutePath,
+ handler: Arc<dyn AdminRouteHandler>,
+}
+
+/// Method/path registry supplied by the consuming service.
+#[derive(Default)]
+pub struct AdminRouter {
+ routes: Vec<RegisteredRoute>,
+}
+
+impl AdminRouter {
+ #[must_use]
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ pub fn route<F, Fut>(
+ &mut self,
+ method: AdminHttpMethod,
+ path: impl Into<String>,
+ handler: F,
+ ) -> Result<(), AdminRouteRegistrationError>
+ where
+ F: Fn(AdminRequest) -> Fut + Send + Sync + 'static,
+ Fut: Future<Output = AdminRouteOutcome> + Send + 'static,
+ {
+ let path = AdminRoutePath::new(path).map_err(AdminRouteRegistrationError::InvalidPath)?;
+ if self
+ .routes
+ .iter()
+ .any(|route| route.method == method && route.path == path)
+ {
+ return Err(AdminRouteRegistrationError::Duplicate);
+ }
+ if self.routes.iter().any(|route| {
+ route.method == method
+ && route.path.literal_count == path.literal_count
+ && route.path.overlaps(&path)
+ }) {
+ return Err(AdminRouteRegistrationError::Ambiguous);
+ }
+ self.routes.push(RegisteredRoute {
+ method,
+ path,
+ handler: Arc::new(FunctionRouteHandler(handler)),
+ });
+ Ok(())
+ }
+}
+
+impl fmt::Debug for AdminRouter {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("AdminRouter")
+ .field("route_count", &self.routes.len())
+ .finish()
+ }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AdminServerConfigError {
+ NoRoutes,
+ ResponseLimitTooSmall,
+}
+
+impl fmt::Display for AdminServerConfigError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("admin server configuration is invalid")
+ }
+}
+
+impl Error for AdminServerConfigError {}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AdminServerError {
+ ListenerClone { kind: io::ErrorKind },
+ ListenerRegistration { kind: io::ErrorKind },
+ Accept { kind: io::ErrorKind },
+ ConnectionTaskPanicked,
+}
+
+impl fmt::Display for AdminServerError {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter.write_str("bounded admin server failed")
+ }
+}
+
+impl Error for AdminServerError {}
+
+struct AdminServerState {
+ routes: Vec<RegisteredRoute>,
+ limits: AdminTransportLimits,
+ entropy: Arc<dyn EntropySource>,
+}
+
+/// One bounded HTTP/1.1 server that can only consume a Unix admin binding.
+pub struct AdminServer {
+ state: Arc<AdminServerState>,
+}
+
+impl fmt::Debug for AdminServer {
+ fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+ formatter
+ .debug_struct("AdminServer")
+ .field("route_count", &self.state.routes.len())
+ .field("limits", &self.state.limits)
+ .field("entropy", &"[injected]")
+ .finish()
+ }
+}
+
+impl AdminServer {
+ pub fn new<E>(
+ router: AdminRouter,
+ limits: AdminTransportLimits,
+ entropy: E,
+ ) -> Result<Self, AdminServerConfigError>
+ where
+ E: EntropySource + 'static,
+ {
+ if router.routes.is_empty() {
+ return Err(AdminServerConfigError::NoRoutes);
+ }
+ if limits.response_body_utf8_bytes() < ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES {
+ return Err(AdminServerConfigError::ResponseLimitTooSmall);
+ }
+ Ok(Self {
+ state: Arc::new(AdminServerState {
+ routes: router.routes,
+ limits,
+ entropy: Arc::new(entropy),
+ }),
+ })
+ }
+
+ pub fn with_system_entropy(
+ router: AdminRouter,
+ limits: AdminTransportLimits,
+ ) -> Result<Self, AdminServerConfigError> {
+ Self::new(router, limits, SystemEntropy)
+ }
+
+ /// Serves until cancellation, then stops admission and drains every bounded connection task.
+ pub async fn serve(
+ self,
+ binding: UnixAdminSocketBinding,
+ cancellation: CancellationToken,
+ ) -> Result<(), AdminServerError> {
+ let listener = binding
+ .listener()
+ .try_clone()
+ .map_err(|error| AdminServerError::ListenerClone { kind: error.kind() })?;
+ listener
+ .set_nonblocking(true)
+ .map_err(|error| AdminServerError::ListenerRegistration { kind: error.kind() })?;
+ let listener = tokio::net::UnixListener::from_std(listener)
+ .map_err(|error| AdminServerError::ListenerRegistration { kind: error.kind() })?;
+ let permits = Arc::new(Semaphore::new(
+ self.state.limits.concurrent_connections() as usize
+ ));
+ let mut tasks = JoinSet::new();
+
+ let mut result = loop {
+ tokio::select! {
+ biased;
+ () = cancellation.cancelled() => break Ok(()),
+ joined = tasks.join_next(), if !tasks.is_empty() => {
+ if joined.is_some_and(|result| result.is_err()) {
+ cancellation.cancel();
+ break Err(AdminServerError::ConnectionTaskPanicked);
+ }
+ }
+ accepted = listener.accept() => {
+ let (stream, _) = match accepted {
+ Ok(accepted) => accepted,
+ Err(error) => {
+ cancellation.cancel();
+ break Err(AdminServerError::Accept { kind: error.kind() });
+ }
+ };
+ let Ok(permit) = Arc::clone(&permits).try_acquire_owned() else {
+ drop(stream);
+ continue;
+ };
+ let state = Arc::clone(&self.state);
+ let connection_cancellation = cancellation.clone();
+ tasks.spawn(async move {
+ let _permit = permit;
+ serve_connection(stream, state, connection_cancellation).await;
+ });
+ }
+ }
+ };
+
+ while let Some(joined) = tasks.join_next().await {
+ if joined.is_err() {
+ result = Err(AdminServerError::ConnectionTaskPanicked);
+ }
+ }
+ drop(binding);
+ result
+ }
+}
+
+async fn serve_connection(
+ stream: tokio::net::UnixStream,
+ state: Arc<AdminServerState>,
+ cancellation: CancellationToken,
+) {
+ let service_state = Arc::clone(&state);
+ let service = service_fn(move |request| {
+ let state = Arc::clone(&service_state);
+ async move { Ok::<_, Infallible>(serve_request(request, state).await) }
+ });
+
+ let mut builder = http1::Builder::new();
+ builder
+ .keep_alive(false)
+ .auto_date_header(false)
+ .max_headers(state.limits.header_count() as usize)
+ .max_buf_size((state.limits.header_bytes() as usize).max(HTTP_MINIMUM_MAX_BUFFER_SIZE))
+ .header_read_timeout(state.limits.idle_timeout())
+ .timer(TokioTimer::new());
+
+ let mut connection = Box::pin(builder.serve_connection(TokioIo::new(stream), service));
+ let connection_deadline = state
+ .limits
+ .request_deadline()
+ .checked_add(state.limits.idle_timeout())
+ .unwrap_or(state.limits.idle_timeout());
+ tokio::select! {
+ biased;
+ () = cancellation.cancelled() => {
+ connection.as_mut().graceful_shutdown();
+ let _ = tokio::time::timeout(connection_deadline, connection).await;
+ }
+ _ = tokio::time::timeout(connection_deadline, &mut connection) => {}
+ }
+}
+
+async fn serve_request(
+ request: Request<Incoming>,
+ state: Arc<AdminServerState>,
+) -> Response<Full<Bytes>> {
+ let (initial_correlation, deferred_entropy_failure) =
+ match generated_correlation(&*state.entropy) {
+ Ok(correlation) => (correlation, None),
+ Err(error) => (fallback_correlation(), Some(error)),
+ };
+ let correlation = Arc::new(CorrelationSlot::new(initial_correlation));
+ let deadline = state.limits.request_deadline();
+ match tokio::time::timeout(
+ deadline,
+ process_request(
+ request,
+ Arc::clone(&state),
+ Arc::clone(&correlation),
+ deferred_entropy_failure,
+ ),
+ )
+ .await
+ {
+ Ok(response) => response,
+ Err(_) => failure_response(
+ StatusCode::GATEWAY_TIMEOUT,
+ correlation.current(),
+ known_error("request_timeout", "admin request deadline elapsed"),
+ state.limits,
+ ),
+ }
+}
+
+async fn process_request(
+ request: Request<Incoming>,
+ state: Arc<AdminServerState>,
+ correlation: Arc<CorrelationSlot>,
+ deferred_entropy_failure: Option<AdminError>,
+) -> Response<Full<Bytes>> {
+ let correlation_id = correlation.current();
+ if request.version() != Version::HTTP_11 {
+ return failure_response(
+ StatusCode::HTTP_VERSION_NOT_SUPPORTED,
+ correlation_id,
+ known_error(
+ "http_version_unsupported",
+ "admin transport requires HTTP/1.1",
+ ),
+ state.limits,
+ );
+ }
+ if header_bytes(request.headers()) > u64::from(state.limits.header_bytes())
+ || request.headers().len() > state.limits.header_count() as usize
+ {
+ return failure_response(
+ StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE,
+ correlation_id,
+ known_error(
+ "request_headers_too_large",
+ "admin request headers exceed the limit",
+ ),
+ state.limits,
+ );
+ }
+ if query_item_count(request.uri().query()) > state.limits.query_items() as usize {
+ return failure_response(
+ StatusCode::BAD_REQUEST,
+ correlation_id,
+ known_error("query_limit_exceeded", "admin query exceeds the item limit"),
+ state.limits,
+ );
+ }
+
+ if unknown_major_version(request.uri().path()) {
+ let response = AdminFailureResponse::unsupported_contract_version(correlation_id.clone());
+ return bounded_json_response(
+ StatusCode::BAD_REQUEST,
+ &response,
+ correlation_id,
+ state.limits,
+ );
+ }
+ let Some(method) = AdminHttpMethod::from_http(request.method()) else {
+ return failure_response(
+ StatusCode::NOT_FOUND,
+ correlation_id,
+ known_error("route_not_found", "admin route was not found"),
+ state.limits,
+ );
+ };
+ let route = state
+ .routes
+ .iter()
+ .filter(|route| route.method == method)
+ .filter_map(|route| {
+ route
+ .path
+ .match_path(request.uri().path())
+ .map(|parameters| {
+ (
+ route.path.literal_count,
+ route.path.clone(),
+ parameters,
+ Arc::clone(&route.handler),
+ )
+ })
+ })
+ .max_by_key(|(literal_count, _, _, _)| *literal_count);
+ let Some((_, path, parameters, handler)) = route else {
+ return failure_response(
+ StatusCode::NOT_FOUND,
+ correlation_id,
+ known_error("route_not_found", "admin route was not found"),
+ state.limits,
+ );
+ };
+
+ if method == AdminHttpMethod::Post && !is_json_content_type(request.headers()) {
+ return failure_response(
+ StatusCode::UNSUPPORTED_MEDIA_TYPE,
+ correlation_id,
+ known_error(
+ "content_type_required",
+ "admin mutation requires JSON content type",
+ ),
+ state.limits,
+ );
+ }
+ let query = request.uri().query().map(str::to_owned);
+ let body = match Limited::new(
+ request.into_body(),
+ state.limits.request_body_utf8_bytes() as usize,
+ )
+ .collect()
+ .await
+ {
+ Ok(collected) => collected.to_bytes(),
+ Err(error) if error.downcast_ref::<LengthLimitError>().is_some() => {
+ return failure_response(
+ StatusCode::PAYLOAD_TOO_LARGE,
+ correlation_id,
+ known_error(
+ "request_body_too_large",
+ "admin request body exceeds the limit",
+ ),
+ state.limits,
+ );
+ }
+ Err(_) => {
+ return failure_response(
+ StatusCode::BAD_REQUEST,
+ correlation_id,
+ known_error(
+ "request_body_unavailable",
+ "admin request body is unavailable",
+ ),
+ state.limits,
+ );
+ }
+ };
+ if method == AdminHttpMethod::Get && !body.is_empty() {
+ return failure_response(
+ StatusCode::BAD_REQUEST,
+ correlation_id,
+ known_error("malformed_json", "admin request body is not valid JSON"),
+ state.limits,
+ );
+ }
+
+ if method == AdminHttpMethod::Post {
+ let envelope = match serde_json::from_slice::<AdminMutationRequest<StrictJsonValue>>(&body)
+ {
+ Ok(envelope) => envelope,
+ Err(_) => {
+ return failure_response(
+ StatusCode::BAD_REQUEST,
+ correlation.current(),
+ known_error("malformed_json", "admin request body is not valid JSON"),
+ state.limits,
+ );
+ }
+ };
+ if let Some(caller_correlation) = envelope.correlation_id().cloned() {
+ correlation.replace(caller_correlation);
+ } else if let Some(error) = deferred_entropy_failure {
+ return failure_response(
+ StatusCode::INTERNAL_SERVER_ERROR,
+ correlation.current(),
+ error,
+ state.limits,
+ );
+ }
+ if envelope.validate_contract_version().is_err() {
+ let correlation_id = correlation.current();
+ let response =
+ AdminFailureResponse::unsupported_contract_version(correlation_id.clone());
+ return bounded_json_response(
+ StatusCode::BAD_REQUEST,
+ &response,
+ correlation_id,
+ state.limits,
+ );
+ }
+ } else if let Some(error) = deferred_entropy_failure {
+ return failure_response(
+ StatusCode::INTERNAL_SERVER_ERROR,
+ correlation.current(),
+ error,
+ state.limits,
+ );
+ }
+ let correlation_id = correlation.current();
+
+ let request = AdminRequest {
+ method,
+ path,
+ query,
+ correlation_id: correlation_id.clone(),
+ parameters,
+ body,
+ response_body_limit: state.limits.response_body_utf8_bytes() as usize,
+ };
+ match handler.handle(request).await.0 {
+ AdminRouteOutcomeKind::Success(result) => {
+ let envelope = ServerSuccessEnvelope {
+ contract_version: ADMIN_CONTRACT_VERSION,
+ ok: true,
+ correlation_id: &correlation_id,
+ result: &result,
+ };
+ bounded_json_response(
+ StatusCode::OK,
+ &envelope,
+ correlation_id.clone(),
+ state.limits,
+ )
+ }
+ AdminRouteOutcomeKind::Failure(failure) => failure_response(
+ failure.status.http_status(),
+ correlation_id,
+ failure.error,
+ state.limits,
+ ),
+ }
+}
+
+fn generated_correlation(entropy: &dyn EntropySource) -> Result<AdminCorrelationId, AdminError> {
+ let mut bytes = [0_u8; 16];
+ if entropy.fill_bytes(&mut bytes).is_err() {
+ return Err(known_error(
+ "correlation_unavailable",
+ "admin correlation identity is unavailable",
+ ));
+ }
+ let value = bytes
+ .iter()
+ .map(|byte| format!("{byte:02x}"))
+ .collect::<String>();
+ Ok(AdminCorrelationId::new(value).expect("hex correlation must be valid"))
+}
+
+fn fallback_correlation() -> AdminCorrelationId {
+ AdminCorrelationId::new(FALLBACK_CORRELATION_ID)
+ .expect("static correlation fallback must be valid")
+}
+
+fn unknown_major_version(path: &str) -> bool {
+ path.strip_prefix('/')
+ .and_then(|path| path.split('/').next())
+ .is_some_and(|segment| {
+ segment != "v1"
+ && segment.strip_prefix('v').is_some_and(|version| {
+ !version.is_empty() && version.bytes().all(|byte| byte.is_ascii_digit())
+ })
+ })
+}
+
+struct CorrelationSlot(Mutex<AdminCorrelationId>);
+
+impl CorrelationSlot {
+ fn new(correlation_id: AdminCorrelationId) -> Self {
+ Self(Mutex::new(correlation_id))
+ }
+
+ fn current(&self) -> AdminCorrelationId {
+ self.0
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner)
+ .clone()
+ }
+
+ fn replace(&self, correlation_id: AdminCorrelationId) {
+ *self
+ .0
+ .lock()
+ .unwrap_or_else(std::sync::PoisonError::into_inner) = correlation_id;
+ }
+}
+
+fn header_bytes(headers: &HeaderMap) -> u64 {
+ headers.iter().fold(0_u64, |total, (name, value)| {
+ total
+ .saturating_add(name.as_str().len() as u64)
+ .saturating_add(value.as_bytes().len() as u64)
+ })
+}
+
+fn query_item_count(query: Option<&str>) -> usize {
+ match query {
+ None | Some("") => 0,
+ Some(query) => query.split('&').count(),
+ }
+}
+
+fn is_json_content_type(headers: &HeaderMap) -> bool {
+ headers
+ .get(CONTENT_TYPE)
+ .and_then(|value| value.to_str().ok())
+ .is_some_and(|value| {
+ value == JSON_CONTENT_TYPE || value == "application/json; charset=utf-8"
+ })
+}
+
+fn failure_response(
+ status: StatusCode,
+ correlation_id: AdminCorrelationId,
+ error: AdminError,
+ limits: AdminTransportLimits,
+) -> Response<Full<Bytes>> {
+ let envelope = AdminFailureResponse::new(correlation_id.clone(), error);
+ bounded_json_response(status, &envelope, correlation_id, limits)
+}
+
+fn bounded_json_response<T>(
+ status: StatusCode,
+ value: &T,
+ correlation_id: AdminCorrelationId,
+ limits: AdminTransportLimits,
+) -> Response<Full<Bytes>>
+where
+ T: Serialize,
+{
+ let limit = limits.response_body_utf8_bytes() as usize;
+ if let Ok(encoded) = encode_bounded(value, limit) {
+ return json_response(status, encoded);
+ }
+ let fallback = AdminFailureResponse::new(
+ correlation_id,
+ known_error(
+ "response_body_too_large",
+ "admin response body exceeds the limit",
+ ),
+ );
+ match encode_bounded(&fallback, limit) {
+ Ok(encoded) => json_response(StatusCode::INTERNAL_SERVER_ERROR, encoded),
+ Err(_) => json_response(
+ StatusCode::INTERNAL_SERVER_ERROR,
+ br#"{"contract_version":1,"ok":false,"correlation_id":"correlation-unavailable","error":{"code":"response_body_too_large","message":"admin response body exceeds the limit"}}"#.to_vec(),
+ ),
+ }
+}
+
+enum BoundedEncodingError {
+ Limit,
+ Encoding(serde_json::Error),
+}
+
+struct CappedWriter {
+ bytes: Vec<u8>,
+ limit: usize,
+ exceeded: bool,
+}
+
+impl CappedWriter {
+ fn new(limit: usize) -> Self {
+ Self {
+ bytes: Vec::with_capacity(limit.min(4096)),
+ limit,
+ exceeded: false,
+ }
+ }
+}
+
+impl io::Write for CappedWriter {
+ fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+ if bytes.len() > self.limit.saturating_sub(self.bytes.len()) {
+ self.exceeded = true;
+ return Err(io::Error::other("bounded JSON output limit exceeded"));
+ }
+ self.bytes.extend_from_slice(bytes);
+ Ok(bytes.len())
+ }
+
+ fn flush(&mut self) -> io::Result<()> {
+ Ok(())
+ }
+}
+
+fn encode_bounded<T>(value: &T, limit: usize) -> Result<Vec<u8>, BoundedEncodingError>
+where
+ T: Serialize,
+{
+ let mut writer = CappedWriter::new(limit);
+ match serde_json::to_writer(&mut writer, value) {
+ Ok(()) => Ok(writer.bytes),
+ Err(_) if writer.exceeded => Err(BoundedEncodingError::Limit),
+ Err(error) => Err(BoundedEncodingError::Encoding(error)),
+ }
+}
+
+fn json_response(status: StatusCode, body: Vec<u8>) -> Response<Full<Bytes>> {
+ let mut response = Response::new(Full::new(Bytes::from(body)));
+ *response.status_mut() = status;
+ response
+ .headers_mut()
+ .insert(CONTENT_TYPE, HeaderValue::from_static(JSON_CONTENT_TYPE));
+ response
+}
+
+fn known_error(code: &'static str, message: &'static str) -> AdminError {
+ AdminError::new(
+ AdminErrorCode::new(code).expect("static admin error code must be valid"),
+ AdminErrorMessage::new(message).expect("static admin error message must be valid"),
+ )
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use serde::{Deserialize, Serialize};
+ use std::sync::atomic::{AtomicUsize, Ordering};
+ use tokio::io::{AsyncReadExt, AsyncWriteExt};
+ use tokio::sync::Notify;
+
+ #[derive(Clone, Copy)]
+ struct FixedEntropy(u8);
+
+ impl EntropySource for FixedEntropy {
+ fn fill_bytes(&self, destination: &mut [u8]) -> Result<(), crate::EntropyError> {
+ destination.fill(self.0);
+ Ok(())
+ }
+ }
+
+ struct FailingEntropy;
+
+ impl EntropySource for FailingEntropy {
+ fn fill_bytes(&self, _destination: &mut [u8]) -> Result<(), crate::EntropyError> {
+ Err(crate::EntropyError::Unavailable)
+ }
+ }
+
+ struct CountingFailingEntropy(Arc<AtomicUsize>);
+
+ impl EntropySource for CountingFailingEntropy {
+ fn fill_bytes(&self, _destination: &mut [u8]) -> Result<(), crate::EntropyError> {
+ self.0.fetch_add(1, Ordering::SeqCst);
+ Err(crate::EntropyError::Unavailable)
+ }
+ }
+
+ #[derive(Debug, Deserialize, Serialize)]
+ #[serde(deny_unknown_fields)]
+ struct EchoRequest {
+ value: String,
+ }
+
+ fn echo_router() -> AdminRouter {
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Post, "/v1/echo", |request| async move {
+ match request.decode_json::<AdminMutationRequest<EchoRequest>>() {
+ Ok(envelope) => {
+ let request_body = envelope.into_request();
+ request
+ .success(&serde_json::json!({"value": request_body.value}))
+ .expect("echo result should encode")
+ }
+ Err(_) => AdminRouteOutcome::failure(AdminRouteFailure::new(
+ AdminRouteFailureStatus::BadRequest,
+ known_error("malformed_echo", "echo request is malformed"),
+ )),
+ }
+ })
+ .expect("echo route");
+ router
+ }
+
+ fn limits_with(
+ request_body: u32,
+ concurrency: u32,
+ deadline: std::time::Duration,
+ ) -> AdminTransportLimits {
+ let mut values = AdminTransportLimits::DEFAULT.values();
+ values.request_body_utf8_bytes = request_body;
+ values.concurrent_connections = concurrency;
+ values.request_deadline = deadline;
+ AdminTransportLimits::new(values).expect("test limits")
+ }
+
+ async fn binding(
+ directory: &tempfile::TempDir,
+ ) -> (std::path::PathBuf, UnixAdminSocketBinding) {
+ let socket = directory.path().join("admin.sock");
+ let authority = super::super::UnixAdminSocketWriterAuthority::acquire(directory.path())
+ .expect("writer authority");
+ let binding = UnixAdminSocketBinding::bind(authority, &socket)
+ .await
+ .expect("socket binding");
+ (socket, binding)
+ }
+
+ async fn exchange(socket: &std::path::Path, request: &[u8]) -> Vec<u8> {
+ let mut stream = tokio::net::UnixStream::connect(socket)
+ .await
+ .expect("connect admin server");
+ stream.write_all(request).await.expect("write request");
+ let mut response = Vec::new();
+ stream
+ .read_to_end(&mut response)
+ .await
+ .expect("read response");
+ response
+ }
+
+ #[tokio::test]
+ async fn serves_valid_json_with_exact_caller_correlation_and_no_web_headers() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let server = AdminServer::new(
+ echo_router(),
+ AdminTransportLimits::DEFAULT,
+ FixedEntropy(7),
+ )
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task_cancellation = cancellation.clone();
+ let task = tokio::spawn(server.serve(binding, task_cancellation));
+
+ let body = r#"{"contract_version":1,"operation_id":"echo-01","correlation_id":"caller-01","request":{"value":"ok"}}"#;
+ let request = format!(
+ "POST /v1/echo HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let response = exchange(&socket, request.as_bytes()).await;
+ let text = String::from_utf8(response).expect("UTF-8 response");
+ assert!(text.starts_with("HTTP/1.1 200 OK\r\n"), "{text}");
+ assert!(text.contains("content-type: application/json"));
+ assert!(!text.to_ascii_lowercase().contains("access-control-allow"));
+ assert!(text.ends_with(
+ r#"{"contract_version":1,"ok":true,"correlation_id":"caller-01","result":{"value":"ok"}}"#
+ ));
+
+ let unsupported =
+ exchange(&socket, b"GET /v2/status HTTP/1.1\r\nHost: local\r\n\r\n").await;
+ let unsupported = String::from_utf8(unsupported).expect("unsupported response");
+ assert!(unsupported.starts_with("HTTP/1.1 400 Bad Request\r\n"));
+ assert!(unsupported.contains("unsupported_contract_version"));
+ assert!(unsupported.contains("07070707070707070707070707070707"));
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ assert!(!socket.exists());
+ }
+
+ #[tokio::test]
+ async fn rejects_oversized_and_malformed_json_before_the_handler() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let calls = Arc::new(AtomicUsize::new(0));
+ let handler_calls = Arc::clone(&calls);
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Post, "/v1/test", move |request| {
+ let calls = Arc::clone(&handler_calls);
+ async move {
+ calls.fetch_add(1, Ordering::SeqCst);
+ request
+ .success(&serde_json::json!({"ok": true}))
+ .expect("test result")
+ }
+ })
+ .expect("test route");
+ let server = AdminServer::new(
+ router,
+ limits_with(8, 4, std::time::Duration::from_secs(1)),
+ FixedEntropy(8),
+ )
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+
+ let oversized = exchange(
+ &socket,
+ b"POST /v1/test HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: 9\r\n\r\n123456789",
+ )
+ .await;
+ assert!(
+ String::from_utf8(oversized)
+ .expect("response")
+ .contains("request_body_too_large")
+ );
+ let malformed = exchange(
+ &socket,
+ b"POST /v1/test HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: 1\r\n\r\n{",
+ )
+ .await;
+ assert!(
+ String::from_utf8(malformed)
+ .expect("response")
+ .contains("malformed_json")
+ );
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn rejects_invalid_mutation_envelopes_duplicates_and_nested_null_before_dispatch() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let calls = Arc::new(AtomicUsize::new(0));
+ let handler_calls = Arc::clone(&calls);
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Post, "/v1/test", move |request| {
+ let calls = Arc::clone(&handler_calls);
+ async move {
+ calls.fetch_add(1, Ordering::SeqCst);
+ request.success(&true).expect("test result")
+ }
+ })
+ .expect("test route");
+ let server = AdminServer::new(router, AdminTransportLimits::DEFAULT, FixedEntropy(0x21))
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+
+ for body in [
+ r#"{"operation_id":"missing-version","request":{"value":true}}"#,
+ r#"{"contract_version":1,"operation_id":"missing-request"}"#,
+ r#"{"contract_version":1,"operation_id":"duplicate-correlation","correlation_id":"first","correlation_id":"second","request":{"value":true}}"#,
+ r#"{"contract_version":1,"operation_id":"nested-duplicate","request":{"value":true,"value":false}}"#,
+ r#"{"contract_version":1,"operation_id":"nested-null","request":{"nested":[true,null]}}"#,
+ ] {
+ let request = format!(
+ "POST /v1/test HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let response = String::from_utf8(exchange(&socket, request.as_bytes()).await)
+ .expect("invalid envelope response");
+ assert!(response.contains("malformed_json"), "{response}");
+ }
+
+ let body = r#"{"contract_version":2,"operation_id":"future-version","correlation_id":"caller-v2","request":{"value":true}}"#;
+ let request = format!(
+ "POST /v1/test HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let response = String::from_utf8(exchange(&socket, request.as_bytes()).await)
+ .expect("version response");
+ assert!(response.contains("unsupported_contract_version"));
+ assert!(response.contains("caller-v2"));
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn parameterized_routes_percent_decode_bounded_values_without_service_authority() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let calls = Arc::new(AtomicUsize::new(0));
+ let handler_calls = Arc::clone(&calls);
+ let mut router = AdminRouter::new();
+ router
+ .route(
+ AdminHttpMethod::Post,
+ "/v1/connections/{connection_id}/approve",
+ move |request| {
+ let calls = Arc::clone(&handler_calls);
+ async move {
+ calls.fetch_add(1, Ordering::SeqCst);
+ let parameter = request
+ .parameter("connection_id")
+ .expect("decoded route parameter")
+ .to_owned();
+ request
+ .success(&serde_json::json!({"connection_id": parameter}))
+ .expect("parameter response")
+ }
+ },
+ )
+ .expect("parameterized route");
+ let server = AdminServer::new(router, AdminTransportLimits::DEFAULT, FixedEntropy(0x22))
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+ let body =
+ r#"{"contract_version":1,"operation_id":"approve-01","request":{"approve":true}}"#;
+
+ let request = format!(
+ "POST /v1/connections/farm%2D01/approve HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let response = String::from_utf8(exchange(&socket, request.as_bytes()).await)
+ .expect("parameter response");
+ assert!(response.starts_with("HTTP/1.1 200 OK"), "{response}");
+ assert!(response.contains(r#""connection_id":"farm-01""#));
+
+ for path in [
+ "/v1/connections/farm%2/approve",
+ "/v1/connections/farm%2F01/approve",
+ &format!(
+ "/v1/connections/{}/approve",
+ "x".repeat(ADMIN_ROUTE_PARAMETER_VALUE_MAX_UTF8_BYTES + 1)
+ ),
+ ] {
+ let request = format!(
+ "POST {path} HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let response = String::from_utf8(exchange(&socket, request.as_bytes()).await)
+ .expect("invalid parameter response");
+ assert!(response.contains("route_not_found"), "{response}");
+ }
+ assert_eq!(calls.load(Ordering::SeqCst), 1);
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn caller_correlation_precedes_entropy_and_survives_timeout_handoff() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let entropy_calls = Arc::new(AtomicUsize::new(0));
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Post, "/v1/slow", |request| async move {
+ tokio::time::sleep(std::time::Duration::from_secs(60)).await;
+ request.success(&true).expect("slow result")
+ })
+ .expect("slow route");
+ let server = AdminServer::new(
+ router,
+ limits_with(1024, 4, std::time::Duration::from_millis(20)),
+ CountingFailingEntropy(Arc::clone(&entropy_calls)),
+ )
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+
+ let body = r#"{"contract_version":1,"operation_id":"slow-01","correlation_id":"caller-timeout","request":{"wait":true}}"#;
+ let request = format!(
+ "POST /v1/slow HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let response = String::from_utf8(exchange(&socket, request.as_bytes()).await)
+ .expect("timeout response");
+ assert!(response.starts_with("HTTP/1.1 504 Gateway Timeout"));
+ assert!(response.contains("caller-timeout"));
+ assert_eq!(entropy_calls.load(Ordering::SeqCst), 1);
+
+ let body = r#"{"contract_version":1,"operation_id":"slow-02","request":{"wait":true}}"#;
+ let request = format!(
+ "POST /v1/slow HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let response = String::from_utf8(exchange(&socket, request.as_bytes()).await)
+ .expect("entropy response");
+ assert!(response.contains("correlation_unavailable"));
+ assert_eq!(entropy_calls.load(Ordering::SeqCst), 2);
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn rejects_http_1_0_before_dispatch() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let calls = Arc::new(AtomicUsize::new(0));
+ let handler_calls = Arc::clone(&calls);
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Get, "/v1/status", move |request| {
+ let calls = Arc::clone(&handler_calls);
+ async move {
+ calls.fetch_add(1, Ordering::SeqCst);
+ request.success(&true).expect("status result")
+ }
+ })
+ .expect("status route");
+ let server = AdminServer::new(router, AdminTransportLimits::DEFAULT, FixedEntropy(0x23))
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+
+ let response = String::from_utf8(
+ exchange(&socket, b"GET /v1/status HTTP/1.0\r\nHost: local\r\n\r\n").await,
+ )
+ .expect("HTTP/1.0 response");
+ assert!(response.contains("505 HTTP Version Not Supported"));
+ assert!(response.contains("http_version_unsupported"));
+ assert_eq!(calls.load(Ordering::SeqCst), 0);
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn request_deadline_returns_a_safe_timeout_and_cancels_the_handler_future() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Get, "/v1/slow", |request| async move {
+ tokio::time::sleep(std::time::Duration::from_secs(60)).await;
+ request
+ .success(&serde_json::json!({"late": true}))
+ .expect("late result")
+ })
+ .expect("slow route");
+ let server = AdminServer::new(
+ router,
+ limits_with(1024, 4, std::time::Duration::from_millis(20)),
+ FixedEntropy(9),
+ )
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+
+ let response = exchange(&socket, b"GET /v1/slow HTTP/1.1\r\nHost: local\r\n\r\n").await;
+ let text = String::from_utf8(response).expect("response");
+ assert!(text.starts_with("HTTP/1.1 504 Gateway Timeout\r\n"));
+ assert!(text.contains("request_timeout"));
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn enforces_header_query_response_and_body_correlation_boundaries() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let calls = Arc::new(AtomicUsize::new(0));
+ let handler_calls = Arc::clone(&calls);
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Get, "/v1/large", move |request| {
+ let calls = Arc::clone(&handler_calls);
+ async move {
+ calls.fetch_add(1, Ordering::SeqCst);
+ match request.success(&"x".repeat(1024)) {
+ Ok(outcome) => outcome,
+ Err(AdminRouteOutcomeError::ResponseLimit) => {
+ AdminRouteOutcome::failure(AdminRouteFailure::new(
+ AdminRouteFailureStatus::Internal,
+ known_error(
+ "response_body_too_large",
+ "admin response body exceeds the limit",
+ ),
+ ))
+ }
+ Err(_) => panic!("large result should only exceed the cap"),
+ }
+ }
+ })
+ .expect("large route");
+ router
+ .route(AdminHttpMethod::Post, "/v1/input", |request| async move {
+ request.success(&true).expect("input result")
+ })
+ .expect("input route");
+ let mut values = AdminTransportLimits::DEFAULT.values();
+ values.header_bytes = 64;
+ values.query_items = 1;
+ values.response_body_utf8_bytes = ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES;
+ let limits = AdminTransportLimits::new(values).expect("boundary limits");
+ let server = AdminServer::new(router, limits, FixedEntropy(11)).expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+
+ let headers = exchange(
+ &socket,
+ b"GET /v1/large HTTP/1.1\r\nHost: local\r\nX-Large: 0123456789012345678901234567890123456789012345678901234567890123456789\r\n\r\n",
+ )
+ .await;
+ assert!(
+ String::from_utf8(headers)
+ .expect("header response")
+ .contains("request_headers_too_large")
+ );
+
+ let query = exchange(
+ &socket,
+ b"GET /v1/large?a=1&b=2 HTTP/1.1\r\nHost: local\r\n\r\n",
+ )
+ .await;
+ assert!(
+ String::from_utf8(query)
+ .expect("query response")
+ .contains("query_limit_exceeded")
+ );
+
+ let response = exchange(&socket, b"GET /v1/large HTTP/1.1\r\nHost: local\r\n\r\n").await;
+ let response = String::from_utf8(response).expect("bounded response");
+ assert!(response.contains("response_body_too_large"));
+ assert!(response.len() < ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES as usize + 256);
+ assert_eq!(calls.load(Ordering::SeqCst), 1);
+
+ let body = r#"{"contract_version":1,"operation_id":"input-01","correlation_id":null,"request":{"value":true}}"#;
+ let invalid_correlation = format!(
+ "POST /v1/input HTTP/1.1\r\nHost: local\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{body}",
+ body.len()
+ );
+ let invalid_correlation = exchange(&socket, invalid_correlation.as_bytes()).await;
+ assert!(
+ String::from_utf8(invalid_correlation)
+ .expect("correlation response")
+ .contains("malformed_json")
+ );
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn connection_admission_never_exceeds_the_configured_limit() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let entered = Arc::new(Notify::new());
+ let release = Arc::new(Notify::new());
+ let handler_entered = Arc::clone(&entered);
+ let handler_release = Arc::clone(&release);
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Get, "/v1/block", move |request| {
+ let entered = Arc::clone(&handler_entered);
+ let release = Arc::clone(&handler_release);
+ async move {
+ entered.notify_one();
+ release.notified().await;
+ request
+ .success(&serde_json::json!({"done": true}))
+ .expect("blocking result")
+ }
+ })
+ .expect("blocking route");
+ let server = AdminServer::new(
+ router,
+ limits_with(1024, 1, std::time::Duration::from_secs(1)),
+ FixedEntropy(10),
+ )
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let task = tokio::spawn(server.serve(binding, cancellation.clone()));
+
+ let first_socket = socket.clone();
+ let first = tokio::spawn(async move {
+ exchange(
+ &first_socket,
+ b"GET /v1/block HTTP/1.1\r\nHost: local\r\n\r\n",
+ )
+ .await
+ });
+ entered.notified().await;
+ let second = tokio::time::timeout(
+ std::time::Duration::from_millis(200),
+ exchange(&socket, b"GET /v1/block HTTP/1.1\r\nHost: local\r\n\r\n"),
+ )
+ .await
+ .expect("second connection must close without waiting");
+ assert!(
+ second.is_empty(),
+ "excess connection must receive no response"
+ );
+ release.notify_one();
+ assert!(
+ String::from_utf8(first.await.expect("first task"))
+ .expect("first response")
+ .starts_with("HTTP/1.1 200 OK")
+ );
+
+ cancellation.cancel();
+ task.await.expect("server task").expect("server shutdown");
+ }
+
+ #[tokio::test]
+ async fn graceful_cancellation_stops_admission_and_drains_an_active_request() {
+ let directory = tempfile::tempdir().expect("runtime directory");
+ let (socket, binding) = binding(&directory).await;
+ let entered = Arc::new(Notify::new());
+ let release = Arc::new(Notify::new());
+ let handler_entered = Arc::clone(&entered);
+ let handler_release = Arc::clone(&release);
+ let mut router = AdminRouter::new();
+ router
+ .route(AdminHttpMethod::Get, "/v1/drain", move |request| {
+ let entered = Arc::clone(&handler_entered);
+ let release = Arc::clone(&handler_release);
+ async move {
+ entered.notify_one();
+ release.notified().await;
+ request.success(&true).expect("drained result")
+ }
+ })
+ .expect("drain route");
+ let server = AdminServer::new(
+ router,
+ limits_with(1024, 1, std::time::Duration::from_secs(1)),
+ FixedEntropy(12),
+ )
+ .expect("admin server");
+ let cancellation = CancellationToken::new();
+ let mut task = tokio::spawn(server.serve(binding, cancellation.clone()));
+ let client_socket = socket.clone();
+ let client = tokio::spawn(async move {
+ exchange(
+ &client_socket,
+ b"GET /v1/drain HTTP/1.1\r\nHost: local\r\n\r\n",
+ )
+ .await
+ });
+
+ entered.notified().await;
+ cancellation.cancel();
+ assert!(
+ tokio::time::timeout(std::time::Duration::from_millis(20), &mut task)
+ .await
+ .is_err(),
+ "graceful cancellation must retain active request work"
+ );
+ release.notify_one();
+ assert!(
+ String::from_utf8(client.await.expect("client task"))
+ .expect("drained response")
+ .starts_with("HTTP/1.1 200 OK")
+ );
+ task.await.expect("server task").expect("server shutdown");
+ assert!(!socket.exists());
+ }
+
+ #[test]
+ fn route_and_server_configuration_fail_closed() {
+ let mut router = AdminRouter::new();
+ assert!(matches!(
+ router.route(AdminHttpMethod::Get, "/v2/status", |request| async move {
+ request.success(&true).expect("bool")
+ }),
+ Err(AdminRouteRegistrationError::InvalidPath(
+ AdminRoutePathError::WrongVersionPrefix
+ ))
+ ));
+ router
+ .route(AdminHttpMethod::Get, "/v1/status", |request| async move {
+ request.success(&true).expect("bool")
+ })
+ .expect("first route");
+ assert_eq!(
+ router
+ .route(AdminHttpMethod::Get, "/v1/status", |request| async move {
+ request.success(&true).expect("bool")
+ })
+ .expect_err("duplicate route"),
+ AdminRouteRegistrationError::Duplicate
+ );
+
+ let mut values = AdminTransportLimits::DEFAULT.values();
+ values.response_body_utf8_bytes = ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES - 1;
+ let limits = AdminTransportLimits::new(values).expect("small response limit");
+ assert_eq!(
+ AdminServer::new(router, limits, FixedEntropy(1)).expect_err("small response cap"),
+ AdminServerConfigError::ResponseLimitTooSmall
+ );
+ assert_eq!(ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES, 512);
+ let longest_correlation = AdminCorrelationId::new("x".repeat(128)).expect("correlation");
+ let fallback = AdminFailureResponse::new(
+ longest_correlation,
+ known_error(
+ "response_body_too_large",
+ "admin response body exceeds the limit",
+ ),
+ );
+ assert!(encode_bounded(&fallback, ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES as usize).is_ok());
+
+ let mut parameterized = AdminRouter::new();
+ parameterized
+ .route(
+ AdminHttpMethod::Get,
+ "/v1/items/{item_id}",
+ |request| async move { request.success(&true).expect("parameterized result") },
+ )
+ .expect("parameterized route");
+ parameterized
+ .route(
+ AdminHttpMethod::Get,
+ "/v1/items/status",
+ |request| async move { request.success(&true).expect("static result") },
+ )
+ .expect("more-specific static route");
+ assert_eq!(
+ parameterized
+ .route(
+ AdminHttpMethod::Get,
+ "/v1/items/{other_id}",
+ |request| async move { request.success(&true).expect("ambiguous result") }
+ )
+ .expect_err("ambiguous route"),
+ AdminRouteRegistrationError::Ambiguous
+ );
+ assert!(matches!(
+ AdminRoutePath::new("/v1/items/{Bad}"),
+ Err(AdminRoutePathError::InvalidParameter)
+ ));
+ }
+
+ #[test]
+ fn capped_encoding_and_result_validation_fail_before_unbounded_response_allocation() {
+ let oversized = "x".repeat(ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES as usize + 1);
+ assert!(matches!(
+ encode_bounded(&oversized, ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES as usize),
+ Err(BoundedEncodingError::Limit)
+ ));
+
+ let mut writer = CappedWriter::new(8);
+ assert!(io::Write::write_all(&mut writer, b"123456789").is_err());
+ assert!(writer.exceeded);
+ assert!(writer.bytes.len() <= 8);
+
+ let request = AdminRequest {
+ method: AdminHttpMethod::Get,
+ path: AdminRoutePath::new("/v1/test").expect("path"),
+ query: None,
+ correlation_id: AdminCorrelationId::new("safe-id").expect("correlation"),
+ parameters: BTreeMap::new(),
+ body: Bytes::new(),
+ response_body_limit: ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES as usize,
+ };
+ assert!(matches!(
+ request.success(&Option::<bool>::None),
+ Err(AdminRouteOutcomeError::InvalidPayload(_))
+ ));
+ }
+
+ #[test]
+ fn generated_correlation_is_exact_and_entropy_failure_is_safe() {
+ let correlation = generated_correlation(&FixedEntropy(0xab)).expect("fixed entropy");
+ assert_eq!(correlation.as_str(), "abababababababababababababababab");
+
+ let failure = generated_correlation(&FailingEntropy).expect_err("entropy failure");
+ assert_eq!(failure.code().as_str(), "correlation_unavailable");
+ assert_eq!(fallback_correlation().as_str(), FALLBACK_CORRELATION_ID);
+ }
+
+ #[test]
+ fn package_source_has_no_forbidden_listener_or_web_authority() {
+ let source = include_str!("server.rs");
+ for forbidden in [
+ concat!("Tcp", "Listener"),
+ concat!("Access-Control", "-Allow"),
+ concat!("co", "rs"),
+ concat!("browser", "_auth"),
+ ] {
+ assert!(
+ !source.contains(forbidden),
+ "forbidden server authority: {forbidden}"
+ );
+ }
+ }
+
+ #[test]
+ fn request_and_outcome_debug_redact_body_query_and_result() {
+ let request = AdminRequest {
+ method: AdminHttpMethod::Post,
+ path: AdminRoutePath::new("/v1/test").expect("path"),
+ query: Some("secret=query".to_owned()),
+ correlation_id: AdminCorrelationId::new("safe-id").expect("correlation"),
+ parameters: BTreeMap::from([("item_id".to_owned(), "secret-item".to_owned())]),
+ body: Bytes::from_static(b"{\"secret\":true}"),
+ response_body_limit: 1024,
+ };
+ let debug = format!("{request:?}");
+ assert!(!debug.contains("secret=query"));
+ assert!(!debug.contains("secret\":true"));
+ assert!(!debug.contains("secret-item"));
+ let outcome = request
+ .success(&serde_json::json!({"secret": true}))
+ .expect("outcome");
+ assert!(!format!("{outcome:?}").contains("secret"));
+ }
+}
diff --git a/crates/service_host/src/lib.rs b/crates/service_host/src/lib.rs
@@ -21,8 +21,14 @@ pub use admin::{
};
#[cfg(any(target_os = "linux", target_os = "macos"))]
pub use admin::{
- UNIX_ADMIN_ACTIVE_PROBE_TIMEOUT, UNIX_ADMIN_OWNER_DIRECTORY_MODE, UNIX_ADMIN_OWNER_SOCKET_MODE,
- UnixAdminSocketBinding, UnixAdminSocketError, UnixAdminSocketWriterAuthority,
+ ADMIN_MIN_RESPONSE_BODY_UTF8_BYTES, ADMIN_ROUTE_PARAMETER_NAME_MAX_UTF8_BYTES,
+ ADMIN_ROUTE_PARAMETER_VALUE_MAX_UTF8_BYTES, ADMIN_ROUTE_PATH_MAX_UTF8_BYTES, AdminHttpMethod,
+ AdminRequest, AdminRequestDecodeError, AdminRouteFailure, AdminRouteFailureStatus,
+ AdminRouteOutcome, AdminRouteOutcomeError, AdminRoutePath, AdminRoutePathError,
+ AdminRouteRegistrationError, AdminRouter, AdminServer, AdminServerConfigError,
+ AdminServerError, UNIX_ADMIN_ACTIVE_PROBE_TIMEOUT, UNIX_ADMIN_OWNER_DIRECTORY_MODE,
+ UNIX_ADMIN_OWNER_SOCKET_MODE, UnixAdminSocketBinding, UnixAdminSocketError,
+ UnixAdminSocketWriterAuthority,
};
pub use build_info::{
BuildInfo, BuildInfoEnvironment, BuildInfoError, BuildInfoField, BuildMode, ContractVersions,
diff --git a/crates/service_host/tests/package_boundary.rs b/crates/service_host/tests/package_boundary.rs
@@ -6,6 +6,7 @@ const ADMIN_SOURCE: &str = concat!(
include_str!("../src/admin/mod.rs"),
include_str!("../src/admin/limits.rs"),
include_str!("../src/admin/model.rs"),
+ include_str!("../src/admin/server.rs"),
include_str!("../src/admin/unix.rs"),
);
const STATUS_SOURCE: &str = concat!(
@@ -40,8 +41,13 @@ fn service_host_is_unpublished_lint_governed_and_dependency_bounded() {
assert_eq!(
dependency_keys(MANIFEST),
BTreeSet::from([
+ "bytes",
"fs2",
"getrandom",
+ "http",
+ "http-body-util",
+ "hyper",
+ "hyper-util",
"radroots_runtime_paths",
"rustix",
"serde",
