crates/estate-front/src/web/serve.rs
//! **The front's two web ports**, served by this process — there is no Caddy
//! on an OS-less front.
//!
//! * **tcp/80**: the ACME HTTP-01 challenges ([`super::acme::Challenges`]) and,
//! for everything else, a permanent redirect to `https://` on the same host
//! and path;
//! * **tcp/443**: TLS with the certificate [`super::certs::Resolver`] answers
//! on each handshake (swapped live), HTTP/1.1 and HTTP/2, serving whatever
//! [`App`] holds at the moment the connection is accepted — so the web
//! face (the console) can be replaced, e.g. when the primary moves, without
//! a restart and without dropping a connection already open.
//!
//! Around the face, per accepted connection and in this order: the payment
//! vendors' callbacks carried to the product's monetize ([`with_callbacks`],
//! when the product set a target — the paid front), the callback-only hosts
//! ([`narrowed`]), monetize's own host ([`controlled`], the paid front's
//! control API from outside), and the names carried to the primary's
//! repository ([`routed`], when the product routes one).
use std::net::SocketAddr;
use std::sync::{Arc, RwLock};
use axum::extract::{ConnectInfo, Path as UrlPath, State};
use axum::http::{header, HeaderMap, StatusCode, Uri};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::Router;
use super::acme::Challenges;
use super::certs::Resolver;
/// **What 443 serves right now** — replaced whole, read per connection — which
/// of its names answer callbacks only, where callbacks go, and which names are
/// carried to the primary's repository.
#[derive(Clone)]
pub struct App {
face: Arc<RwLock<Router>>,
callback_only: Arc<RwLock<Option<CallbackOnly>>>,
callbacks: Arc<RwLock<Option<SocketAddr>>>,
routed: Arc<RwLock<Option<Routed>>>,
control_host: Arc<RwLock<Option<String>>>,
outer: Arc<RwLock<Option<Outer>>>,
hosts: Arc<RwLock<Vec<String>>>,
}
/// **A product's own outer layer** on :443 (installed through [`super::Hooks::outer`], RA4-22): wrapped around everything the
/// shared front serves (the face, its callbacks, the narrowing, the repository route).
pub type Outer = Arc<dyn Fn(Router) -> Router + Send + Sync>;
/// **Names that answer a payment vendor's webhook and nothing else**
/// (`web_callback_only`): `/callback/<vendor>` is served there, every other
/// path is a 308 to the same path on `to` (the first web host). A vendor's
/// webhook name (the Stripe account's) is a machine's door, not a second
/// console origin — a passkey is bound to its relying-party id.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CallbackOnly {
pub hosts: Vec<String>,
pub to: String,
}
impl App {
pub fn new(router: Router) -> App {
App {
face: Arc::new(RwLock::new(router)),
callback_only: Arc::new(RwLock::new(None)),
callbacks: Arc::new(RwLock::new(None)),
routed: Arc::new(RwLock::new(None)),
control_host: Arc::new(RwLock::new(None)),
outer: Arc::new(RwLock::new(None)),
hosts: Arc::new(RwLock::new(Vec::new())),
}
}
/// From the next accepted connection on, serve `router`.
pub fn set(&self, router: Router) {
*self.face.write().unwrap_or_else(|e| e.into_inner()) = router;
}
pub fn current(&self) -> Router {
self.face.read().unwrap_or_else(|e| e.into_inner()).clone()
}
/// From the next accepted connection on, `hosts` answer callbacks only and
/// send everything else to `to`. No hosts: nothing is narrowed.
pub fn set_callback_only(&self, hosts: &[String], to: &str) {
let rule = (!hosts.is_empty()).then(|| CallbackOnly {
hosts: hosts.to_vec(),
to: to.to_string(),
});
*self
.callback_only
.write()
.unwrap_or_else(|e| e.into_inner()) = rule;
}
pub fn callback_only(&self) -> Option<CallbackOnly> {
self.callback_only
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
/// From the next accepted connection on, `POST /callback/<vendor>` is
/// carried to monetize at `to` (loopback, HTTP/1.1) — the paid front sets
/// it; an open front leaves it `None` and has no callback door at all.
pub fn set_callbacks(&self, to: Option<SocketAddr>) {
*self.callbacks.write().unwrap_or_else(|e| e.into_inner()) = to;
}
pub fn callbacks(&self) -> Option<SocketAddr> {
*self.callbacks.read().unwrap_or_else(|e| e.into_inner())
}
/// From the next accepted connection on, route `r` (the product's
/// repository); `None` routes nothing. Answers whether it changed.
pub fn set_routed(&self, r: Option<Routed>) -> bool {
let mut cur = self.routed.write().unwrap_or_else(|e| e.into_inner());
if *cur == r {
return false;
}
*cur = r;
true
}
pub fn routed(&self) -> Option<Routed> {
self.routed
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
/// From the next accepted connection on, every request to `host` is
/// monetize's ([`controlled`], at the [`App::set_callbacks`] target);
/// `None`: no such host.
pub fn set_control_host(&self, host: Option<&str>) {
*self.control_host.write().unwrap_or_else(|e| e.into_inner()) = host.map(str::to_string);
}
pub fn control_host(&self) -> Option<String> {
self.control_host
.read()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
/// `outer` wraps everything this front serves ([`Outer`]; A32-11: Gunnar's CLI hole, login
/// names, git paths and access log). `None`: nothing. Crate-only (RA4-22): a product installs
/// its outer layer ONE way, [`super::Hooks::outer`], before anything is served.
pub(crate) fn set_outer(&self, outer: Option<Outer>) {
*self.outer.write().unwrap_or_else(|e| e.into_inner()) = outer;
}
/// From the next accepted connection (and the next :80 request) on, the served names are
/// `hosts` (`web_hosts`): a served `<h>` whose `www.<h>` is served too is sent to
/// `https://www.<h>` ([`to_www`], WWW-CANON). Empty: nothing is redirected.
pub fn set_hosts(&self, hosts: &[String]) {
*self.hosts.write().unwrap_or_else(|e| e.into_inner()) = hosts.to_vec();
}
pub fn hosts(&self) -> Vec<String> {
self.hosts.read().unwrap_or_else(|e| e.into_inner()).clone()
}
/// **The router one accepted connection is served by**: the face, its
/// callbacks, monetize's own host, the callback-only narrowing, the
/// repository route, and the product's outer layer around all of it.
pub fn router_now(&self) -> Router {
let mut router = self.current();
if let Some(to) = self.callbacks() {
router = with_callbacks(router, to);
router = controlled(router, self.control_host(), to);
}
let callback_only = self.callback_only();
// RA1-16: the route wraps outside the narrowing, so it carries only the product's OWN
// names: never a callback-only one (the route's feed refuses it too).
let route = self.routed().map(|mut r| {
r.hosts
.retain(|h| !callback_only.as_ref().is_some_and(|c| c.hosts.contains(h)));
r
});
let router = routed(
narrowed(router, callback_only),
route.filter(|r| !r.hosts.is_empty()),
);
let router = match self.outer.read().unwrap_or_else(|e| e.into_inner()).clone() {
Some(outer) => outer(router),
None => router,
};
// WWW-CANON: outside everything, the product's layers too: the bare `<site>` serves nothing.
to_www(router, self.hosts())
}
}
/// Pure (WWW-CANON, owner 2026-10-30: the sites are `www.gunnar.rs` and `www.holger.rs`): **the
/// canonical name of `host`** — `host` (any case, a port or a trailing dot allowed) is a served
/// name `<h>`, not itself a `www.` name, and `www.<h>` is served too: `www.<h>`. Every other host
/// (`www.<h>` itself, `git.<h>`, a name whose `www.` is not served): `None`. The passkeys' relying
/// party stays `<h>`; `https://www.<h>` is an origin of it.
pub fn www_of(host: &str, hosts: &[String]) -> Option<String> {
let host = host
.split(':')
.next()
.unwrap_or(host)
.trim_end_matches('.')
.to_ascii_lowercase();
if host.starts_with("www.") || !hosts.contains(&host) {
return None;
}
let www = format!("www.{host}");
hosts.contains(&www).then_some(www)
}
/// Pure: **a request a machine makes to the bare host, never redirected** (WWW-CANON): a gRPC
/// call (a CLI dialing `https://<site>`; a redirect is not a gRPC answer) and a payment vendor's
/// webhook (`/callback/…`; Stripe does not follow redirects).
pub fn stays_on_bare(headers: &HeaderMap, path: &str) -> bool {
path.starts_with("/callback/")
|| headers
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.is_some_and(|v| v.starts_with("application/grpc"))
}
/// The redirect to `https://<to><path and query>`: 301 for a page load (GET, HEAD), 308 for
/// every other method, so a POST keeps its method and body.
fn canonical_redirect(to: &str, method: &http::Method, uri: &Uri) -> Response {
let path = uri.path_and_query().map(|p| p.as_str()).unwrap_or("/");
let status = if method == http::Method::GET || method == http::Method::HEAD {
StatusCode::MOVED_PERMANENTLY
} else {
StatusCode::PERMANENT_REDIRECT
};
(status, [(header::LOCATION, format!("https://{to}{path}"))]).into_response()
}
/// **`face` with the bare `<site>` sent to `www.<site>`** (WWW-CANON): every request whose host
/// has a canonical `www.` name ([`www_of`]) is a 301 (308 for a method other than GET and HEAD)
/// to `https://www.<h>` with the same path and query, and reaches nothing else, except what
/// [`stays_on_bare`]; every other host is `face`. No hosts: `face`.
pub fn to_www(face: Router, hosts: Vec<String>) -> Router {
if hosts.is_empty() {
return face;
}
let hosts = Arc::new(hosts);
Router::new()
.fallback_service(face)
.layer(axum::middleware::from_fn(
move |req: axum::extract::Request, next: axum::middleware::Next| {
let hosts = Arc::clone(&hosts);
async move {
match request_host(&req).and_then(|h| www_of(&h, &hosts)) {
Some(www) if !stays_on_bare(req.headers(), req.uri().path()) => {
canonical_redirect(&www, req.method(), req.uri())
}
_ => next.run(req).await,
}
}
},
))
}
/// **Names carried to the primary's repository**: a request to one of `hosts`
/// goes to `grpc` when it is a gRPC call (`content-type: application/grpc…`,
/// HTTP/2 cleartext), else to `http` (HTTP/1.1) — the body streamed both ways,
/// never buffered, trailers kept. TLS ends HERE, where the certificate is; the
/// leg to the primary is the estate's private utility network.
///
/// **Only the gRPC services the product declares are carried** (RA5-1): a call to
/// any other path is answered HERE as `UNIMPLEMENTED` and audited, so a service
/// the primary serves for its own boxes or its console (its identity core, say)
/// is never reachable from the internet through the front ([`carries`]).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Routed {
pub hosts: Vec<String>,
pub http: std::net::SocketAddrV4,
pub grpc: std::net::SocketAddrV4,
/// The gRPC services carried, each `package.Service` (every method), or
/// `package.Service/Method` (that one), or a package prefix ending in `.`
/// (`holger.`: every service of every `holger.*` package). Empty: no gRPC call
/// is carried.
pub grpc_services: Vec<String>,
}
/// Pure: the gRPC call at `path` (`/package.Service/Method`) is one of `services`
/// ([`Routed::grpc_services`]). A path that is not a gRPC method's shape is never one.
pub fn carries(services: &[String], path: &str) -> bool {
let Some((service, method)) = path.strip_prefix('/').and_then(|p| p.split_once('/')) else {
return false;
};
if service.is_empty() || method.is_empty() || method.contains('/') || !service.contains('.') {
return false;
}
services.iter().any(|s| match s.split_once('/') {
Some((svc, m)) => svc == service && m == method,
None if s.ends_with('.') => service.starts_with(s.as_str()),
None => s == service,
})
}
/// Headers that belong to one hop, never carried to the primary. `te` is
/// carried: gRPC needs `te: trailers` end to end.
const ROUTE_HOP_BY_HOP: &[&str] = &[
"connection",
"keep-alive",
"proxy-connection",
"transfer-encoding",
"trailer",
"upgrade",
];
/// Headers that say who the client is and what public face it reached. The primary believes
/// them from the front alone (Holger's `ron_trusted_proxies`), so what a client sent in them is
/// dropped, never carried; the front writes its own (`x-forwarded-proto`, `-host`, `-for`).
const ROUTE_FORWARDED: &[&str] = &[
"x-forwarded-proto",
"x-forwarded-host",
"x-forwarded-port",
"x-forwarded-prefix",
"x-forwarded-for",
"x-real-ip",
"forwarded",
];
type RouteClient = hyper_util::client::legacy::Client<
hyper_util::client::legacy::connect::HttpConnector,
axum::body::Body,
>;
/// How often an idle pooled connection to the primary is pinged (HTTP/2), and how
/// long a ping may go unanswered before the connection is dropped.
pub const ROUTE_KEEPALIVE: std::time::Duration = std::time::Duration::from_secs(10);
pub const ROUTE_KEEPALIVE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
/// A pooled connection idle longer than this is closed rather than reused.
pub const ROUTE_POOL_IDLE: std::time::Duration = std::time::Duration::from_secs(30);
/// **The two pooled clients the route carries with.** The HTTP/2 one PINGS its
/// connection: a primary that restarts takes its end of the pooled TCP connection
/// with it, nothing tells the front (no FIN crosses the utility network), and every
/// gRPC request after the restart failed `SendRequest` on the dead connection —
/// MEASURED on holger's mock 2026-09-28 (HTTP 502 forever while the primary served
/// again). A ping unanswered for [`ROUTE_KEEPALIVE_TIMEOUT`] closes it, and the next
/// request dials afresh.
fn route_clients() -> &'static (RouteClient, RouteClient) {
static CLIENTS: std::sync::OnceLock<(RouteClient, RouteClient)> = std::sync::OnceLock::new();
CLIENTS.get_or_init(|| {
let exec = hyper_util::rt::TokioExecutor::new();
let h1 = hyper_util::client::legacy::Client::builder(exec.clone())
.pool_idle_timeout(ROUTE_POOL_IDLE)
.pool_timer(hyper_util::rt::TokioTimer::new())
.build_http();
let h2 = hyper_util::client::legacy::Client::builder(exec)
.http2_only(true)
.timer(hyper_util::rt::TokioTimer::new())
.pool_timer(hyper_util::rt::TokioTimer::new())
.pool_idle_timeout(ROUTE_POOL_IDLE)
.http2_keep_alive_interval(ROUTE_KEEPALIVE)
.http2_keep_alive_timeout(ROUTE_KEEPALIVE_TIMEOUT)
.http2_keep_alive_while_idle(true)
.build_http();
(h1, h2)
})
}
/// The calls one address may make at once on the routed gRPC door (holger lane B3).
pub const ROUTE_GRPC_BURST: u32 = 200;
/// The calls a second one address may make after its burst.
pub const ROUTE_GRPC_PER_SEC: u32 = 50;
/// **One caller's token bucket**: `burst` calls at once, refilled at `per_sec` a second.
#[derive(Debug, Clone)]
pub struct Bucket {
tokens: f64,
burst: f64,
per_sec: f64,
at: Option<std::time::Instant>,
}
impl Bucket {
pub fn full(burst: u32, per_sec: u32) -> Self {
Bucket {
tokens: f64::from(burst),
burst: f64::from(burst),
per_sec: f64::from(per_sec),
at: None,
}
}
/// Refilled to its burst by `now`: a caller that has been quiet, safe to forget.
pub fn idle(&self, now: std::time::Instant) -> bool {
self.at.is_none_or(|at| {
self.tokens + now.saturating_duration_since(at).as_secs_f64() * self.per_sec
>= self.burst
})
}
/// Take one call's token at `now`, or answer `false` when the caller has none left.
pub fn take(&mut self, now: std::time::Instant) -> bool {
if let Some(at) = self.at {
let elapsed = now.saturating_duration_since(at).as_secs_f64();
self.tokens = (self.tokens + elapsed * self.per_sec).min(self.burst);
}
self.at = Some(now);
if self.tokens >= 1.0 {
self.tokens -= 1.0;
true
} else {
false
}
}
}
/// The routed gRPC door's buckets: one per caller and one per IPv6 site ([`Limiter`]).
fn grpc_buckets() -> &'static std::sync::Mutex<Limiter> {
static B: std::sync::OnceLock<std::sync::Mutex<Limiter>> = std::sync::OnceLock::new();
B.get_or_init(|| std::sync::Mutex::new(Limiter::new(ROUTE_GRPC_CALLERS)))
}
/// How many calling addresses the gRPC door keeps a bucket for (and as many IPv6 sites).
pub const ROUTE_GRPC_CALLERS: usize = 4096;
/// The calls one IPv6 /48 site may make at once, over all its /64s (RA4-23): four callers' worth.
pub const ROUTE_GRPC_SITE_BURST: u32 = 4 * ROUTE_GRPC_BURST;
/// The calls a second one IPv6 /48 site may make after its burst.
pub const ROUTE_GRPC_SITE_PER_SEC: u32 = 4 * ROUTE_GRPC_PER_SEC;
/// Pure: **the key a caller's bucket is kept under** (RA2-3): an IPv4 address, or an IPv6
/// /64 (one end site holds a whole /64, so a per-address key would give it 2^64 buckets). An
/// IPv4-mapped IPv6 address is its IPv4 caller.
/// The one partition rule, `nornir_auth_webauthn::client_key::ClientKey` (RA5-34).
pub fn caller_key(ip: std::net::IpAddr) -> String {
nornir_auth_webauthn::client_key::ClientKey::of_ip(ip)
.as_str()
.to_owned()
}
/// Pure: **the IPv6 site a caller is part of** (RA4-23): its /48, the block one end site is
/// given, so a /48 holder that rotates its 65,536 /64s shares one site bucket. `None` for an
/// IPv4 (or IPv4-mapped) caller, whose address is already its whole key.
pub fn site_key(ip: std::net::IpAddr) -> Option<String> {
match ip {
std::net::IpAddr::V4(_) => None,
std::net::IpAddr::V6(v6) if v6.to_ipv4_mapped().is_some() => None,
std::net::IpAddr::V6(v6) => {
let s = v6.segments();
Some(format!("{:x}:{:x}:{:x}::/48", s[0], s[1], s[2]))
}
}
}
/// **The buckets a door keeps, at most `cap`, the least recently seen dropped first** (RA4-23:
/// an LRU, O(log n) under the mutex, where a full map scanned every bucket for its oldest). A
/// full map drops one bucket at a time, never all at once, so a flooder's empty bucket is not
/// handed back full (RA2-3): the flooder is the most recently seen.
#[derive(Debug)]
pub struct Callers {
cap: usize,
seen: u64,
buckets: std::collections::HashMap<String, (Bucket, u64)>,
order: std::collections::BTreeMap<u64, String>,
}
impl Callers {
/// At most `cap` buckets (at least one).
pub fn new(cap: usize) -> Self {
Callers {
cap: cap.max(1),
seen: 0,
buckets: Default::default(),
order: Default::default(),
}
}
/// How many buckets are kept.
pub fn len(&self) -> usize {
self.buckets.len()
}
/// Whether no bucket is kept.
pub fn is_empty(&self) -> bool {
self.buckets.is_empty()
}
/// Whether `key` has a bucket.
pub fn contains(&self, key: &str) -> bool {
self.buckets.contains_key(key)
}
/// **Take one token for `key` at `now`** from its bucket (a new one holds `burst`, refilled at
/// `per_sec`), dropping the least recently seen bucket when a new key finds the map full.
pub fn take(&mut self, key: &str, now: std::time::Instant, burst: u32, per_sec: u32) -> bool {
self.seen += 1;
let seen = self.seen;
if let Some((bucket, at)) = self.buckets.get_mut(key) {
self.order.remove(at);
*at = seen;
self.order.insert(seen, key.to_owned());
return bucket.take(now);
}
while self.buckets.len() >= self.cap {
let Some((_, oldest)) = self.order.pop_first() else {
break;
};
self.buckets.remove(&oldest);
}
let mut bucket = Bucket::full(burst, per_sec);
let ok = bucket.take(now);
self.buckets.insert(key.to_owned(), (bucket, seen));
self.order.insert(seen, key.to_owned());
ok
}
}
/// **The routed door's limit: a bucket per caller, and a second tier per IPv6 site** (RA4-23).
/// A call is admitted when its caller (an IPv4 address or an IPv6 /64, [`caller_key`]) has a
/// token AND, for IPv6, its /48 ([`site_key`]) has one.
#[derive(Debug)]
pub struct Limiter {
callers: Callers,
sites: Callers,
}
impl Limiter {
/// At most `cap` callers and `cap` sites kept.
pub fn new(cap: usize) -> Self {
Limiter {
callers: Callers::new(cap),
sites: Callers::new(cap),
}
}
/// May `ip` make one more call at `now`?
pub fn admit(&mut self, ip: std::net::IpAddr, now: std::time::Instant) -> bool {
self.admit_key(&caller_key(ip), site_key(ip).as_deref(), now)
}
/// [`Limiter::admit`] by keys: the caller's, and its site's when it has one.
pub fn admit_key(&mut self, caller: &str, site: Option<&str>, now: std::time::Instant) -> bool {
if !self
.callers
.take(caller, now, ROUTE_GRPC_BURST, ROUTE_GRPC_PER_SEC)
{
return false;
}
site.is_none_or(|site| {
self.sites
.take(site, now, ROUTE_GRPC_SITE_BURST, ROUTE_GRPC_SITE_PER_SEC)
})
}
/// The callers' buckets.
pub fn callers(&self) -> &Callers {
&self.callers
}
}
/// The plain-HTTP paths on a routed host that carry credentials (a repository's settings
/// and passkey doors, holger's `/-/settings/` and `/-/auth/`): counted per caller like the
/// gRPC door. Package paths are not (package clients fetch in bursts).
pub const ROUTE_COUNTED_HTTP: &[&str] = &["/-/settings/", "/-/auth/"];
/// Pure: **is this plain-HTTP path counted** by the per-caller limit ([`ROUTE_COUNTED_HTTP`]).
pub fn counted_http(path: &str) -> bool {
ROUTE_COUNTED_HTTP.iter().any(|p| path.starts_with(p))
}
/// May the caller at `ip` (or an unknown one, all under one key) make one more routed call now?
fn grpc_admitted(ip: Option<std::net::IpAddr>) -> bool {
let now = std::time::Instant::now();
let mut all = grpc_buckets().lock().unwrap_or_else(|p| p.into_inner());
match ip {
Some(ip) => all.admit(ip, now),
None => all.admit_key("unknown", None, now),
}
}
/// The front's own gRPC answer to a caller over its rate: `RESOURCE_EXHAUSTED`, said in
/// the headers (a trailers-only response), with when to come back.
/// The answer to a gRPC call the route does not carry ([`carries`]): `UNIMPLEMENTED`.
fn grpc_not_carried() -> Response {
let mut h = http::HeaderMap::new();
h.insert(
header::CONTENT_TYPE,
http::HeaderValue::from_static("application/grpc"),
);
h.insert("grpc-status", http::HeaderValue::from_static("12"));
h.insert(
"grpc-message",
http::HeaderValue::from_static("this%20front%20carries%20no%20such%20service"),
);
(StatusCode::OK, h).into_response()
}
fn grpc_busy() -> Response {
let mut r = Response::new(axum::body::Body::empty());
let h = r.headers_mut();
h.insert(
header::CONTENT_TYPE,
http::HeaderValue::from_static("application/grpc"),
);
h.insert("grpc-status", http::HeaderValue::from_static("8"));
h.insert("grpc-message", http::HeaderValue::from_static("the%20front%27s%20gRPC%20route%20is%20busy%20for%20your%20address%3B%20retry%20shortly"));
h.insert(header::RETRY_AFTER, http::HeaderValue::from_static("1"));
r
}
/// **`face` with `route`'s names carried to the primary** — every other name
/// is `face`, unchanged. `None`: `face` itself.
pub fn routed(face: Router, route: Option<Routed>) -> Router {
let Some(route) = route else { return face };
let route = Arc::new(route);
Router::new().fallback_service(face).layer(axum::middleware::from_fn(move |req: axum::extract::Request, next: axum::middleware::Next| {
let route = Arc::clone(&route);
async move {
if !request_host(&req).is_some_and(|h| route.hosts.contains(&h)) {
return next.run(req).await;
}
// The gRPC door: rate-limited per calling address, only the declared services
// (RA5-1), and audited (a line per call: who, which RPC, what came back). The
// primary decides everything else.
let grpc = req.headers().get(header::CONTENT_TYPE).and_then(|v| v.to_str().ok()).is_some_and(|v| v.starts_with("application/grpc"));
// The caller: an IPv4 address or an IPv6 /64 (RA2-3), within its /48 (RA4-23).
let ip = req.extensions().get::<ConnectInfo<SocketAddr>>().map(|c| c.0.ip());
let peer = ip.map(caller_key).unwrap_or_else(|| "unknown".into());
if !grpc {
// The credential doors on plain HTTP are counted too (RA2-3); package paths not.
if counted_http(req.uri().path()) && !grpc_admitted(ip) {
tracing::warn!(target: "estate_front::audit", %peer, path = %req.uri().path(), "web: routed HTTP call refused: over the per-caller rate");
return (StatusCode::TOO_MANY_REQUESTS, [(header::RETRY_AFTER, "1")], "over the per-caller rate; retry shortly\n").into_response();
}
return carry(req, &route, false).await;
}
let rpc = req.uri().path().to_owned();
if !grpc_admitted(ip) {
tracing::warn!(target: "estate_front::audit", %peer, %rpc, "web: routed gRPC call refused: over the per-caller rate");
return grpc_busy();
}
if !carries(&route.grpc_services, &rpc) {
tracing::warn!(target: "estate_front::audit", %peer, %rpc, "web: routed gRPC call refused: not a service this front carries");
return grpc_not_carried();
}
let answer = carry(req, &route, false).await;
tracing::info!(target: "estate_front::audit", %peer, %rpc, status = answer.status().as_u16(), "web: routed gRPC call carried to the primary");
answer
}
}))
}
/// **monetize's own host** (`monetize_control_host`, `monetize.<apex>`): every
/// request there is carried to monetize at `to` — its control API as gRPC over
/// h2c, anything else (its `/callback/<vendor>`) as HTTP/1.1 — exactly as the
/// repository route carries. Every other host is `face`. `None`: `face`.
///
/// WHO may call is monetize's decision, never this proxy's: it adds nothing
/// and forwards the caller's per-call signature (`monetize-authorization`,
/// SSHSIG, one of monetize's `control_principals`). From outside that is the
/// ONLY way in: an `authorization` bearer is stripped here, so monetize's
/// loopback bearer never opens its API to the internet, whoever learned it.
pub fn controlled(face: Router, host: Option<String>, to: SocketAddr) -> Router {
let (Some(host), SocketAddr::V4(to)) = (host, to) else {
return face;
};
// Monetize decides who calls what (its per-call signature); the list is the repository route's.
let route = Arc::new(Routed {
hosts: vec![host],
http: to,
grpc: to,
grpc_services: Vec::new(),
});
Router::new()
.fallback_service(face)
.layer(axum::middleware::from_fn(
move |req: axum::extract::Request, next: axum::middleware::Next| {
let route = Arc::clone(&route);
async move {
if !request_host(&req).is_some_and(|h| route.hosts.contains(&h)) {
return next.run(req).await;
}
carry(req, &route, true).await
}
},
))
}
/// Carry `req` to `route`; `strip_bearer`: monetize's host, where only a
/// per-call signature passes ([`controlled`]).
async fn carry(req: axum::extract::Request, route: &Routed, strip_bearer: bool) -> Response {
let grpc = req
.headers()
.get(header::CONTENT_TYPE)
.and_then(|v| v.to_str().ok())
.is_some_and(|v| v.starts_with("application/grpc"));
let (to, version) = if grpc {
(route.grpc, http::Version::HTTP_2)
} else {
(route.http, http::Version::HTTP_11)
};
let peer = req
.extensions()
.get::<ConnectInfo<SocketAddr>>()
.map(|c| c.0.ip().to_string());
let (mut parts, body) = req.into_parts();
let path = parts
.uri
.path_and_query()
.map(|p| p.as_str())
.unwrap_or("/")
.to_string();
let Ok(uri) = format!("http://{to}{path}").parse::<Uri>() else {
return (StatusCode::BAD_REQUEST, "the path cannot be carried\n").into_response();
};
// The name the client asked for travels as `Host` (HTTP/1.1) so the
// repository builds its own URLs with it; HTTP/2 names it in `:authority`.
let host = request_host_of(&parts.headers, &parts.uri);
for h in ROUTE_HOP_BY_HOP.iter().chain(ROUTE_FORWARDED) {
parts.headers.remove(*h);
}
parts.headers.remove(header::HOST);
if strip_bearer && parts.headers.remove(header::AUTHORIZATION).is_some() {
tracing::info!(path = %path, "web: monetize's host: an `authorization` bearer from outside was stripped (only monetize-authorization, SSHSIG, opens monetize from here)");
}
if !grpc {
if let Some(h) = &host {
if let Ok(v) = h.parse() {
parts.headers.insert(header::HOST, v);
}
}
}
// The public face the client reached: https on this front, and the name it asked for
// (also for a gRPC call, whose name travels in `:authority`).
parts
.headers
.insert("x-forwarded-proto", http::HeaderValue::from_static("https"));
if let Some(v) = host
.as_deref()
.and_then(|h| http::HeaderValue::from_str(h).ok())
{
parts.headers.insert("x-forwarded-host", v);
}
if let Some(ip) = peer.and_then(|ip| ip.parse().ok()) {
parts.headers.insert("x-forwarded-for", ip);
}
parts.uri = uri;
parts.version = version;
let req = http::Request::from_parts(parts, body);
let (h1, h2) = route_clients();
let answer = if grpc {
h2.request(req).await
} else {
h1.request(req).await
};
match answer {
Ok(resp) => {
let (parts, body) = resp.into_parts();
Response::from_parts(parts, axum::body::Body::new(body))
}
Err(e) => {
let what = if strip_bearer {
"monetize"
} else {
"the repository"
};
tracing::warn!(%to, error = %e, "web: {what} did not answer");
(StatusCode::BAD_GATEWAY, format!("{what} did not answer\n")).into_response()
}
}
}
/// [`request_host`], from the parts.
fn request_host_of(headers: &HeaderMap, uri: &Uri) -> Option<String> {
headers
.get(header::HOST)
.and_then(|h| h.to_str().ok())
.map(str::to_string)
.or_else(|| uri.authority().map(|a| a.to_string()))
}
/// The request's host, lower case and without a port: `Host` (HTTP/1.1), else
/// the URI's authority (HTTP/2 `:authority`).
fn request_host(req: &axum::extract::Request) -> Option<String> {
let h = req
.headers()
.get(header::HOST)
.and_then(|h| h.to_str().ok())
.map(str::to_string)
.or_else(|| req.uri().host().map(str::to_string))?;
Some(h.split(':').next().unwrap_or(&h).to_ascii_lowercase())
}
/// **`face` narrowed by `rule`** — on a callback-only host, `/callback/…`
/// passes to `face` (which carries the callbacks) and every other path is a
/// 308 to the same path on `rule.to`. Other hosts are untouched.
pub fn narrowed(face: Router, rule: Option<CallbackOnly>) -> Router {
let Some(rule) = rule else { return face };
let rule = Arc::new(rule);
Router::new()
.fallback_service(face)
.layer(axum::middleware::from_fn(
move |req: axum::extract::Request, next: axum::middleware::Next| {
let rule = Arc::clone(&rule);
async move {
let narrowed = request_host(&req).is_some_and(|h| rule.hosts.contains(&h));
if narrowed && !req.uri().path().starts_with("/callback/") {
let path = req
.uri()
.path_and_query()
.map(|p| p.as_str())
.unwrap_or("/");
return (
StatusCode::PERMANENT_REDIRECT,
[(header::LOCATION, format!("https://{}{path}", rule.to))],
)
.into_response();
}
next.run(req).await
}
},
))
}
/// The page 443 serves before the web face is up: a plain word, never a
/// stack trace, and `/health` for a probe.
pub fn placeholder(what: &'static str) -> Router {
Router::new()
.route("/health", get(|| async { "ok\n" }))
.fallback(move || async move { (StatusCode::SERVICE_UNAVAILABLE, format!("{what}\n")) })
}
/// **The :80 router**: the challenge, else the redirect.
pub fn http_router(challenges: Arc<Challenges>) -> Router {
http_router_for(challenges, App::new(Router::new()))
}
/// [`http_router`] with the served names of `app` ([`App::set_hosts`]): the challenge (for the
/// bare `<site>` too: its certificate is kept), else the bare `<site>` is a 301 straight to
/// `https://www.<site>` (WWW-CANON), else the redirect to the same host's https.
pub fn http_router_for(challenges: Arc<Challenges>, app: App) -> Router {
Router::new()
.route("/.well-known/acme-challenge/{token}", get(challenge))
.fallback(redirect)
.with_state((challenges, app))
}
async fn challenge(
State((ch, _)): State<(Arc<Challenges>, App)>,
UrlPath(token): UrlPath<String>,
) -> Response {
match ch.answer(&token) {
Some(key_auth) => (
[(header::CONTENT_TYPE, "application/octet-stream")],
key_auth,
)
.into_response(),
None => StatusCode::NOT_FOUND.into_response(),
}
}
async fn redirect(
State((_, app)): State<(Arc<Challenges>, App)>,
method: http::Method,
headers: HeaderMap,
uri: Uri,
) -> Response {
let Some(host) = headers
.get(header::HOST)
.and_then(|h| h.to_str().ok())
.map(|h| h.split(':').next().unwrap_or(h).to_string())
else {
return (StatusCode::BAD_REQUEST, "no Host header\n").into_response();
};
if let Some(www) = www_of(&host, &app.hosts()) {
return canonical_redirect(&www, &method, &uri);
}
let path = uri.path_and_query().map(|p| p.as_str()).unwrap_or("/");
(
StatusCode::PERMANENT_REDIRECT,
[(header::LOCATION, format!("https://{host}{path}"))],
)
.into_response()
}
/// **Serve :80** on `listener` for the life of the process.
pub async fn serve_http(listener: tokio::net::TcpListener, challenges: Arc<Challenges>) {
serve_http_for(listener, challenges, App::new(Router::new())).await
}
/// [`serve_http`] with the served names of `app` ([`http_router_for`]).
pub async fn serve_http_for(
listener: tokio::net::TcpListener,
challenges: Arc<Challenges>,
app: App,
) {
let app = http_router_for(challenges, app);
if let Err(e) = axum::serve(listener, app).await {
tracing::error!(error = %e, "web: the :80 listener stopped");
}
}
/// The rustls server config: the live resolver, HTTP/2 and HTTP/1.1 by ALPN.
pub fn tls_config(resolver: Arc<Resolver>) -> Result<rustls::ServerConfig, String> {
tls_config_with(resolver, None)
}
/// [`tls_config`] with the product's TLS provider (`None`: rustls' ring provider).
pub fn tls_config_with(
resolver: Arc<Resolver>,
provider: Option<Arc<rustls::crypto::CryptoProvider>>,
) -> Result<rustls::ServerConfig, String> {
let provider = provider.unwrap_or_else(|| Arc::new(rustls::crypto::ring::default_provider()));
let mut cfg = rustls::ServerConfig::builder_with_provider(provider)
.with_safe_default_protocol_versions()
.map_err(|e| format!("rustls: {e}"))?
.with_no_client_auth()
.with_cert_resolver(resolver);
cfg.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
Ok(cfg)
}
/// **Serve :443** on `listener` for the life of the process. Each connection
/// is TLS-accepted with the certificate current at its handshake and served
/// by the [`App`] current at its accept; the peer's address is in the
/// request's `ConnectInfo<SocketAddr>` (the console's client-IP rate limits
/// read it).
pub async fn serve_https(listener: tokio::net::TcpListener, resolver: Arc<Resolver>, app: App) {
serve_https_with(listener, resolver, app, None).await
}
/// [`serve_https`] with the product's TLS provider ([`super::Hooks::tls`]).
pub async fn serve_https_with(
listener: tokio::net::TcpListener,
resolver: Arc<Resolver>,
app: App,
provider: Option<Arc<rustls::crypto::CryptoProvider>>,
) {
let acceptor = match tls_config_with(resolver, provider) {
Ok(c) => tokio_rustls::TlsAcceptor::from(Arc::new(c)),
Err(e) => {
tracing::error!(error = %e, "web: no TLS configuration — :443 is not served");
return;
}
};
serve_https_limited(listener, acceptor, app, ConnLimits::DOOR).await
}
/// **The bounds of one :443 door** (RA4-11, RA5-22): connections at once, per
/// peer, and how long a handshake, a request's head and a silent connection
/// may take.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ConnLimits {
pub(crate) total: usize,
pub(crate) per_peer: usize,
pub(crate) handshake: std::time::Duration,
pub(crate) header_read: std::time::Duration,
pub(crate) idle: std::time::Duration,
}
impl ConnLimits {
/// The front's own.
pub(crate) const DOOR: ConnLimits = ConnLimits {
total: MAX_TLS_CONNS,
per_peer: MAX_TLS_CONNS_PER_PEER,
handshake: TLS_HANDSHAKE_TIMEOUT,
header_read: HEADER_READ_TIMEOUT,
idle: IDLE_TIMEOUT,
};
}
/// The most :443 connections one peer holds at once (RA5-22): an address, an
/// IPv6 one by its /64.
pub const MAX_TLS_CONNS_PER_PEER: usize = 64;
/// How long a TLS handshake may take.
pub const TLS_HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
/// How long a request's head may take to arrive, from its first byte on (RA5-22).
pub const HEADER_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(15);
/// How long a connection may stay silent, no byte either way, before it is
/// closed (RA5-22). A repository's slow pack keeps it alive with git's own
/// keep-alive (upload-pack sends one every 5 s by default).
pub const IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);
pub(crate) async fn serve_https_limited(
listener: tokio::net::TcpListener,
acceptor: tokio_rustls::TlsAcceptor,
app: App,
limits: ConnLimits,
) {
let conns = Arc::new(tokio::sync::Semaphore::new(limits.total));
let peers = Arc::new(Peers::default());
let mut refused_said = std::time::Instant::now() - std::time::Duration::from_secs(120);
loop {
let (tcp, peer) = match listener.accept().await {
Ok(x) => x,
Err(e) => {
tracing::warn!(error = %e, "web: accept on :443");
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
continue;
}
};
// RA5-22: at most `per_peer` connections of one peer (an address, an IPv6 /64), and
// RA4-11: at most `total` at once. Past either a connection is closed at once (said
// once per minute), so one host cannot take the door from every other visitor.
let Some(slot) = peers.take(peer.ip(), limits.per_peer) else {
if refused_said.elapsed() >= std::time::Duration::from_secs(60) {
tracing::warn!(%peer, max = limits.per_peer, "web: a :443 peer is at its connection cap; its connection was closed unanswered");
refused_said = std::time::Instant::now();
}
drop(tcp);
continue;
};
let Ok(permit) = Arc::clone(&conns).try_acquire_owned() else {
if refused_said.elapsed() >= std::time::Duration::from_secs(60) {
tracing::warn!(%peer, max = limits.total, "web: :443 is at its connection cap; a connection was closed unanswered");
refused_said = std::time::Instant::now();
}
drop(tcp);
continue;
};
let acceptor = acceptor.clone();
let router = app.router_now();
// A32-14, stays a task: one :443 connection, under the caps' permit and slot (RA4-11, RA5-22).
tokio::spawn(async move {
let _held = (permit, slot);
let tls = match tokio::time::timeout(limits.handshake, acceptor.accept(tcp)).await {
Ok(Ok(s)) => s,
Ok(Err(e)) => {
tracing::debug!(%peer, error = %e, "web: TLS handshake refused");
return;
}
Err(_) => return,
};
serve_conn(
Idle::new(tls, limits.idle),
peer,
router,
limits.header_read,
)
.await;
});
}
}
/// **The :443 connections each peer holds** (RA5-22), keyed by its address (an
/// IPv6 one by its /64: one host holds a whole /64).
#[derive(Default)]
struct Peers {
open: std::sync::Mutex<std::collections::HashMap<std::net::IpAddr, usize>>,
}
/// One connection of a peer; its count drops with it.
struct PeerSlot {
peers: Arc<Peers>,
key: std::net::IpAddr,
}
impl Peers {
fn key(ip: std::net::IpAddr) -> std::net::IpAddr {
use std::net::{IpAddr, Ipv6Addr};
match ip {
IpAddr::V6(v6) => match v6.to_ipv4_mapped() {
Some(v4) => IpAddr::V4(v4),
None => {
let s = v6.segments();
IpAddr::V6(Ipv6Addr::new(s[0], s[1], s[2], s[3], 0, 0, 0, 0))
}
},
v4 => v4,
}
}
/// A slot for `ip`, or `None` when it already holds `max`.
fn take(self: &Arc<Self>, ip: std::net::IpAddr, max: usize) -> Option<PeerSlot> {
let key = Peers::key(ip);
let mut open = self.open.lock().unwrap_or_else(|p| p.into_inner());
let n = open.entry(key).or_insert(0);
if *n >= max {
return None;
}
*n += 1;
Some(PeerSlot {
peers: Arc::clone(self),
key,
})
}
}
impl Drop for PeerSlot {
fn drop(&mut self) {
let mut open = self.peers.open.lock().unwrap_or_else(|p| p.into_inner());
if let Some(n) = open.get_mut(&self.key) {
*n = n.saturating_sub(1);
if *n == 0 {
open.remove(&self.key);
}
}
}
}
/// **A connection that ends when it falls silent** (RA5-22): no byte read or
/// written for `idle` and its next read or write fails (`TimedOut`), so hyper
/// closes it.
struct Idle<S> {
io: S,
idle: std::time::Duration,
deadline: std::pin::Pin<Box<tokio::time::Sleep>>,
}
impl<S> Idle<S> {
fn new(io: S, idle: std::time::Duration) -> Self {
Idle {
io,
idle,
deadline: Box::pin(tokio::time::sleep(idle)),
}
}
fn heard(&mut self) {
let at = tokio::time::Instant::now() + self.idle;
self.deadline.as_mut().reset(at);
}
fn silent<T>(
&mut self,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<T>> {
use std::future::Future as _;
match self.deadline.as_mut().poll(cx) {
std::task::Poll::Ready(()) => std::task::Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
"the connection was silent too long",
))),
std::task::Poll::Pending => std::task::Poll::Pending,
}
}
}
impl<S: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for Idle<S> {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let this = &mut *self;
let before = buf.filled().len();
match std::pin::Pin::new(&mut this.io).poll_read(cx, buf) {
std::task::Poll::Ready(r) => {
if buf.filled().len() > before {
this.heard();
}
std::task::Poll::Ready(r)
}
std::task::Poll::Pending => this.silent(cx),
}
}
}
impl<S: tokio::io::AsyncWrite + Unpin> tokio::io::AsyncWrite for Idle<S> {
fn poll_write(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::io::Result<usize>> {
let this = &mut *self;
match std::pin::Pin::new(&mut this.io).poll_write(cx, buf) {
std::task::Poll::Ready(Ok(n)) => {
if n > 0 {
this.heard();
}
std::task::Poll::Ready(Ok(n))
}
std::task::Poll::Ready(Err(e)) => std::task::Poll::Ready(Err(e)),
std::task::Poll::Pending => this.silent(cx),
}
}
fn poll_flush(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
let this = &mut *self;
match std::pin::Pin::new(&mut this.io).poll_flush(cx) {
std::task::Poll::Pending => this.silent(cx),
done => done,
}
}
fn poll_shutdown(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::io::Result<()>> {
std::pin::Pin::new(&mut self.io).poll_shutdown(cx)
}
}
/// The most :443 connections served at once (RA4-11).
pub const MAX_TLS_CONNS: usize = 1024;
async fn serve_conn<S>(io: S, peer: SocketAddr, router: Router, header_read: std::time::Duration)
where
S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static,
{
use tower::ServiceExt as _;
let svc = router.map_request(move |mut req: http::Request<hyper::body::Incoming>| {
req.extensions_mut().insert(ConnectInfo(peer));
req
});
let hyper_svc = hyper_util::service::TowerToHyperService::new(svc);
let mut builder =
hyper_util::server::conn::auto::Builder::new(hyper_util::rt::TokioExecutor::new());
// RA5-22: a request's head arrives within `header_read` of its first byte, or the connection ends.
builder
.http1()
.timer(hyper_util::rt::TokioTimer::new())
.header_read_timeout(header_read);
if let Err(e) = builder
.serve_connection_with_upgrades(hyper_util::rt::TokioIo::new(io), hyper_svc)
.await
{
tracing::debug!(%peer, error = %e, "web: connection ended");
}
}
/// **A payment vendor's webhook** (LANE-A-API.md "stripe plugin for the
/// front", lane E 3): `POST https://<web host>/callback/<vendor>` — or, per
/// product instance, `/callback/<vendor>/<instance>` (`[vendors.stripe.<instance>]`,
/// monetize's vendor id `stripe/<instance>`) — is handed to monetize on
/// loopback (`http://<to>/callback/<vendor>[/<instance>]`, HTTP/1.1), the body
/// byte for byte and the headers intact — `Stripe-Signature` is what monetize's
/// hmac door verifies — and monetize's status handed back (401 = not verified,
/// 200 = taken or ignored). A third segment is not a callback (404). Every
/// other path is the web face, unchanged.
pub fn with_callbacks(face: Router, to: SocketAddr) -> Router {
Router::new()
.route("/callback/{vendor}", axum::routing::post(callback))
.route(
"/callback/{vendor}/{instance}",
axum::routing::post(instance_callback),
)
.with_state(to)
.fallback_service(face)
}
/// Headers that belong to one hop, never forwarded.
const HOP_BY_HOP: &[&str] = &[
"connection",
"keep-alive",
"proxy-connection",
"transfer-encoding",
"te",
"trailer",
"upgrade",
"host",
"content-length",
];
async fn callback(
State(to): State<SocketAddr>,
UrlPath(vendor): UrlPath<String>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Response {
carry_callback(to, &[&vendor], headers, body).await
}
async fn instance_callback(
State(to): State<SocketAddr>,
UrlPath((vendor, instance)): UrlPath<(String, String)>,
headers: HeaderMap,
body: axum::body::Bytes,
) -> Response {
carry_callback(to, &[&vendor, &instance], headers, body).await
}
/// Every segment is checked before anything is forwarded: a decoded `/`, `..`
/// or an empty segment never reaches monetize.
async fn carry_callback(
to: SocketAddr,
segments: &[&str],
headers: HeaderMap,
body: axum::body::Bytes,
) -> Response {
if segments.iter().any(|s| {
s.is_empty()
|| !s
.chars()
.all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
}) {
return (StatusCode::NOT_FOUND, "no such vendor\n").into_response();
}
let vendor = segments.join("/");
let mut req = http::Request::post(format!("http://{to}/callback/{vendor}"));
for (k, v) in headers.iter() {
if !HOP_BY_HOP.contains(&k.as_str()) {
req = req.header(k, v);
}
}
let Ok(req) = req.body(axum::body::Body::from(body)) else {
return (StatusCode::BAD_REQUEST, "a header could not be carried\n").into_response();
};
let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new())
.build_http::<axum::body::Body>();
match tokio::time::timeout(std::time::Duration::from_secs(30), client.request(req)).await {
Ok(Ok(resp)) => {
let (parts, body) = resp.into_parts();
Response::from_parts(parts, axum::body::Body::new(body))
}
Ok(Err(e)) => {
tracing::warn!(%vendor, error = %e, "web: callback: monetize did not answer");
(StatusCode::BAD_GATEWAY, "monetize did not answer\n").into_response()
}
Err(_) => (
StatusCode::GATEWAY_TIMEOUT,
"monetize did not answer in time\n",
)
.into_response(),
}
}