use std::collections::BTreeMap;
use std::sync::Arc;
use axum::Router;
use axum::extract::{OriginalUri, State};
use axum::http::{HeaderName, HeaderValue, Method, StatusCode};
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use bytes::Bytes;
use crate::admin::require_admin;
use crate::config::Config;
use crate::events::EventBus;
use crate::http::{Json, MAX_BODY_BYTES, X_REQUEST_ID};
use crate::module::HARNESS_API;
use crate::ports::{Clock, Dispatcher, Kid, Payload, RateLimiter, Signer};
use crate::problem::Problem;
use crate::problems::SLUGS;
use crate::rate_limit::{
RateLimit, RateLimitFailure, check_rate_limit, client_ip, rate_limit_keys,
};
use crate::scope::Scope;
use crate::signer::{HmacSigner, KeyRing, MIN_SECRET_BYTES, TokenPolicy};
pub const HARNESS_SIDECARS: &str = "HARNESS_SIDECARS";
pub const X_HARNESS_API: &str = "x-harness-api";
pub const X_HARNESS_MODULE: &str = "x-harness-module";
pub const SIDECAR_GATEWAY_SECRET: &str = "SIDECAR_GATEWAY_SECRET";
pub const SIDECAR_REQUIRE_GATEWAY: &str = "SIDECAR_REQUIRE_GATEWAY";
pub const HARNESS_ONE_WORKER: &str = "HARNESS_ONE_WORKER";
pub const X_HARNESS_GATEWAY: &str = "x-harness-gateway";
pub const GATEWAY_PURPOSE: &str = "sidecar-gateway";
pub const GATEWAY_ADMIN_PURPOSE: &str = "sidecar-gateway-admin";
pub const GATEWAY_TOKEN_TTL_SECS: i64 = 120;
const FORWARDED_REQUEST_HEADERS: &[&str] = &[
"content-type",
"content-length",
"accept",
"accept-language",
"user-agent",
];
const FORWARDED_RESPONSE_HEADERS: &[&str] = &[
"content-type",
"location",
"cache-control",
"etag",
"last-modified",
"vary",
"retry-after",
"content-disposition",
X_HARNESS_API,
X_HARNESS_MODULE,
X_REQUEST_ID,
];
#[must_use]
pub(crate) fn truthy(config: &dyn Config, key: &str) -> bool {
config.get(key).is_some_and(|raw| {
matches!(
raw.to_ascii_lowercase().as_str(),
"1" | "true" | "yes" | "on"
)
})
}
pub(crate) fn gateway_signer(
config: &dyn Config,
clock: Arc<dyn Clock>,
) -> Option<Arc<HmacSigner>> {
let secret = config
.get(SIDECAR_GATEWAY_SECRET)
.filter(|value| !value.trim().is_empty())?;
let mut ring = KeyRing::new();
match ring.rotate_signing(Kid::Cur, secret.into_bytes()) {
Ok(_) => {}
Err(crate::signer::SignerError::SecretTooShort) => {
tracing::error!("{SIDECAR_GATEWAY_SECRET} must be at least {MIN_SECRET_BYTES} bytes");
return None;
}
Err(err) => {
tracing::error!(error = %err, "{SIDECAR_GATEWAY_SECRET} could not enter the gateway ring");
return None;
}
}
let policy = TokenPolicy::default()
.with_max(GATEWAY_PURPOSE, Some(GATEWAY_TOKEN_TTL_SECS))
.with_max(GATEWAY_ADMIN_PURPOSE, Some(GATEWAY_TOKEN_TTL_SECS));
Some(Arc::new(
HmacSigner::from_ring(ring)
.with_clock(clock)
.with_policy(policy),
))
}
pub(crate) fn mint_gateway(signer: &HmacSigner, mount: &str, admin: bool) -> String {
let purpose = if admin {
GATEWAY_ADMIN_PURPOSE
} else {
GATEWAY_PURPOSE
};
signer.sign(&Payload {
purpose: purpose.to_owned(),
subject: mount.to_owned(),
exp: None,
kid: Kid::Cur,
})
}
pub(crate) fn gateway_grant(signer: &HmacSigner, token: &str) -> Option<bool> {
if signer.verify(token, GATEWAY_ADMIN_PURPOSE).is_some() {
return Some(true);
}
signer.verify(token, GATEWAY_PURPOSE).map(|_| false)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SidecarMount {
pub name: String,
pub binding: String,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct SidecarMounts(Vec<SidecarMount>);
impl SidecarMounts {
pub fn from_config(config: &dyn Config) -> Result<Self, Vec<String>> {
let Some(raw) = config
.get(HARNESS_SIDECARS)
.filter(|v| !v.trim().is_empty())
else {
return Ok(Self::default());
};
if truthy(config, HARNESS_ONE_WORKER) {
return Err(vec![format!(
"{HARNESS_SIDECARS} names a sidecar while {HARNESS_ONE_WORKER} is set: a one-Worker \
deployment serves every module in-process and can mount nothing (issue #131)"
)]);
}
let parsed: BTreeMap<String, String> = serde_json::from_str(&raw).map_err(|err| {
vec![format!(
"{HARNESS_SIDECARS} must be a JSON object of {{\"module-name\": \"BINDING\"}}: {err}"
)]
})?;
let mut errors = Vec::new();
let mut mounts = Vec::new();
for (name, binding) in parsed {
if !is_kebab(&name) {
errors.push(format!(
"sidecar name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
));
}
if binding.trim().is_empty() {
errors.push(format!("sidecar `{name}` has an empty service binding"));
}
mounts.push(SidecarMount { name, binding });
}
if errors.is_empty() {
Ok(Self(mounts))
} else {
Err(errors)
}
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
pub fn iter(&self) -> impl Iterator<Item = &SidecarMount> {
self.0.iter()
}
#[must_use]
pub fn collisions(&self, module_names: &[&str]) -> Vec<String> {
self.0
.iter()
.filter(|m| module_names.contains(&m.name.as_str()))
.map(|m| {
format!(
"sidecar `{}` claims `/v1/{}`, already served in-process",
m.name, m.name
)
})
.collect()
}
}
fn is_kebab(s: &str) -> bool {
!s.is_empty()
&& !s.starts_with('-')
&& !s.ends_with('-')
&& !s.contains("--")
&& s.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
}
struct SidecarState {
mount: SidecarMount,
dispatcher: Option<Arc<dyn Dispatcher>>,
config: Arc<dyn Config>,
rate_limiter: Option<Arc<dyn RateLimiter>>,
gateway: Option<Arc<HmacSigner>>,
}
pub(crate) fn router(
mount: SidecarMount,
dispatcher: Option<Arc<dyn Dispatcher>>,
config: Arc<dyn Config>,
rate_limiter: Option<Arc<dyn RateLimiter>>,
gateway: Option<Arc<HmacSigner>>,
) -> Router {
Router::new()
.fallback(forward)
.with_state(Arc::new(SidecarState {
mount,
dispatcher,
config,
rate_limiter,
gateway,
}))
}
async fn forward(
State(state): State<Arc<SidecarState>>,
scope: Scope,
OriginalUri(uri): OriginalUri,
request: axum::extract::Request,
) -> Response {
let unavailable = |detail: String| -> Response {
Problem::new(&SLUGS.sidecar_unavailable)
.with_detail(detail)
.instance(&scope.request_id)
.into_response()
};
let Some(dispatcher) = state.dispatcher.clone() else {
tracing::warn!(
module = state.mount.name,
"sidecar mounted but the runtime provides no dispatcher"
);
return unavailable(format!(
"`{}` is mounted as a sidecar but this runtime cannot dispatch",
state.mount.name
));
};
if !dispatcher.has(&state.mount.binding) {
tracing::warn!(
module = state.mount.name,
binding = state.mount.binding,
"sidecar binding is not present in this deployment"
);
return unavailable(format!(
"`{}` is mounted on binding `{}`, which this deployment does not have",
state.mount.name, state.mount.binding
));
}
let (parts, body) = request.into_parts();
let Ok(body) = axum::body::to_bytes(body, MAX_BODY_BYTES).await else {
return Problem::new(&SLUGS.request_too_large)
.instance(&scope.request_id)
.into_response();
};
let admin = is_admin_path(uri.path());
if admin && let Err(problem) = require_admin(state.config.as_ref(), &parts.headers) {
return problem.instance(&scope.request_id).into_response();
}
if matches!(
parts.method,
Method::POST | Method::PUT | Method::PATCH | Method::DELETE
) {
let keys = rate_limit_keys(client_ip(&parts.headers).as_deref(), None);
if let RateLimit::Denied { retry_after } = check_rate_limit(
state.rate_limiter.as_ref(),
&keys,
RateLimitFailure::FailClosed,
)
.await
{
return crate::http::rate_limited(retry_after);
}
}
let mut outbound = http::Request::builder()
.method(parts.method.clone())
.uri(uri);
if let Some(headers) = outbound.headers_mut() {
write_forwarded_headers(headers, &state, &scope, &parts.headers, admin);
}
let outbound = match outbound.body(body) {
Ok(req) => req,
Err(err) => return unavailable(format!("could not build the forwarded request: {err}")),
};
match dispatcher.dispatch(&state.mount.binding, outbound).await {
Ok(response) => match contract_of(&response) {
Some(api) if api != HARNESS_API => {
tracing::warn!(
module = state.mount.name,
sidecar_api = api,
host_api = HARNESS_API,
"sidecar contract mismatch"
);
Problem::new(&SLUGS.sidecar_contract_mismatch)
.with_detail(format!(
"`{}` answers contract {api}; this harness speaks {HARNESS_API}",
state.mount.name
))
.instance(&scope.request_id)
.into_response()
}
_ => into_axum(response),
},
Err(err) => {
tracing::warn!(module = state.mount.name, error = %err, "sidecar dispatch failed");
unavailable(err.to_string())
}
}
}
fn write_forwarded_headers(
headers: &mut http::HeaderMap,
state: &SidecarState,
scope: &Scope,
inbound: &http::HeaderMap,
admin: bool,
) {
for name in FORWARDED_REQUEST_HEADERS {
let name = HeaderName::from_static(name);
if let Some(value) = inbound.get(&name) {
headers.append(name, value.clone());
}
}
if let Ok(value) = scope.request_id.parse() {
headers.insert(X_REQUEST_ID, value);
}
if let Some(ip) = client_ip(inbound)
&& let Ok(value) = HeaderValue::from_str(&ip)
{
headers.insert(HeaderName::from_static("cf-connecting-ip"), value);
}
if let Some(signer) = state.gateway.as_ref() {
let token = mint_gateway(signer, &state.mount.name, admin);
if let Ok(value) = HeaderValue::from_str(&token) {
headers.insert(HeaderName::from_static(X_HARNESS_GATEWAY), value);
}
}
}
fn contract_of(response: &http::Response<Bytes>) -> Option<u32> {
response
.headers()
.get(X_HARNESS_API)
.and_then(|v| v.to_str().ok())
.and_then(|v| v.parse().ok())
}
fn into_axum(response: http::Response<Bytes>) -> Response {
let (parts, body) = response.into_parts();
let mut out = Response::new(axum::body::Body::from(body));
*out.status_mut() = parts.status;
*out.version_mut() = parts.version;
for name in FORWARDED_RESPONSE_HEADERS {
let name = HeaderName::from_static(name);
if let Some(value) = parts.headers.get(&name) {
out.headers_mut().append(name.clone(), value.clone());
}
}
out
}
#[derive(serde::Deserialize)]
pub(crate) struct EventEnvelope {
event: String,
payload: serde_json::Value,
}
pub struct EventForwarder {
mounts: Vec<SidecarMount>,
dispatcher: Arc<dyn Dispatcher>,
gateway: Option<Arc<HmacSigner>>,
}
impl EventForwarder {
pub(crate) fn new(
mounts: Vec<SidecarMount>,
dispatcher: Arc<dyn Dispatcher>,
gateway: Option<Arc<HmacSigner>>,
) -> Self {
Self {
mounts,
dispatcher,
gateway,
}
}
#[must_use]
pub fn is_empty(&self) -> bool {
self.mounts.is_empty()
}
pub(crate) async fn forward(&self, request_id: &str, name: &str, payload: &serde_json::Value) {
let body = match serde_json::to_vec(&serde_json::json!({
"event": name,
"payload": payload,
})) {
Ok(body) => body,
Err(err) => {
tracing::warn!(event = %name, error = %err, "event payload did not serialize");
return;
}
};
for mount in &self.mounts {
if !self.dispatcher.has(&mount.binding) {
continue;
}
let mut builder = http::Request::builder()
.method(Method::POST)
.uri("/__events")
.header(HeaderName::from_static("content-type"), "application/json");
if let Ok(value) = HeaderValue::from_str(request_id) {
builder = builder.header(HeaderName::from_static(X_REQUEST_ID), value);
}
if let Some(signer) = self.gateway.as_ref() {
builder = builder.header(
HeaderName::from_static(X_HARNESS_GATEWAY),
mint_gateway(signer, &mount.name, false),
);
}
let request = match builder.body(Bytes::from(body.clone())) {
Ok(request) => request,
Err(err) => {
tracing::warn!(event = %name, error = %err, "event forward could not be built");
continue;
}
};
match self.dispatcher.dispatch(&mount.binding, request).await {
Ok(response) if response.status().is_success() => {}
Ok(response) => {
let detail = format!(
"event forward to `{}` answered {}",
mount.name,
response.status()
);
tracing::warn!(event = %name, module = %mount.name, detail, "event forward was not accepted");
crate::logging::forward_internal_error(&detail);
}
Err(err) => {
let detail = format!(
"event forward to `{mount}` failed: {err}",
mount = mount.name
);
tracing::warn!(event = %name, module = %mount.name, error = %err, "event forward failed");
crate::logging::forward_internal_error(&detail);
}
}
}
}
}
pub(crate) struct InboundEvents {
pub(crate) bus: EventBus,
pub(crate) gateway: Option<Arc<HmacSigner>>,
}
pub(crate) async fn events_inbound(
State(state): State<Arc<InboundEvents>>,
scope: Scope,
headers: axum::http::HeaderMap,
body: bytes::Bytes,
) -> Response {
if let Some(signer) = state.gateway.as_ref() {
let presented = headers
.get(X_HARNESS_GATEWAY)
.and_then(|value| value.to_str().ok());
if presented
.and_then(|token| gateway_grant(signer, token))
.is_none()
{
return Problem::new(&SLUGS.sidecar_unauthorized)
.instance(&scope.request_id)
.into_response();
}
}
let Ok(envelope) = serde_json::from_slice::<EventEnvelope>(&body) else {
return Problem::new(&SLUGS.validation_failed)
.with_detail("body must be `{\"event\": \"<name>\", \"payload\": …}`")
.instance(&scope.request_id)
.into_response();
};
let handled = state
.bus
.deliver_inbound(&scope, &envelope.event, &envelope.payload);
if handled == 0 {
let detail = format!(
"event `{}` arrived over the boundary with no subscriber",
envelope.event
);
tracing::warn!(event = %envelope.event, "{detail}");
crate::logging::forward_internal_error(&detail);
}
(
StatusCode::ACCEPTED,
Json(serde_json::json!({ "accepted": handled > 0, "handlers": handled })),
)
.into_response()
}
fn is_admin_path(path: &str) -> bool {
path == "/admin" || path.ends_with("/admin") || path.contains("/admin/")
}
#[derive(Clone)]
pub(crate) struct GatewayGuard {
pub require: bool,
pub signer: Option<Arc<HmacSigner>>,
pub admin_token: Option<String>,
}
pub(crate) async fn gateway_guard(
State(guard): State<GatewayGuard>,
request: axum::extract::Request,
next: Next,
) -> Response {
let path = request.uri().path().to_owned();
if !guard.require || !(path.starts_with("/v1/") || path == "/__surface" || path == "/__events")
{
return next.run(request).await;
}
let instance = request
.extensions()
.get::<Scope>()
.map(|scope| scope.request_id.clone());
let refused = |problem: Problem| match &instance {
Some(id) => problem.instance(id),
None => problem,
};
let Some(signer) = guard.signer.as_ref() else {
tracing::error!(
"{SIDECAR_REQUIRE_GATEWAY} is set but {SIDECAR_GATEWAY_SECRET} is \
missing; refusing every guarded request"
);
return refused(Problem::new(&SLUGS.sidecar_unavailable).with_detail(
"this sidecar requires a gateway token but no gateway secret is configured",
))
.into_response();
};
let presented = request
.headers()
.get(X_HARNESS_GATEWAY)
.and_then(|value| value.to_str().ok());
let Some(host_authorized_admin) = presented.and_then(|token| gateway_grant(signer, token))
else {
return refused(Problem::new(&SLUGS.sidecar_unauthorized)).into_response();
};
let mut request = request;
if host_authorized_admin
&& is_admin_path(&path)
&& !request.headers().contains_key("authorization")
&& let Some(token) = &guard.admin_token
&& let Ok(value) = HeaderValue::from_str(&format!("Bearer {token}"))
{
request
.headers_mut()
.insert(HeaderName::from_static("authorization"), value);
}
next.run(request).await
}