use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;
use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::routing::{delete, get, post};
use axum::{Json, Router};
use serde_json::json;
use subtle::ConstantTimeEq;
use tower::ServiceBuilder;
use crate::audit::{audit_log_middleware, AuditConfig, TrustedProxies};
use crate::http_metrics::{http_metrics_middleware, HttpMetricsLayerConfig, RouteAllowList};
use crate::middleware::{
bearer_auth, body_limit_layer, concurrency_limit_layer, cors_layer, host_validate,
tenant_scope, timeout_layer, trace_layer_with_propagation, AuthConfig, CorsConfig,
KernelPublishTokens, TenantConfig, TrustedHosts, ENV_API_TOKENS, ENV_TRUSTED_HOSTS,
INVOKE_CONCURRENCY_LIMIT, MAX_REQUEST_BODY_BYTES, PROBE_CONCURRENCY_LIMIT,
READ_CONCURRENCY_LIMIT, WRITE_CONCURRENCY_LIMIT,
};
use crate::rate_limit::{rate_limit, RateLimitConfig, RateLimiter};
use crate::routes::{
create_function, delete_function, get_job, healthz, invoke_function, invoke_function_async,
invoke_function_stream, metrics, snapshot_restore, snapshot_save, AppState,
};
use crate::trace_propagation::{inject_trace_id_header, install_w3c_propagator};
pub fn build_router(state: Arc<AppState>) -> Router {
let auth = AuthConfig::from_env();
let tenant = TenantConfig::from_env();
let limiter = RateLimiter::new(RateLimitConfig::from_env());
let audit = AuditConfig::from_env();
let cors = CorsConfig::from_env();
let state = apply_app_config_from_env(state);
build_router_with_audit(state, auth, tenant, limiter, audit, cors)
}
fn apply_app_config_from_env(state: Arc<AppState>) -> Arc<AppState> {
let cfg = crate::config::AppConfig::from_env().unwrap_or_else(|e| {
panic!(
"invalid snapshot configuration (TENSOR_WASM_API_SNAPSHOT_*): {e}; \
refusing to start with a misconfigured snapshot signing key",
);
});
match Arc::try_unwrap(state) {
Ok(inner) => Arc::new(inner.with_app_config(cfg)),
Err(shared) => {
Arc::new((*shared).clone().with_app_config(cfg))
}
}
}
pub fn build_router_with_config(
state: Arc<AppState>,
auth: AuthConfig,
tenant: TenantConfig,
) -> Router {
build_router_with_full_config(
state,
auth,
tenant,
RateLimiter::new(RateLimitConfig::disabled()),
)
}
pub fn build_router_with_full_config(
state: Arc<AppState>,
auth: AuthConfig,
tenant: TenantConfig,
limiter: RateLimiter,
) -> Router {
build_router_with_audit(
state,
auth,
tenant,
limiter,
AuditConfig::disabled(),
CorsConfig::default(),
)
}
pub fn build_router_with_audit(
state: Arc<AppState>,
auth: AuthConfig,
tenant: TenantConfig,
limiter: RateLimiter,
audit: AuditConfig,
cors: CorsConfig,
) -> Router {
build_router_with_trusted_proxies(
state,
auth,
tenant,
limiter,
audit,
cors,
TrustedProxies::from_env(),
)
}
pub fn build_router_with_trusted_proxies(
state: Arc<AppState>,
auth: AuthConfig,
tenant: TenantConfig,
limiter: RateLimiter,
audit: AuditConfig,
cors: CorsConfig,
trusted_proxies: TrustedProxies,
) -> Router {
build_router_full(
state,
auth,
tenant,
limiter,
audit,
cors,
trusted_proxies,
KernelPublishTokens::from_env(),
)
}
#[allow(clippy::too_many_arguments)]
pub fn build_router_with_kernel_publish_tokens(
state: Arc<AppState>,
auth: AuthConfig,
tenant: TenantConfig,
limiter: RateLimiter,
audit: AuditConfig,
cors: CorsConfig,
trusted_proxies: TrustedProxies,
kernel_publish_tokens: KernelPublishTokens,
) -> Router {
build_router_full(
state,
auth,
tenant,
limiter,
audit,
cors,
trusted_proxies,
kernel_publish_tokens,
)
}
static T16_HOST_WARN_FIRED: AtomicBool = AtomicBool::new(false);
pub const ENV_METRICS_TOKEN: &str = "TENSOR_WASM_API_METRICS_TOKEN";
static H3_METRICS_WARN_FIRED: AtomicBool = AtomicBool::new(false);
#[derive(Debug, Clone, Default)]
struct MetricsAuth {
token: Option<Arc<str>>,
}
impl MetricsAuth {
fn from_env() -> Self {
let raw = std::env::var(ENV_METRICS_TOKEN).unwrap_or_default();
let trimmed = raw.trim();
if trimmed.is_empty() {
if H3_METRICS_WARN_FIRED
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
tracing::warn!(
target: "tensor_wasm_api::server",
"GET /metrics is unauthenticated (TENSOR_WASM_API_METRICS_TOKEN \
unset) — it exposes per-tenant gauges and route/error \
distributions. This is the expected Prometheus-scrape \
posture behind an internal interface or ingress ACL; set \
TENSOR_WASM_API_METRICS_TOKEN=<token> to require \
`Authorization: Bearer <token>` on the scrape path.",
);
}
Self { token: None }
} else {
tracing::info!(
target: "tensor_wasm_api::server",
"GET /metrics requires a bearer token (TENSOR_WASM_API_METRICS_TOKEN set)",
);
Self {
token: Some(Arc::from(trimmed)),
}
}
}
}
#[doc(hidden)]
pub fn __reset_metrics_warn_for_test() {
H3_METRICS_WARN_FIRED.store(false, Ordering::Release);
}
async fn metrics_auth_gate(req: axum::extract::Request, next: axum::middleware::Next) -> Response {
let configured = req
.extensions()
.get::<MetricsAuth>()
.and_then(|m| m.token.clone());
let Some(expected) = configured else {
return next.run(req).await;
};
let presented = req
.headers()
.get(axum::http::header::AUTHORIZATION)
.and_then(|h| h.to_str().ok())
.and_then(parse_bearer_token);
let authorized = match presented {
Some(tok) => {
let a = tok.as_bytes();
let b = expected.as_bytes();
a.len() == b.len() && a.ct_eq(b).into()
}
None => false,
};
if !authorized {
return metrics_unauthorized();
}
next.run(req).await
}
fn metrics_unauthorized() -> Response {
(
StatusCode::UNAUTHORIZED,
Json(json!({
"error": {
"kind": "unauthorized",
"message": "GET /metrics requires a valid Authorization: Bearer <token> header",
}
})),
)
.into_response()
}
fn parse_bearer_token(value: &str) -> Option<&str> {
if value.bytes().any(|b| b != b'\t' && (b < 0x20 || b == 0x7F)) {
return None;
}
let split = value
.as_bytes()
.iter()
.position(|&b| b == b' ' || b == b'\t')?;
let (scheme, rest) = value.split_at(split);
if !scheme.eq_ignore_ascii_case("bearer") {
return None;
}
let token = rest.trim_matches(|c: char| c == ' ' || c == '\t');
if token.is_empty() {
None
} else {
Some(token)
}
}
fn maybe_warn_host_validation_disabled() {
let tokens_set = std::env::var(ENV_API_TOKENS).is_ok_and(|v| !v.is_empty());
if !tokens_set {
return;
}
let trusted_hosts_set =
std::env::var(ENV_TRUSTED_HOSTS).is_ok_and(|v| !TrustedHosts::from_raw(&v).is_empty());
if trusted_hosts_set {
return;
}
if T16_HOST_WARN_FIRED
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
tracing::warn!(
target: "tensor_wasm_api::server",
"TENSOR_WASM_API_TOKENS is set (production-mode) but \
TENSOR_WASM_API_TRUSTED_HOSTS is unset — Host header \
validation is disabled. This is OK behind an ingress that \
strips/validates Host, otherwise enable trusted hosts to \
avoid virtual-host confusion. Set \
TENSOR_WASM_API_TRUSTED_HOSTS=host1.example.com,host2.example.com \
to enable.",
);
}
}
#[doc(hidden)]
pub fn __reset_host_validation_warn_for_test() {
T16_HOST_WARN_FIRED.store(false, Ordering::Release);
}
#[allow(clippy::too_many_arguments)]
fn build_router_full(
state: Arc<AppState>,
auth: AuthConfig,
tenant: TenantConfig,
limiter: RateLimiter,
audit: AuditConfig,
cors: CorsConfig,
trusted_proxies: TrustedProxies,
kernel_publish_tokens: KernelPublishTokens,
) -> Router {
#[cfg(not(feature = "kernel-registry-api"))]
let _ = kernel_publish_tokens;
maybe_warn_host_validation_disabled();
install_w3c_propagator();
let http_metrics_cfg = HttpMetricsLayerConfig {
metrics: Arc::clone(&state.metrics),
routes: RouteAllowList::new_default(),
};
let common_layers = ServiceBuilder::new()
.layer(axum::Extension(TrustedHosts::from_env()))
.layer(axum::middleware::from_fn(host_validate))
.layer(trace_layer_with_propagation())
.layer(axum::middleware::from_fn(inject_trace_id_header))
.layer(cors_layer(&cors))
.layer(body_limit_layer(MAX_REQUEST_BODY_BYTES))
.layer(timeout_layer(Duration::from_secs(30)))
.layer(axum::Extension(auth))
.layer(axum::Extension(tenant))
.layer(axum::Extension(limiter))
.layer(axum::Extension(audit))
.layer(axum::Extension(trusted_proxies))
.layer(axum::Extension(http_metrics_cfg))
.layer(axum::middleware::from_fn(http_metrics_middleware));
let metrics_router = Router::new()
.route("/metrics", get(metrics))
.layer(axum::middleware::from_fn(metrics_auth_gate))
.layer(axum::Extension(MetricsAuth::from_env()));
let probe_router = Router::new()
.route("/healthz", get(healthz))
.merge(metrics_router)
.layer(concurrency_limit_layer(PROBE_CONCURRENCY_LIMIT));
let invoke_router = Router::new()
.route("/functions/:id/invoke", post(invoke_function))
.route("/functions/:id/invoke-async", post(invoke_function_async))
.route("/functions/:id/invoke-stream", post(invoke_function_stream))
.layer(concurrency_limit_layer(INVOKE_CONCURRENCY_LIMIT));
let write_router = Router::new()
.route("/functions", post(create_function))
.route("/functions/:id", delete(delete_function))
.route("/snapshot/save", post(snapshot_save))
.route("/snapshot/restore", post(snapshot_restore))
.layer(concurrency_limit_layer(WRITE_CONCURRENCY_LIMIT));
let read_router = Router::new()
.route("/jobs/:id", get(get_job))
.layer(concurrency_limit_layer(READ_CONCURRENCY_LIMIT));
let protected_router = invoke_router
.merge(write_router)
.merge(read_router)
.layer(axum::middleware::from_fn(audit_log_middleware))
.layer(axum::middleware::from_fn(rate_limit))
.layer(axum::middleware::from_fn(tenant_scope))
.layer(axum::middleware::from_fn(bearer_auth));
let openai_router: Router<Arc<AppState>> = Router::new()
.route("/v1/completions", post(crate::openai::completions_handler))
.route(
"/v1/chat/completions",
post(crate::openai::chat_completions_handler),
)
.layer(concurrency_limit_layer(INVOKE_CONCURRENCY_LIMIT))
.layer(axum::middleware::from_fn(audit_log_middleware))
.layer(axum::middleware::from_fn(rate_limit))
.layer(axum::middleware::from_fn(tenant_scope))
.layer(axum::middleware::from_fn(bearer_auth));
#[cfg(feature = "kernel-registry-api")]
let kernel_router: Router<Arc<AppState>> = Router::new()
.route(
"/kernels",
post(crate::kernels::publish_kernel).get(crate::kernels::list_kernels),
)
.route(
"/kernels/:name/:version",
get(crate::kernels::resolve_kernel),
)
.layer(concurrency_limit_layer(WRITE_CONCURRENCY_LIMIT))
.layer(axum::Extension(kernel_publish_tokens))
.layer(axum::middleware::from_fn(audit_log_middleware))
.layer(axum::middleware::from_fn(rate_limit))
.layer(axum::middleware::from_fn(tenant_scope))
.layer(axum::middleware::from_fn(bearer_auth));
let router = protected_router.merge(probe_router).merge(openai_router);
#[cfg(feature = "kernel-registry-api")]
let router = router.merge(kernel_router);
router.layer(common_layers).with_state(state)
}
pub async fn serve(state: Arc<AppState>, addr: SocketAddr) -> anyhow::Result<()> {
let router = build_router(state);
let listener = tokio::net::TcpListener::bind(addr).await?;
tracing::info!(target: "tensor_wasm_api::server", %addr, "tensor-wasm-api listening");
axum::serve(
listener,
router.into_make_service_with_connect_info::<SocketAddr>(),
)
.await?;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_bearer_token_accepts_canonical_scheme() {
assert_eq!(parse_bearer_token("Bearer abc123"), Some("abc123"));
assert_eq!(parse_bearer_token("bearer abc123"), Some("abc123"));
assert_eq!(parse_bearer_token("BEARER\tabc123"), Some("abc123"));
assert_eq!(parse_bearer_token("Bearer spaced "), Some("spaced"));
}
#[test]
fn parse_bearer_token_rejects_non_bearer_and_empty() {
assert_eq!(parse_bearer_token("Basic abc123"), None);
assert_eq!(parse_bearer_token("Bearer"), None);
assert_eq!(parse_bearer_token("Bearer "), None);
assert_eq!(parse_bearer_token(""), None);
}
#[test]
fn parse_bearer_token_rejects_control_bytes() {
assert_eq!(parse_bearer_token("Bearer ab\r\nc"), None);
assert_eq!(parse_bearer_token("Bearer ab\0c"), None);
}
#[test]
fn metrics_auth_default_is_unauthenticated() {
let auth = MetricsAuth::default();
assert!(auth.token.is_none());
}
#[test]
fn metrics_unauthorized_envelope_shape() {
let resp = metrics_unauthorized();
assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
}
}