crates/monetize-product-wasm/src/doors.rs
//! **The doors** — what a guest may ask the host for, each behind a grant.
//!
//! The host does the socket work: HTTPS with rustls under a host allow-list,
//! gRPC unary over plaintext HTTP/2 with the block's bearer (a signup guest's), and an
//! environment read limited to the names the block listed. (The signing door and its identity
//! key went in monetize v2.0 P2 T13, A28 M3.) The guest sends a [`wire::frame`] and gets one
//! back; `0` from a door means the door itself failed and the reason is in
//! the faults, which surface after the call as a named refusal.
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use bytes::{Buf, BufMut};
use guest_host::module::{fault, read_arg, write_to_wasm, GuestState};
use guest_host::BoundedLimiter;
use monetize_product::HttpAuth;
use monetize_product_guest::wire::{frame, GrpcError, GrpcHead, HmacHead, HostHead, HttpHead, StateKey, StreamOpen, StreamRef};
use monetize_product_guest::{grants, imports};
use guest_host::wasmtime::{self, Caller, Linker};
/// How large a gRPC answer a guest may receive: the linked holger plugin's number.
pub const MAX_GRPC_DECODE: usize = 64 * 1024 * 1024;
/// **How many keys the host will remember for one product**, and how large
/// each may be. A guest is a fresh instance per call and cannot remember a
/// nonce it has honoured or the phase a restore reached; the host does it,
/// which is also what stops a guest forging its own history. Bounded because a
/// guest that could write without limit would grow the server.
pub const MAX_STATE_KEYS: usize = 4_096;
pub const MAX_STATE_BYTES: usize = 64 * 1024;
/// **The largest HTTP answer a guest may receive.** Streamed and counted, never
/// buffered past the cap: a host that read whatever a peer sent would let one
/// answer grow the server.
pub const MAX_HTTP_BODY: usize = 16 * 1024 * 1024;
/// **How many redirects the HTTP door follows**, and only while every hop is in
/// `http_hosts`: a redirect is a request to wherever the peer says, and the
/// allow-list is the operator's, not the peer's.
pub const MAX_REDIRECTS: usize = 5;
/// **How many signatures one `host_hmac_verify` compares.** Stripe sends one per
/// active secret during a rotation — two, rarely three.
pub const MAX_HMAC_CANDIDATES: usize = 8;
/// **How many streams one guest call may hold open**, and the number is the
/// restore: the primary being filled and the twin being read.
pub const MAX_OPEN_STREAMS: usize = 2;
/// The gRPC door: one lazy channel to the block's `endpoint`, the bearer added
/// on every call.
pub struct GrpcDoor {
channel: tonic::transport::Channel,
bearer: Option<tonic::metadata::MetadataValue<tonic::metadata::Ascii>>,
endpoint: String,
/// The OTHER addresses this block allows (`guest_grpc_hosts`). A restore
/// reads the twin's log while it fills the primary, and both are named by
/// an operator or neither is dialled.
also: Vec<String>,
/// The CA a `https://` address is verified against; `None` is the
/// compiled-in Mozilla root set. The twin is verified the same way the
/// primary is — an estate's second box is not a second trust decision.
tls_ca: Option<PathBuf>,
/// The client identity presented on every dial (f-grpc-tls), cert + key PEM paths.
tls_client: Option<(PathBuf, PathBuf)>,
timeout: Duration,
}
/// **A client identity for the control plane** (`tls_client_cert` +
/// `tls_client_key` in the block): presented on every https dial, so a
/// product that REQUIRES a client certificate — gunnar's self-minted issuer,
/// an estate attached from another cloud that trusts the central's attach CA
/// (f-grpc-tls, f-attach-enrol) — can be reached with mutual TLS.
#[derive(Clone, Copy, Debug)]
pub struct TlsClient<'a> {
pub cert: &'a Path,
pub key: &'a Path,
}
/// How often an idle gRPC connection of the door is pinged (HTTP/2 PING) —
/// gunnar-ui's own interval for the same main.
pub const KEEPALIVE_EVERY: Duration = Duration::from_secs(10);
/// How long a ping may go unanswered before the connection is given up.
///
/// Twenty seconds, not five (R5.1-G, 2026-10-07): a gunnar whose control-plane
/// workers were all in slow synchronous calls answered no PING for seconds, and
/// five seconds killed the connection with the `Entitlement.Set` in flight on
/// it (an `Internal` stream reset). A gunnar that is gone still answers RST at
/// once; one gone in silence is noticed after `KEEPALIVE_EVERY + KEEPALIVE_TIMEOUT`.
pub const KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
/// **One channel, dialled the way the scheme says.**
///
/// `https://` is verified — against the CA the block names, or the compiled-in
/// Mozilla root set when it names none. `http://` is plaintext, and naming a CA
/// for a plaintext endpoint is REFUSED by name rather than ignored: a block
/// that wrote `tls_ca` believes its traffic is verified, and the one thing that
/// must not happen is the host quietly not verifying it.
///
/// The roots are compiled in rather than read out of the box's trust store, so
/// verification is the same on every machine; an estate with its own CA names
/// the PEM in its own block instead of installing it system-wide.
fn dial(uri: &str, tls_ca: Option<&Path>, tls_client: Option<(&Path, &Path)>, timeout: Duration) -> Result<tonic::transport::Channel, String> {
let https = uri.starts_with("https://");
if !https {
if let Some((cert, _)) = tls_client {
return Err(format!(
"{uri:?} is a PLAINTEXT endpoint and `tls_client_cert` names {}: a client identity is only presented \
inside TLS, so this block believes it authenticates and does not. Write the endpoint as `https://…`, \
or remove `tls_client_cert`/`tls_client_key`",
cert.display()
));
}
if let Some(ca) = tls_ca {
return Err(format!(
"{uri:?} is a PLAINTEXT endpoint and `tls_ca` names {}: nothing on a plaintext connection is \
verified against a CA, so this block believes it is protected and is not. Write the endpoint \
as `https://…`, or remove `tls_ca`",
ca.display()
));
}
}
let mut ep = tonic::transport::Channel::from_shared(uri.to_string())
.map_err(|e| format!("{uri:?} is not a usable control-plane address: {e}"))?
.connect_timeout(Duration::from_secs(5))
.timeout(timeout)
// **A peer that went away in silence is noticed while idle** (the
// follow-up of recharge-main-front-cut-off, as gunnar-ui's grpc.rs
// a891f91): a main that rebooted forgets its connections without a
// FIN; the ping is answered RST (or not at all) and the channel dials
// again, instead of the next call being written into the dead
// connection (`transport error`). The DoorRuntime is a driven worker,
// so the ping is sent while no guest call is running.
.http2_keep_alive_interval(KEEPALIVE_EVERY)
.keep_alive_timeout(KEEPALIVE_TIMEOUT)
.keep_alive_while_idle(true);
if https {
let mut tls = match tls_ca {
None => tonic::transport::ClientTlsConfig::new().with_webpki_roots(),
Some(ca) => {
let pem = std::fs::read(ca).map_err(|e| format!("cannot read the CA at {}: {e}", ca.display()))?;
tonic::transport::ClientTlsConfig::new().ca_certificate(tonic::transport::Certificate::from_pem(pem))
}
};
if let Some((cert, key)) = tls_client {
let c = std::fs::read(cert).map_err(|e| format!("cannot read the client certificate at {}: {e}", cert.display()))?;
let k = std::fs::read(key).map_err(|e| format!("cannot read the client key at {}: {e}", key.display()))?;
tls = tls.identity(tonic::transport::Identity::from_pem(c, k));
}
ep = ep.tls_config(tls).map_err(|e| format!("{uri:?}: the TLS configuration was refused: {e}"))?;
}
Ok(ep.connect_lazy())
}
/// **One client-streaming call in flight**, held by the host for the length of
/// ONE guest call.
///
/// A guest is a fresh instance per call, so a handle that outlived one could
/// never be closed by the guest that opened it — every stream is therefore
/// opened, fed and closed (or cancelled, or dropped) inside a single
/// `plugin_call`, and [`Doors::sweep_streams`] cancels whatever is left.
pub struct OpenStream {
kind: StreamKind,
/// The guest call that opened it ([`CallState::call`]); `0` for none named.
owner: u64,
path: String,
/// When this stream must be finished by — the call's own deadline, never
/// longer.
deadline: std::time::Instant,
chunks: u64,
bytes: u64,
}
/// The two directions a guest can hold open. A restore needs ONE of each: the
/// primary it fills, and the twin it reads the log off.
pub enum StreamKind {
/// Client-streaming: the guest pushes, the peer answers once at the close.
Sending {
/// `None` once the sender has been dropped at the close.
tx: Option<tokio::sync::mpsc::Sender<Vec<u8>>>,
join: tokio::task::JoinHandle<Result<Vec<u8>, GrpcError>>,
},
/// Server-streaming: one request went out with the open, and the guest
/// reads until the peer FINISHES.
Receiving { rx: Box<tonic::Streaming<Vec<u8>>> },
}
/// The streams one guest has open. Small on purpose: a restore pushes through
/// ONE.
#[derive(Default)]
pub struct Streams {
next: u64,
open: std::collections::BTreeMap<u64, OpenStream>,
}
/// A client built for one client certificate, and the `(len, mtime)` of the
/// certificate file it was built from.
type CertClient = ((u64, std::time::SystemTime), reqwest::Client);
/// The HTTP door: rustls, a per-call timeout, the hosts the block allows, and
/// the credential the HOST adds per host (`http_auth`) from a secret file.
pub struct HttpDoor {
client: reqwest::Client,
hosts: Vec<String>,
auth: Vec<HttpAuth>,
secrets_dir: Option<PathBuf>,
timeout: Duration,
/// One client per host that presents a client certificate, keyed by the
/// `http_auth` entry's host and rebuilt when the certificate file changes.
cert_clients: std::sync::Mutex<std::collections::BTreeMap<String, CertClient>>,
/// The largest body read for a guest ([`MAX_HTTP_BODY`] unless a test narrows it).
body_max: usize,
/// `http_resolve`, kept for the per-certificate clients.
resolve: Vec<String>,
/// Whose door this is in a log line (`vendors.stripe`, `clouds.upcloud`).
label: String,
/// Where the one line per call goes: stderr, which the front reads as its
/// log (a line saying FAILED is raised to WARN there).
say: Arc<dyn Fn(&str) + Send + Sync>,
}
/// How long the same quiet read is not said again (PROD-FIX 3).
pub const QUIET_READ_EVERY: Duration = Duration::from_secs(10 * 60);
/// A read that takes this long is said every time (the slow call is what a log is read for).
pub const SLOW_READ: Duration = Duration::from_secs(1);
/// **The quiet reads a door said lately** (PROD-FIX 3, prod 2026-10-26: `GET /1.3/storage/<uuid>
/// -> 200` several times a second filled the box's 2000-line log ring in about 3 minutes). A
/// quiet read is a GET or HEAD answered 2xx within [`SLOW_READ`]; the same one (method,
/// endpoint, status) is said once per [`QUIET_READ_EVERY`]. A write, a failure, another status
/// or a slow read is said every time.
#[derive(Default)]
pub struct QuietReads(std::sync::Mutex<std::collections::HashMap<String, std::time::Instant>>);
impl QuietReads {
/// Is the call worth its line? Remembers the quiet reads it lets through.
pub fn say(&self, method: &str, endpoint_line: &str, outcome: Result<u16, &str>, took: Duration) -> bool {
let quiet = matches!(method.to_ascii_uppercase().as_str(), "GET" | "HEAD") && matches!(outcome, Ok(s) if (200..300).contains(&s)) && took < SLOW_READ;
if !quiet {
return true;
}
let now = std::time::Instant::now();
let mut seen = self.0.lock().unwrap_or_else(|e| e.into_inner());
seen.retain(|_, at| now.duration_since(*at) < QUIET_READ_EVERY);
match seen.get(endpoint_line) {
Some(_) => false,
None => {
seen.insert(endpoint_line.to_string(), now);
true
}
}
}
}
/// The quiet reads of this process: one memory over every door (a guest's doors may be opened
/// again; its polls are the same reads).
fn quiet_reads() -> &'static QuietReads {
static Q: std::sync::OnceLock<QuietReads> = std::sync::OnceLock::new();
Q.get_or_init(QuietReads::default)
}
/// The key of a quiet read: its line without the time it took.
fn quiet_key(line: &str) -> &str {
line.rsplit_once(" in ").map(|(k, _)| k).unwrap_or(line)
}
/// stderr, the log a monetize-server's lines are read from.
fn say_on_stderr() -> Arc<dyn Fn(&str) + Send + Sync> {
Arc::new(|l: &str| eprintln!("{l}"))
}
/// **The one line a call through the HTTP door says** (test2 2026-09-30: a
/// 25-33 s signup whose log had nothing between two lines 25 s apart). The
/// endpoint is the URL without its query or fragment — no secret a guest put
/// there reaches the log, and the host's own credential is a header, never in
/// the URL. `Ok(status)` is an answer; `Err(why)` is a call that got none, and
/// says FAILED (WARN on the front), with whether the door's deadline ended it.
pub fn door_line(label: &str, method: &str, url: &str, outcome: Result<u16, &str>, took: Duration, deadline: Duration) -> String {
let endpoint = match reqwest::Url::parse(url) {
Ok(mut u) => {
u.set_query(None);
u.set_fragment(None);
let _ = u.set_username("");
let _ = u.set_password(None);
u.to_string()
}
Err(_) => "<an unparsable URL>".to_string(),
};
let ms = took.as_millis();
// A transport error quotes the URL it was sending to, query and all.
let scrub = |why: &str| -> String {
let mut w = why.replace(url, &endpoint);
if let Ok(u) = reqwest::Url::parse(url) {
w = w.replace(u.as_str(), &endpoint);
}
w
};
match outcome {
Ok(status) => format!("http door [{label}]: {method} {endpoint} -> {status} in {ms} ms"),
Err(why) => {
let ended = if took + Duration::from_millis(50) >= deadline {
format!("the door's {} ms deadline ended it", deadline.as_millis())
} else {
"no answer".to_string()
};
format!("http door [{label}]: {method} {endpoint} -> FAILED after {ms} ms ({ended}): {}", scrub(why))
}
}
}
/// **Everything a guest's doors are opened from** — the block, as the host
/// reads it. [`Doors::open`] is this with no auth, no HMAC and no secrets.
pub struct DoorSpec<'a> {
pub product: &'a str,
pub endpoint: &'a str,
pub bearer: Option<&'a str>,
pub twin: Option<&'a str>,
pub tls_ca: Option<&'a Path>,
/// The client identity the gRPC door presents (f-grpc-tls).
pub tls_client: Option<TlsClient<'a>>,
pub grants: &'a [String],
pub http_hosts: &'a [String],
pub grpc_hosts: &'a [String],
pub call_timeout: Duration,
pub http_auth: &'a [HttpAuth],
pub hmac_secrets: &'a [String],
pub secrets_dir: Option<&'a Path>,
/// `http_resolve = ["<name>=<ip>", …]`: where the HTTP door connects for a
/// host NAME (reqwest `resolve`; the URL's port is kept). The allow-list
/// still judges the URL as written. Empty: the system resolver.
pub http_resolve: &'a [String],
}
/// One HTTP client, following redirects only inside `hosts` and at most
/// [`MAX_REDIRECTS`] times, presenting `identity` when there is one.
fn http_client(hosts: &[String], timeout: Duration, identity: Option<reqwest::Identity>, resolve: &[String]) -> Result<reqwest::Client, String> {
let allow = hosts.to_vec();
let policy = reqwest::redirect::Policy::custom(move |attempt| {
if attempt.previous().len() >= MAX_REDIRECTS {
return attempt.error(format!("[http-redirect-too-many] more than {MAX_REDIRECTS} redirects"));
}
match HttpDoor::allowed(&allow, attempt.url()) {
Ok(()) => attempt.follow(),
Err(e) => attempt.error(format!("[http-redirect-not-allowed] a redirect leaves the allow-list, not followed: {e}")),
}
});
let mut b = tls_roots(reqwest::Client::builder()).timeout(timeout).connect_timeout(Duration::from_secs(5)).redirect(policy);
for (name, addr) in resolve_pairs(resolve)? {
b = b.resolve(&name, addr);
}
if let Some(id) = identity {
b = b.identity(id);
}
b.build().map_err(|e| format!("cannot build the HTTP door: {e}"))
}
/// **`http_resolve`, parsed**, pure: each `<name>=<ip>` a host NAME (never an
/// address: an IP-literal URL is dialled as written) and the address the door
/// connects to for it; the port is the URL's. Anything else is refused by name.
pub fn resolve_pairs(entries: &[String]) -> Result<Vec<(String, std::net::SocketAddr)>, String> {
entries
.iter()
.map(|e| {
let (name, ip) = e.split_once('=').ok_or_else(|| format!("[http-resolve] `{e}` is not <name>=<ip>"))?;
let (name, ip) = (name.trim(), ip.trim());
if name.is_empty() || name.parse::<std::net::IpAddr>().is_ok() {
return Err(format!("[http-resolve] `{e}`: the left side must be a host NAME"));
}
let ip: std::net::IpAddr = ip.parse().map_err(|_| format!("[http-resolve] `{e}`: the right side must be an IP address"))?;
Ok((name.to_ascii_lowercase(), std::net::SocketAddr::new(ip, 0)))
})
.collect()
}
/// An error and every cause under it, on one line: reqwest puts the reason a
/// redirect was refused in the SOURCE, not in the top message.
fn chain(e: &dyn std::error::Error) -> String {
let mut s = e.to_string();
let mut cur = e.source();
while let Some(c) = cur {
let t = c.to_string();
if !s.contains(&t) {
s.push_str(": ");
s.push_str(&t);
}
cur = c.source();
}
s
}
/// The bugfix lane's names for the same two ceilings.
pub const HTTP_BODY_MAX: usize = MAX_HTTP_BODY;
pub const HTTP_REDIRECTS_MAX: usize = MAX_REDIRECTS;
/// **The runtime a guest's doors run on**: ONE worker thread that drives the
/// sockets ALL the time, not only while some call is inside `block_on`.
///
/// It was a current-thread runtime, which runs only inside a caller's
/// `block_on`. Between two calls nobody read the pooled connections, so a
/// keep-alive connection the peer closed while the shop sat idle stayed in the
/// pool looking open, and the next request was written into it: "connection
/// closed before message completed" — a signup refused as "the shop is down"
/// the first time Stripe was asked after an idle spell (RED
/// `a_connection_the_peer_closed_while_idle_is_not_reused`). With a worker, the
/// connection task reads the peer's FIN when it arrives and the pool drops it.
///
/// Dropped with `shutdown_background`, never by joining the worker: the last
/// `Arc` to a replaced guest can be released on an async thread, where a
/// blocking shutdown panics.
pub struct DoorRuntime(Option<tokio::runtime::Runtime>);
impl DoorRuntime {
fn start(product: &str) -> Result<DoorRuntime, String> {
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(1)
.max_blocking_threads(4)
.thread_name(format!("door-{product}"))
.enable_all()
.build()
.map_err(|e| format!("cannot start the guest's runtime: {e}"))?;
Ok(DoorRuntime(Some(rt)))
}
fn rt(&self) -> &tokio::runtime::Runtime {
self.0.as_ref().expect("the door runtime lives until the doors drop")
}
pub fn block_on<F: std::future::Future>(&self, f: F) -> F::Output {
self.rt().block_on(f)
}
pub fn handle(&self) -> &tokio::runtime::Handle {
self.rt().handle()
}
}
impl Drop for DoorRuntime {
fn drop(&mut self) {
if let Some(rt) = self.0.take() {
rt.shutdown_background();
}
}
}
/// Everything the doors need, shared by every call into one guest.
pub struct Doors {
pub rt: DoorRuntime,
pub product: String,
pub grpc: Option<GrpcDoor>,
pub http: Option<HttpDoor>,
pub bearer: Option<String>,
/// What the typed `host_endpoint` door answers (the test kind's): the block's `endpoint`,
/// verbatim. The generic `env` door that once answered it, and the key, the bearer, the
/// twin and the appliance key by name, is retired (owner decision 6, 2026-10-26).
pub endpoint: String,
/// The block's `restoring_from_twin`, which is on the gRPC allow-list.
pub twin: Option<String>,
/// The secrets `host_hmac_verify` may use, by name (`hmac_secrets`).
pub hmac_secrets: Vec<String>,
/// Where `secret.<name>` files are read from.
pub secrets_dir: Option<PathBuf>,
pub granted_imports: Vec<&'static str>,
/// **The host's memory for this product.** In memory only, and on purpose:
/// it is the same lifetime the linked plugin's own `seen_nonces` and
/// restore cell have, so a restart forgets exactly as much as it always
/// did and no more.
///
/// `Arc` so a guest REPLACED live ([`crate::hot::HotGuest::replace`]) keeps it:
/// the old instance and the new one hold the same map, so a nonce honoured by
/// a call still finishing on the old guest is seen by the new one. A
/// replacement is not a restart, and must not forget like one.
pub state: std::sync::Arc<std::sync::Mutex<std::collections::BTreeMap<String, Vec<u8>>>>,
/// The client-streaming calls in flight, and the deadline they are all
/// measured against.
pub streams: std::sync::Mutex<Streams>,
pub call_timeout: Duration,
/// The largest value one state key may hold — [`MAX_STATE_BYTES`] for a
/// product. A cloud's pool ledger is one key that grows by one row per
/// sold resource, so the cloud host raises it for its kind.
pub state_max_bytes: usize,
/// **A key written THROUGH, not remembered**: while a host holds one here, a
/// `state_put` of that key is handed to the sink and answers only once the sink
/// did — the value is never kept in [`Doors::state`]. The cloud host installs
/// one for the length of a journaled growth call, so every step the guest
/// journals is fsync'd on the host BEFORE the guest takes it.
///
/// **One slot per call** ([`CallState::call`]): two journaled calls on two
/// lanes of one guest (a lay and a growth) each write through to their own
/// journal, and neither waits for the other's slot. Key `0` is "no call
/// named" ([`Doors::state_put`]).
pub write_through: std::sync::Mutex<std::collections::BTreeMap<u64, WriteThrough>>,
}
/// See [`Doors::write_through`].
pub struct WriteThrough {
pub key: String,
pub sink: WriteSink,
}
/// Where a [`WriteThrough`] puts each answer's bytes.
pub type WriteSink = Box<dyn Fn(&[u8]) -> Result<(), String> + Send + Sync>;
impl Doors {
/// One unary gRPC call through this block's door, from outside a guest —
/// what a test drives the dial with.
#[doc(hidden)]
pub fn grpc_unary_for_test(&self, path: &str, body: Vec<u8>) -> Result<Vec<u8>, String> {
let g = self.grpc.as_ref().ok_or("the block grants no grpc door")?;
self.rt.block_on(g.unary(path, body, &[], None)).map_err(|e| e.message)
}
/// Build the doors a block grants. Runs the runtime-touching parts on a
/// fresh thread, so this is callable from inside another runtime.
#[allow(clippy::too_many_arguments)]
pub fn open(
product: &str,
endpoint: &str,
bearer: Option<&str>,
twin: Option<&str>,
tls_ca: Option<&Path>,
grants_list: &[String],
http_hosts: &[String],
grpc_hosts: &[String],
call_timeout: Duration,
) -> Result<Doors, String> {
Doors::open_with(DoorSpec {
product,
endpoint,
bearer,
twin,
tls_ca,
tls_client: None,
grants: grants_list,
http_hosts,
grpc_hosts,
call_timeout,
http_auth: &[],
hmac_secrets: &[],
secrets_dir: None,
http_resolve: &[],
})
}
/// Build the doors a block grants, with its per-host auth, its HMAC
/// secrets and where its secrets are.
pub fn open_with(spec: DoorSpec<'_>) -> Result<Doors, String> {
let DoorSpec {
product,
endpoint,
bearer,
twin,
tls_ca,
tls_client,
grants: grants_list,
http_hosts,
grpc_hosts,
call_timeout,
http_auth,
hmac_secrets,
secrets_dir,
http_resolve,
} = spec;
let client = tls_client.map(|c| (c.cert.to_path_buf(), c.key.to_path_buf()));
let resolve = http_resolve.to_vec();
let auth = http_auth.to_vec();
let secrets_kept = secrets_dir.map(Path::to_path_buf);
let secrets_for_http = secrets_kept.clone();
let endpoint_kept = endpoint.to_string();
let wants_grpc = grants_list.iter().any(|g| g == grants::GRPC);
let wants_http = grants_list.iter().any(|g| g == grants::HTTP);
let endpoint = endpoint.to_string();
let hosts = http_hosts.to_vec();
// **The twin is not written twice.** Naming `restoring_from_twin` is
// what puts the address on the gRPC door's allow-list; an operator who
// had to repeat it in `guest_grpc_hosts` would eventually repeat it
// wrongly, and the lane would refuse an address the block already had.
let mut grpc_hosts = grpc_hosts.to_vec();
if let Some(t) = twin.map(str::trim).filter(|t| !t.is_empty()) {
if !grpc_hosts.iter().any(|h| h == t) {
grpc_hosts.push(t.to_string());
}
}
let twin_kept = twin.map(str::trim).filter(|t| !t.is_empty()).map(str::to_string);
let bearer_s = bearer.map(str::to_string);
let ca = tls_ca.map(PathBuf::from);
let product_label = product.to_string();
// On a thread of its own, joined at once: the runtime is started and its doors opened
// off the caller's thread, which may be inside an async runtime already.
let (rt, grpc, http) = std::thread::scope(|s| {
// A32-14, stays a task: a scoped thread, joined at once (the doors open off the caller's runtime).
s.spawn(move || {
let rt = DoorRuntime::start(&product_label)?;
let grpc = if wants_grpc { Some(rt.block_on(GrpcDoor::open(&endpoint, bearer_s.as_deref(), grpc_hosts, ca, client, call_timeout))?) } else { None };
let http = if wants_http {
let client = http_client(&hosts, call_timeout, None, &resolve)?;
Some(HttpDoor { client, hosts, auth, secrets_dir: secrets_for_http, timeout: call_timeout, cert_clients: Default::default(), body_max: MAX_HTTP_BODY, resolve, label: product_label, say: say_on_stderr() })
} else {
None
};
Ok::<_, String>((rt, grpc, http))
})
.join()
})
.map_err(|_| "the guest's runtime thread panicked".to_string())??;
let mut granted_imports: Vec<&'static str> = vec![imports::CLOCK, imports::LOG, imports::SLEEP];
for g in grants_list {
let imps = grants::imports(g);
if imps.is_empty() {
return Err(format!("`grants` names `{g}`, which is not a host door; the doors are {}", grants::VOCABULARY.join(", ")));
}
granted_imports.extend(imps);
}
Ok(Doors {
rt,
product: product.to_string(),
grpc,
http,
bearer: bearer.map(str::to_string),
endpoint: endpoint_kept,
twin: twin_kept,
hmac_secrets: hmac_secrets.to_vec(),
secrets_dir: secrets_kept,
granted_imports,
state: std::sync::Arc::new(std::sync::Mutex::new(std::collections::BTreeMap::new())),
streams: std::sync::Mutex::new(Streams::default()),
call_timeout,
state_max_bytes: MAX_STATE_BYTES,
write_through: std::sync::Mutex::new(std::collections::BTreeMap::new()),
})
}
}
impl GrpcDoor {
async fn open(endpoint: &str, bearer: Option<&str>, also: Vec<String>, tls_ca: Option<PathBuf>, tls_client: Option<(PathBuf, PathBuf)>, timeout: Duration) -> Result<GrpcDoor, String> {
let uri = if endpoint.contains("://") { endpoint.to_owned() } else { format!("http://{endpoint}") };
let channel = dial(&uri, tls_ca.as_deref(), tls_client.as_ref().map(|(c, k)| (c.as_path(), k.as_path())), timeout)?;
let bearer = match bearer.map(str::trim).filter(|t| !t.is_empty()) {
Some(t) => Some(
format!("Bearer {t}")
.parse()
.map_err(|_| "the configured bearer token is not a valid header value".to_string())?,
),
None => None,
};
Ok(GrpcDoor { channel, bearer, endpoint: endpoint.to_string(), also, tls_ca, tls_client, timeout })
}
/// One unary call: `path` is `/package.Service/Method`, the body is the
/// prost-encoded request, the answer the prost-encoded response.
pub async fn unary(&self, path: &str, body: Vec<u8>, headers: &[(String, String)], endpoint: Option<&str>) -> Result<Vec<u8>, GrpcError> {
// The endpoint is the HOST's knowledge — it never crosses the ABI — so
// every status carries it out of here. Without this a guest's refusal
// reads "holger: tcp connect error" and names no box an operator can go
// and look at.
let named = endpoint.unwrap_or(&self.endpoint).to_string();
let status = move |s: tonic::Status| GrpcError { code: format!("{:?}", s.code()), message: format!("{named}: {}", s.message()) };
let path = tonic::codegen::http::uri::PathAndQuery::try_from(path.to_string())
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("{path:?} is not a gRPC path: {e}") })?;
let channel = self.channel_for(endpoint)?;
let mut grpc = tonic::client::Grpc::new(channel).max_decoding_message_size(MAX_GRPC_DECODE);
grpc.ready().await.map_err(|e| GrpcError { code: "Unavailable".into(), message: format!("{}: {e}", endpoint.unwrap_or(&self.endpoint)) })?;
let mut req = tonic::Request::new(body);
if let Some(b) = &self.bearer {
req.metadata_mut().insert("authorization", b.clone());
}
// The guest's own metadata for this one call: a credential it asked the
// host to mint, and nothing else it could have — the guest holds no key.
for (k, v) in headers {
let key: tonic::metadata::MetadataKey<tonic::metadata::Ascii> = k
.parse()
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("{k:?} is not a metadata name: {e}") })?;
let value: tonic::metadata::MetadataValue<tonic::metadata::Ascii> = v
.parse()
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("the value for {k:?} is not a header value: {e}") })?;
req.metadata_mut().insert(key, value);
}
grpc.unary(req, path, RawCodec).await.map(|r| r.into_inner()).map_err(status)
}
}
impl GrpcDoor {
/// **Open a client-streaming call**: the request stream is a channel the
/// host feeds from `host_grpc_stream_send`, and the peer's single answer
/// comes back at the close.
///
/// The call runs as a task on the guest's own current-thread runtime, so
/// it makes progress exactly while the guest is inside a door — which is
/// the only time anything should be moving.
async fn open_stream(&self, path: &str, headers: &[(String, String)], endpoint: Option<&str>) -> Result<(tokio::sync::mpsc::Sender<Vec<u8>>, tokio::task::JoinHandle<Result<Vec<u8>, GrpcError>>), GrpcError> {
let path_q = tonic::codegen::http::uri::PathAndQuery::try_from(path.to_string())
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("{path:?} is not a gRPC path: {e}") })?;
let channel = self.channel_for(endpoint)?;
let mut grpc = tonic::client::Grpc::new(channel).max_decoding_message_size(MAX_GRPC_DECODE);
grpc.ready().await.map_err(|e| GrpcError { code: "Unavailable".into(), message: format!("{}: {e}", endpoint.unwrap_or(&self.endpoint)) })?;
// A small queue: back-pressure is the point. A guest that could push
// faster than the peer accepts would be buffering a whole transaction
// log in the host's memory.
let (tx, rx) = tokio::sync::mpsc::channel::<Vec<u8>>(4);
let mut req = tonic::Request::new(tokio_stream::wrappers::ReceiverStream::new(rx));
if let Some(b) = &self.bearer {
req.metadata_mut().insert("authorization", b.clone());
}
for (k, v) in headers {
let key: tonic::metadata::MetadataKey<tonic::metadata::Ascii> = k
.parse()
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("{k:?} is not a metadata name: {e}") })?;
let value: tonic::metadata::MetadataValue<tonic::metadata::Ascii> = v
.parse()
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("the value for {k:?} is not a header value: {e}") })?;
req.metadata_mut().insert(key, value);
}
let endpoint = endpoint.unwrap_or(&self.endpoint).to_string();
// A32-14, stays a task: one guest's streaming call; its handle is joined.
let join = tokio::spawn(async move {
grpc.client_streaming(req, path_q, RawCodec)
.await
.map(|r| r.into_inner())
.map_err(|s| GrpcError { code: format!("{:?}", s.code()), message: format!("{endpoint}: {}", s.message()) })
});
Ok((tx, join))
}
}
impl GrpcDoor {
/// **A channel to one of the addresses this block allows.** `None` is the
/// block's own endpoint. Anything else must be in `guest_grpc_hosts`: a
/// guest that could dial an address nobody wrote down could take a
/// tenant's log somewhere else.
fn channel_for(&self, endpoint: Option<&str>) -> Result<tonic::transport::Channel, GrpcError> {
let Some(want) = endpoint else { return Ok(self.channel.clone()) };
if want == self.endpoint {
return Ok(self.channel.clone());
}
if !self.also.iter().any(|a| a == want) {
return Err(GrpcError {
code: "PermissionDenied".into(),
message: format!(
"{want:?} is not an address this guest may dial; the block allows {} and guest_grpc_hosts \
names [{}]",
self.endpoint,
self.also.join(", ")
),
});
}
let uri = if want.contains("://") { want.to_owned() } else { format!("http://{want}") };
// The twin is verified the way the primary is: one trust decision per
// block, not one per box.
dial(&uri, self.tls_ca.as_deref(), self.tls_client.as_ref().map(|(c, k)| (c.as_path(), k.as_path())), self.timeout)
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: e })
}
/// **Open a server-streaming call**: one request out, many answers back.
async fn open_server_stream(
&self,
path: &str,
headers: &[(String, String)],
request: Vec<u8>,
endpoint: Option<&str>,
) -> Result<tonic::Streaming<Vec<u8>>, GrpcError> {
let path_q = tonic::codegen::http::uri::PathAndQuery::try_from(path.to_string())
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("{path:?} is not a gRPC path: {e}") })?;
let channel = self.channel_for(endpoint)?;
let mut grpc = tonic::client::Grpc::new(channel).max_decoding_message_size(MAX_GRPC_DECODE);
grpc.ready().await.map_err(|e| GrpcError { code: "Unavailable".into(), message: format!("{}: {e}", endpoint.unwrap_or(&self.endpoint)) })?;
let mut req = tonic::Request::new(request);
if let Some(b) = &self.bearer {
req.metadata_mut().insert("authorization", b.clone());
}
for (k, v) in headers {
let key: tonic::metadata::MetadataKey<tonic::metadata::Ascii> = k
.parse()
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("{k:?} is not a metadata name: {e}") })?;
let value: tonic::metadata::MetadataValue<tonic::metadata::Ascii> = v
.parse()
.map_err(|e| GrpcError { code: "InvalidArgument".into(), message: format!("the value for {k:?} is not a header value: {e}") })?;
req.metadata_mut().insert(key, value);
}
let endpoint_name = endpoint.unwrap_or(&self.endpoint).to_string();
grpc.server_streaming(req, path_q, RawCodec)
.await
.map(|r| r.into_inner())
.map_err(|s| GrpcError { code: format!("{:?}", s.code()), message: format!("{endpoint_name}: {}", s.message()) })
}
}
/// Bytes in, bytes out: the guest carries the protos.
#[derive(Default)]
struct RawCodec;
struct RawEncoder;
struct RawDecoder;
impl tonic::codec::Codec for RawCodec {
type Encode = Vec<u8>;
type Decode = Vec<u8>;
type Encoder = RawEncoder;
type Decoder = RawDecoder;
fn encoder(&mut self) -> RawEncoder {
RawEncoder
}
fn decoder(&mut self) -> RawDecoder {
RawDecoder
}
}
impl tonic::codec::Encoder for RawEncoder {
type Item = Vec<u8>;
type Error = tonic::Status;
fn encode(&mut self, item: Vec<u8>, dst: &mut tonic::codec::EncodeBuf<'_>) -> Result<(), tonic::Status> {
dst.put_slice(&item);
Ok(())
}
}
impl tonic::codec::Decoder for RawDecoder {
type Item = Vec<u8>;
type Error = tonic::Status;
fn decode(&mut self, src: &mut tonic::codec::DecodeBuf<'_>) -> Result<Option<Vec<u8>>, tonic::Status> {
let n = src.remaining();
let mut out = vec![0u8; n];
src.copy_to_slice(&mut out);
Ok(Some(out))
}
}
impl HttpDoor {
/// **The door for `hosts`**: every redirect hop is checked against the
/// same list (a hop off it is an error, never followed — before this a
/// 302 from an allowed host reached any host at all), at most
/// [`HTTP_REDIRECTS_MAX`] hops, and a body is read up to `body_max`.
pub fn new(hosts: Vec<String>, call_timeout: Duration, body_max: usize) -> Result<HttpDoor, String> {
let client = http_client(&hosts, call_timeout, None, &[])?;
Ok(HttpDoor { client, hosts, auth: Vec::new(), secrets_dir: None, timeout: call_timeout, cert_clients: Default::default(), body_max, resolve: Vec::new(), label: String::new(), say: say_on_stderr() })
}
/// [`HttpDoor::new`] with `http_resolve` (`<name>=<ip>` entries).
pub fn with_resolve(hosts: Vec<String>, call_timeout: Duration, body_max: usize, resolve: Vec<String>) -> Result<HttpDoor, String> {
let client = http_client(&hosts, call_timeout, None, &resolve)?;
Ok(HttpDoor { client, hosts, auth: Vec::new(), secrets_dir: None, timeout: call_timeout, cert_clients: Default::default(), body_max, resolve, label: String::new(), say: say_on_stderr() })
}
/// Name the door in its log lines, and send them to `say` (a test's collector).
pub fn with_log(mut self, label: &str, say: Arc<dyn Fn(&str) + Send + Sync>) -> HttpDoor {
self.label = label.to_string();
self.say = say;
self
}
/// Is `url` one of the allowed hosts, on the scheme the entry allows?
/// Does `url` fall under this ONE `http_hosts` spelling — scheme, host and
/// port exactly? The rule `allowed` and `http_auth` share, so a credential
/// can never be added to a request the allow-list would not have sent.
pub fn entry_matches(entry: &str, url: &reqwest::Url) -> bool {
let Some(host) = url.host_str() else { return false };
let port = url.port_or_known_default();
let (scheme, rest) = match entry.strip_prefix("http://") {
Some(r) => ("http", r),
None => ("https", entry.strip_prefix("https://").unwrap_or(entry)),
};
let rest = rest.trim_end_matches('/');
let (e_host, e_port) = match rest.rsplit_once(':') {
Some((h, p)) if p.chars().all(|c| c.is_ascii_digit()) && !h.is_empty() => (h, p.parse::<u16>().ok()),
_ => (rest, None),
};
let e_port = e_port.or(if scheme == "http" { Some(80) } else { Some(443) });
url.scheme() == scheme && e_host.eq_ignore_ascii_case(host) && e_port == port
}
pub fn allowed(hosts: &[String], url: &reqwest::Url) -> Result<(), String> {
if url.host_str().is_none() {
return Err(format!("{url}: has no host"));
}
if hosts.iter().any(|e| Self::entry_matches(e, url)) {
return Ok(());
}
Err(format!(
"{url}: {} is not in this guest's `http_hosts` [{}]; the door reaches only what the block names",
url.host_str().unwrap_or("?"),
hosts.join(", ")
))
}
/// The `http_auth` entry for `url`, if the block wrote one.
fn auth_for(&self, url: &reqwest::Url) -> Option<&HttpAuth> {
self.auth.iter().find(|a| Self::entry_matches(&a.host, url))
}
fn secrets_dir(&self) -> Result<&Path, String> {
self.secrets_dir
.as_deref()
.ok_or_else(|| "[secret-dir] this block names secrets and the server has no secrets_dir; set `secrets_dir` (default <data_dir>/secrets)".to_string())
}
/// The header an entry SETS, lower-case: what a guest's own header of the
/// same name is dropped for. `None` when the entry adds no header.
fn auth_header(a: &HttpAuth) -> Option<String> {
if a.basic_user.is_some() || a.basic_user_secret.is_some() {
return Some("authorization".into());
}
a.secret.as_ref().map(|_| a.header.as_deref().unwrap_or("Authorization").to_ascii_lowercase())
}
/// **Add the credential, from the file, on the host.** Every refusal names
/// the secret; none carries a byte of it. Every value added is pushed onto
/// `used`, so the answer can be scrubbed of it before the guest sees it.
fn apply_auth(&self, mut req: reqwest::RequestBuilder, a: &HttpAuth, used: &mut Vec<Vec<u8>>) -> Result<reqwest::RequestBuilder, String> {
if a.basic_user.is_some() || a.basic_user_secret.is_some() {
let dir = self.secrets_dir()?;
let user = match (&a.basic_user, &a.basic_user_secret) {
(Some(u), _) => u.clone(),
(None, Some(n)) => crate::secrets::read(dir, n)?.text(n)?.to_string(),
(None, None) => unreachable!(),
};
let password = match &a.secret {
Some(n) => Some(crate::secrets::read(dir, n)?.text(n)?.to_string()),
None => None,
};
if a.basic_user_secret.is_some() {
used.push(user.as_bytes().to_vec());
}
if let Some(p) = &password {
used.push(p.as_bytes().to_vec());
}
return Ok(req.basic_auth(user, password));
}
if let Some(n) = &a.secret {
let dir = self.secrets_dir()?;
let s = crate::secrets::read(dir, n)?;
used.push(s.bytes().to_vec());
let value = format!("{}{}", a.prefix, s.text(n)?);
let mut hv = reqwest::header::HeaderValue::from_str(&value)
.map_err(|_| format!("[secret-not-header] secret.{n} (with its prefix) is not a valid header value"))?;
hv.set_sensitive(true);
let name = reqwest::header::HeaderName::from_bytes(a.header.as_deref().unwrap_or("Authorization").as_bytes())
.map_err(|_| format!("[http-auth-header] {:?} is not a header name", a.header))?;
req = req.header(name, hv);
}
Ok(req)
}
/// The client for a host that presents a client certificate, rebuilt when
/// the certificate file changes; the plain client otherwise.
fn client_for(&self, a: Option<&HttpAuth>) -> Result<reqwest::Client, String> {
let Some((entry, name)) = a.and_then(|a| a.client_cert_secret.as_deref().map(|n| (a.host.as_str(), n))) else {
return Ok(self.client.clone());
};
let dir = self.secrets_dir()?;
let stamp = crate::secrets::stamp(dir, name).ok_or_else(|| format!("[secret-missing] secret.{name} (the client certificate for {entry}) cannot be read"))?;
let mut cache = self.cert_clients.lock().map_err(|_| "the client-certificate cache is poisoned".to_string())?;
if let Some((s, c)) = cache.get(entry) {
if *s == stamp {
return Ok(c.clone());
}
}
let pem = crate::secrets::read(dir, name)?;
let id = reqwest::Identity::from_pem(pem.bytes())
.map_err(|_| format!("[client-cert] secret.{name} is not a PEM certificate chain with its private key"))?;
let c = http_client(&self.hosts, self.timeout, Some(id), &self.resolve)?;
cache.insert(entry.to_string(), (stamp, c.clone()));
Ok(c)
}
async fn call(&self, head: &HttpHead, body: Vec<u8>) -> Result<(HostHead, Vec<u8>), String> {
let began = std::time::Instant::now();
let out = self.call_unlogged(head, body).await;
let outcome = match &out {
Ok((h, _)) => Ok(h.status.unwrap_or(0)),
Err(e) => Err(e.as_str()),
};
let took = began.elapsed();
let line = door_line(&self.label, &head.method, &head.url, outcome, took, self.timeout);
if quiet_reads().say(&head.method, quiet_key(&line), outcome, took) {
(self.say)(&line);
}
out
}
async fn call_unlogged(&self, head: &HttpHead, body: Vec<u8>) -> Result<(HostHead, Vec<u8>), String> {
let url = reqwest::Url::parse(&head.url).map_err(|e| format!("{:?}: {e}", head.url))?;
Self::allowed(&self.hosts, &url)?;
let method = reqwest::Method::from_bytes(head.method.as_bytes()).map_err(|e| format!("{:?}: {e}", head.method))?;
let auth = self.auth_for(&url);
let client = self.client_for(auth)?;
let mut req = client.request(method, url);
// A guest's header of the name the host sets is DROPPED: the credential
// is the operator's, and a guest must not be able to choose it.
let owned = auth.and_then(Self::auth_header);
for (k, v) in &head.headers {
if owned.as_deref().is_some_and(|o| k.eq_ignore_ascii_case(o)) {
continue;
}
req = req.header(k.as_str(), v.as_str());
}
let mut used: Vec<Vec<u8>> = Vec::new();
if let Some(a) = auth {
req = self.apply_auth(req, a, &mut used)?;
}
let body = match auth {
Some(a) if !a.body_secrets.is_empty() => self.fill_body(body, a, &mut used)?,
_ => body,
};
if !body.is_empty() {
req = req.body(body);
}
let mut resp = req.send().await.map_err(|e| format!("{}: {}", head.url, chain(&e)))?;
let too_big = |n: u64| format!("[http-body-too-large] {}: the response body is {n} bytes or more, over the door's cap of {} bytes — refused, not read", head.url, self.body_max);
if let Some(n) = resp.content_length().filter(|n| *n > self.body_max as u64) {
return Err(too_big(n));
}
let status = resp.status().as_u16();
let headers: Vec<(String, String)> = resp
.headers()
.iter()
.map(|(k, v)| (k.to_string(), String::from_utf8_lossy(v.as_bytes()).into_owned()))
.collect();
let mut bytes = Vec::new();
while let Some(chunk) = resp.chunk().await.map_err(|e| format!("{}: reading the body: {}", head.url, chain(&e)))? {
if bytes.len() + chunk.len() > self.body_max {
return Err(too_big((bytes.len() + chunk.len()) as u64));
}
bytes.extend_from_slice(&chunk);
}
// **Whatever the host added is scrubbed from the answer** before the
// guest sees it: a peer that echoes its request (a debugging endpoint,
// an order that repeats its push URL) must not hand the guest the
// secret it was never given.
let mut headers = headers;
for secret in used.iter().filter(|s| s.len() >= 4) {
scrub(&mut bytes, secret);
for (_, v) in headers.iter_mut() {
if v.as_bytes().windows(secret.len()).any(|w| w == secret.as_slice()) {
*v = "[scrubbed: the host's secret]".into();
}
}
}
Ok((HostHead { status: Some(status), headers, ..HostHead::ok() }, bytes))
}
/// **Put the entry's body secrets where the guest wrote their
/// placeholders**, `{{secret:<name>}}`. A placeholder for a name the entry
/// does not list is left as written — the guest asked for something the
/// block did not give it, and the peer will say so.
fn fill_body(&self, body: Vec<u8>, a: &HttpAuth, used: &mut Vec<Vec<u8>>) -> Result<Vec<u8>, String> {
let mut body = body;
for n in &a.body_secrets {
let ph = monetize_product::body_secret_placeholder(n);
if !body.windows(ph.len()).any(|w| w == ph.as_bytes()) {
continue;
}
let s = crate::secrets::read(self.secrets_dir()?, n)?;
used.push(s.bytes().to_vec());
body = replace_all(&body, ph.as_bytes(), s.bytes());
}
Ok(body)
}
}
impl Doors {
/// **Open a stream, or refuse before it opens.**
///
/// The refusal is the `flush` rule applied to a replay: a stream the host
/// could not carry to its end must not be started, because a stream cut in
/// half is a box that looks restored and is not. `expect_secs` is what the
/// guest believes it needs; `0` asks for whatever the call's deadline
/// allows and takes the consequences at the close.
pub fn open_stream(&self, head: &StreamOpen) -> Result<(HostHead, Vec<u8>), String> {
self.open_stream_for(0, head)
}
/// [`Doors::open_stream`] for the guest call `owner` ([`CallState::call`]):
/// its end sweeps this stream, and no other call's end does.
pub fn open_stream_for(&self, owner: u64, head: &StreamOpen) -> Result<(HostHead, Vec<u8>), String> {
let Some(door) = self.grpc.as_ref() else {
return Err(format!("{}: no gRPC door is open for this guest", imports::GRPC_STREAM_OPEN));
};
if head.expect_secs > 0 && Duration::from_secs(u64::from(head.expect_secs)) >= self.call_timeout {
return Err(format!(
"a stream to {} expected to need {} s cannot be driven through a guest whose calls are cut off after {} s: it would be interrupted part way, and a truncated stream is a box that LOOKS done and is not. Raise `guest_call_timeout_ms` above {} s, or send less in one call",
head.path,
head.expect_secs,
self.call_timeout.as_secs(),
head.expect_secs
));
}
let mut streams = self.streams.lock().map_err(|_| "the guest's stream table is poisoned".to_string())?;
// TWO, and the number is the restore: the primary being filled and the
// twin being read. A third would be a guest holding the host open for
// something nobody has described.
if streams.open.len() >= MAX_OPEN_STREAMS {
return Err(format!(
"this guest already has {} stream(s) open, which is the ceiling: a restore holds the primary it \
fills and the twin it reads, and nothing described needs a third",
streams.open.len()
));
}
let (tx, join) = self
.rt
.block_on(door.open_stream(&head.path, &head.headers, head.endpoint.as_deref()))
.map_err(|e| format!("{}: {}", e.code, e.message))?;
streams.next += 1;
let id = streams.next;
streams.open.insert(
id,
OpenStream {
kind: StreamKind::Sending { tx: Some(tx), join },
owner,
path: head.path.clone(),
deadline: std::time::Instant::now() + self.call_timeout,
chunks: 0,
bytes: 0,
},
);
Ok((HostHead { stream: Some(id), ..HostHead::ok() }, Vec::new()))
}
/// One chunk. **The deadline is checked before every push**, and reaching
/// it cancels the stream rather than letting it dribble past the call —
/// the close then reports the truncation as an error, which is the one
/// thing that must never be silent.
pub fn send_stream(&self, id: u64, chunk: Vec<u8>) -> Result<(HostHead, Vec<u8>), String> {
let mut streams = self.streams.lock().map_err(|_| "the guest's stream table is poisoned".to_string())?;
let s = streams.open.get_mut(&id).ok_or_else(|| format!("stream {id} is not open"))?;
if std::time::Instant::now() >= s.deadline {
let path = s.path.clone();
let (chunks, bytes) = (s.chunks, s.bytes);
drop(streams);
self.cancel_stream(id);
return Err(format!(
"the call's deadline was reached after {chunks} chunk(s), {bytes} byte(s) on {path}: the stream was CANCELLED rather than left half sent, and nothing pushed is reported as delivered. The replay resumes where the box's own head left it"
));
}
let StreamKind::Sending { tx, .. } = &s.kind else {
return Err(format!("stream {id} is a server stream: it is read with host_grpc_stream_recv, never written to"));
};
let tx = tx.clone().ok_or_else(|| format!("stream {id} is already closed"))?;
// The peer's queue, with back-pressure: the runtime drives the call
// while this waits, which is the only time anything should be moving.
match self.rt.block_on(tx.send(chunk.clone())) {
Ok(()) => {
s.chunks += 1;
s.bytes += chunk.len() as u64;
Ok((HostHead::ok(), Vec::new()))
}
// The receiver is gone: the peer ended the call, and the status is
// on the join handle. Report THAT, not "the channel closed".
Err(_) => {
let path = s.path.clone();
drop(streams);
match self.close_stream(id) {
Ok(_) => Err(format!("{path}: the peer ended the stream before it was closed, and answered nothing")),
Err(e) => Err(e),
}
}
}
}
/// Close the stream and read the peer's single answer.
pub fn close_stream(&self, id: u64) -> Result<(HostHead, Vec<u8>), String> {
let mut streams = self.streams.lock().map_err(|_| "the guest's stream table is poisoned".to_string())?;
let s = streams.open.remove(&id).ok_or_else(|| format!("stream {id} is not open"))?;
drop(streams);
let path = s.path.clone();
match s.kind {
// Dropping the sender is what ends the request half; the peer then
// answers.
StreamKind::Sending { tx, join } => {
drop(tx);
match self.rt.block_on(join) {
Ok(Ok(body)) => Ok((HostHead::ok(), body)),
Ok(Err(e)) => Err(format!("{}: {}", e.code, e.message)),
Err(e) => Err(format!("{path}: the streaming call did not finish ({e})")),
}
}
// A server stream has no answer of its own: it ENDED, and the
// guest has already read every message it will get.
StreamKind::Receiving { .. } => Ok((HostHead::ok(), Vec::new())),
}
}
/// **HMAC-SHA256 of `message` under a secret the guest never sees**,
/// compared in constant time against every candidate. The secret must be
/// one the block names in `hmac_secrets`.
pub fn hmac_verify(&self, head: &HmacHead, message: &[u8]) -> Result<(HostHead, Vec<u8>), String> {
use hmac::{KeyInit as _, Mac as _};
if !self.hmac_secrets.contains(&head.secret) {
return Err(format!(
"[hmac-secret-not-named] `{}` is not in this block's hmac_secrets [{}]",
head.secret,
self.hmac_secrets.join(", ")
));
}
if head.algo.eq_ignore_ascii_case("eq") {
// **Is the message the secret itself?** For a vendor that signs
// nothing and authenticates its push with a shared credential
// (Klarna's basic auth): constant time, and the answer is one bit.
let dir = self.secrets_dir.as_deref().ok_or_else(|| "[secret-dir] the server has no secrets_dir".to_string())?;
let secret = crate::secrets::read(dir, &head.secret)?;
let same = ct_eq(secret.bytes(), message);
return Ok((HostHead { value: Some(if same { "match" } else { "mismatch" }.to_string()), ..HostHead::ok() }, Vec::new()));
}
if !head.algo.eq_ignore_ascii_case("sha256") {
return Err(format!("[hmac-algo] {:?} is not a comparison this host makes; the two are sha256 and eq", head.algo));
}
if head.candidates_hex.len() > MAX_HMAC_CANDIDATES {
return Err(format!("[hmac-candidates] {} signatures to compare; the ceiling is {MAX_HMAC_CANDIDATES}", head.candidates_hex.len()));
}
let dir = self.secrets_dir.as_deref().ok_or_else(|| "[secret-dir] the server has no secrets_dir".to_string())?;
let secret = crate::secrets::read(dir, &head.secret)?;
let mut mac = hmac::Hmac::<sha2::Sha256>::new_from_slice(secret.bytes()).map_err(|_| "[hmac-key] the secret is not an HMAC key".to_string())?;
mac.update(message);
let matched = head.candidates_hex.iter().filter_map(|h| hex_decode(h)).any(|sig| mac.clone().verify_slice(&sig).is_ok());
Ok((HostHead { value: Some(if matched { "match" } else { "mismatch" }.to_string()), ..HostHead::ok() }, Vec::new()))
}
/// What this product remembered. `end` means the key was never written,
/// which is a real answer and not an error.
pub fn state_get(&self, key: &str) -> Result<(HostHead, Vec<u8>), String> {
let state = self.state.lock().map_err(|_| "the guest's state is poisoned".to_string())?;
match state.get(key) {
Some(v) => Ok((HostHead::ok(), v.clone())),
None => Ok((HostHead { end: true, ..HostHead::ok() }, Vec::new())),
}
}
/// Remember it, within the ceilings. A guest that exceeds either is
/// refused BY NAME rather than growing the server quietly.
pub fn state_put(&self, key: &str, value: Vec<u8>) -> Result<(HostHead, Vec<u8>), String> {
self.state_put_for(0, key, value)
}
/// [`Doors::state_put`] from call `call`: its own write-through slot, if one is held.
pub fn state_put_for(&self, call: u64, key: &str, value: Vec<u8>) -> Result<(HostHead, Vec<u8>), String> {
if key.trim().is_empty() {
return Err("a state key may not be blank".to_string());
}
if value.len() > self.state_max_bytes {
return Err(format!(
"{key:?} is {} bytes and the ceiling is {}: the host remembers what a guest cannot, \
not whatever it likes",
value.len(),
self.state_max_bytes
));
}
if let Some(w) = self.write_through.lock().map_err(|_| "the write-through slot is poisoned".to_string())?.get(&call) {
if w.key == key {
(w.sink)(&value).map_err(|e| format!("{key:?} was not written through: {e}"))?;
return Ok((HostHead::ok(), Vec::new()));
}
}
let mut state = self.state.lock().map_err(|_| "the guest's state is poisoned".to_string())?;
if !state.contains_key(key) && state.len() >= MAX_STATE_KEYS {
return Err(format!(
"this guest already holds {MAX_STATE_KEYS} keys, which is the ceiling; nothing described needs \
more, and a guest that could keep writing would grow the server"
));
}
state.insert(key.to_string(), value);
Ok((HostHead::ok(), Vec::new()))
}
/// **Remember `key` once**: `true` the first time (now remembered), `false` when it was
/// already there. One lock for the look and the write, under the same ceiling as
/// [`Doors::state_put`]. The host's ticket ledger ([`crate::tickets`]).
pub fn state_admit(&self, key: &str) -> Result<bool, String> {
if key.trim().is_empty() {
return Err("a state key may not be blank".to_string());
}
let mut state = self.state.lock().map_err(|_| "the guest's state is poisoned".to_string())?;
if state.contains_key(key) {
return Ok(false);
}
if state.len() >= MAX_STATE_KEYS {
return Err(format!("this product already holds {MAX_STATE_KEYS} keys in the host's memory, which is the ceiling"));
}
state.insert(key.to_string(), b"1".to_vec());
Ok(true)
}
/// **Open a server-streaming call**, one request out.
pub fn open_server_stream(&self, head: &StreamOpen, request: Vec<u8>) -> Result<(HostHead, Vec<u8>), String> {
self.open_server_stream_for(0, head, request)
}
/// [`Doors::open_server_stream`] for the guest call `owner`.
pub fn open_server_stream_for(&self, owner: u64, head: &StreamOpen, request: Vec<u8>) -> Result<(HostHead, Vec<u8>), String> {
let Some(door) = self.grpc.as_ref() else {
return Err(format!("{}: no gRPC door is open for this guest", imports::GRPC_SERVER_STREAM));
};
let mut streams = self.streams.lock().map_err(|_| "the guest's stream table is poisoned".to_string())?;
if streams.open.len() >= MAX_OPEN_STREAMS {
return Err(format!("this guest already has {} stream(s) open, which is the ceiling", streams.open.len()));
}
let rx = self
.rt
.block_on(door.open_server_stream(&head.path, &head.headers, request, head.endpoint.as_deref()))
.map_err(|e| format!("{}: {}", e.code, e.message))?;
streams.next += 1;
let id = streams.next;
streams.open.insert(
id,
OpenStream {
kind: StreamKind::Receiving { rx: Box::new(rx) },
owner,
path: head.path.clone(),
deadline: std::time::Instant::now() + self.call_timeout,
chunks: 0,
bytes: 0,
},
);
Ok((HostHead { stream: Some(id), ..HostHead::ok() }, Vec::new()))
}
/// **The next message, or the END.** `end` is the only honest way a stream
/// stops: an error is reported as an error, so a reader cannot mistake a
/// broken twin for a log that ran out.
pub fn recv_stream(&self, id: u64) -> Result<(HostHead, Vec<u8>), String> {
let mut streams = self.streams.lock().map_err(|_| "the guest's stream table is poisoned".to_string())?;
let s = streams.open.get_mut(&id).ok_or_else(|| format!("stream {id} is not open"))?;
if std::time::Instant::now() >= s.deadline {
let (path, chunks) = (s.path.clone(), s.chunks);
drop(streams);
self.cancel_stream(id);
return Err(format!(
"the call's deadline was reached after {chunks} message(s) on {path}: the stream was CANCELLED \
rather than half read, and what did not arrive is not reported as the end"
));
}
let StreamKind::Receiving { rx } = &mut s.kind else {
return Err(format!("stream {id} is a client stream: it is written to, never read"));
};
match self.rt.block_on(rx.message()) {
Ok(Some(body)) => {
s.chunks += 1;
s.bytes += body.len() as u64;
Ok((HostHead::ok(), body))
}
Ok(None) => Ok((HostHead { end: true, ..HostHead::ok() }, Vec::new())),
Err(e) => Err(format!("{:?}: {}", e.code(), e.message())),
}
}
/// Abandon a stream. The peer sees the request half end without a close,
/// and nothing pushed is reported as delivered.
pub fn cancel_stream(&self, id: u64) {
let Ok(mut streams) = self.streams.lock() else { return };
if let Some(s) = streams.open.remove(&id) {
if let StreamKind::Sending { tx, join } = s.kind {
drop(tx);
join.abort();
}
}
}
/// **Whatever the guest left open when its call ended.** A guest is a
/// fresh instance per call; a stream that survived one would be a stream
/// nothing could ever close.
/// **What ONE call left open** (round 3): the streams `owner` opened.
pub fn sweep_streams_of(&self, owner: u64) -> usize {
let Ok(mut streams) = self.streams.lock() else { return 0 };
let ids: Vec<u64> = streams.open.iter().filter(|(_, s)| s.owner == owner).map(|(id, _)| *id).collect();
for id in &ids {
if let Some(s) = streams.open.remove(id) {
if let StreamKind::Sending { tx, join } = s.kind {
drop(tx);
join.abort();
}
}
}
ids.len()
}
/// Test seam: a client-streaming entry owned by `owner`, nothing behind it.
#[cfg(test)]
fn test_stream(&self, owner: u64) -> u64 {
let (tx, _rx) = tokio::sync::mpsc::channel::<Vec<u8>>(1);
// A32-14, stays a task: one empty stream for a test of the stream table; its handle is kept in the table.
let join = self.rt.rt().spawn(async { Ok::<Vec<u8>, GrpcError>(Vec::new()) });
let mut streams = self.streams.lock().unwrap();
streams.next += 1;
let id = streams.next;
let deadline = std::time::Instant::now() + self.call_timeout;
streams.open.insert(id, OpenStream { kind: StreamKind::Sending { tx: Some(tx), join }, owner, path: "/test".into(), deadline, chunks: 0, bytes: 0 });
id
}
#[cfg(test)]
fn open_count(&self) -> usize {
self.streams.lock().unwrap().open.len()
}
pub fn sweep_streams(&self) -> usize {
let Ok(mut streams) = self.streams.lock() else { return 0 };
let ids: Vec<u64> = streams.open.keys().copied().collect();
for id in &ids {
if let Some(s) = streams.open.remove(id) {
if let StreamKind::Sending { tx, join } = s.kind {
drop(tx);
join.abort();
}
}
}
ids.len()
}
}
/// Constant-time equality over the bytes. The length is not secret, so a
/// mismatch answers false at once.
fn ct_eq(a: &[u8], b: &[u8]) -> bool {
if a.len() != b.len() {
return false;
}
a.iter().zip(b).fold(0u8, |diff, (x, y)| diff | (x ^ y)) == 0
}
/// Every occurrence of `from` in `hay`, replaced by `to`.
fn replace_all(hay: &[u8], from: &[u8], to: &[u8]) -> Vec<u8> {
let mut out = Vec::with_capacity(hay.len());
let mut i = 0;
while i < hay.len() {
if hay[i..].starts_with(from) {
out.extend_from_slice(to);
i += from.len();
} else {
out.push(hay[i]);
i += 1;
}
}
out
}
/// Overwrite every occurrence of `secret` in `bytes` with `*`, in place.
fn scrub(bytes: &mut [u8], secret: &[u8]) {
let mut i = 0;
while i + secret.len() <= bytes.len() {
if &bytes[i..i + secret.len()] == secret {
bytes[i..i + secret.len()].fill(b'*');
i += secret.len();
} else {
i += 1;
}
}
}
/// Lower- or upper-case hex to bytes; `None` for anything else.
fn hex_decode(s: &str) -> Option<Vec<u8>> {
let s = s.trim();
if s.len() % 2 != 0 || s.len() > 256 {
return None;
}
let nib = |c: u8| match c {
b'0'..=b'9' => Some(c - b'0'),
b'a'..=b'f' => Some(c - b'a' + 10),
b'A'..=b'F' => Some(c - b'A' + 10),
_ => None,
};
s.as_bytes().chunks(2).map(|p| Some(nib(p[0])? << 4 | nib(p[1])?)).collect()
}
/// What the host lends one call.
pub struct CallState {
pub doors: Arc<Doors>,
pub faults: Vec<String>,
pub limits: BoundedLimiter,
/// **Which call this is** (round 3): the streams it opens are its own,
/// swept at ITS end ([`Doors::sweep_streams_of`]) and never at the end of
/// another call running beside it on the same guest.
pub call: u64,
}
/// The calls' numbers, process-wide; `0` is "no call named".
static NEXT_CALL: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1);
impl CallState {
/// One call's state, with a number no other call has.
pub fn new(doors: Arc<Doors>, limits: BoundedLimiter) -> CallState {
CallState { doors, faults: Vec::new(), limits, call: NEXT_CALL.fetch_add(1, std::sync::atomic::Ordering::Relaxed) }
}
}
impl GuestState for CallState {
fn limiter(&mut self) -> &mut BoundedLimiter {
&mut self.limits
}
fn faults(&mut self) -> &mut Vec<String> {
&mut self.faults
}
}
fn answer(caller: &mut Caller<'_, CallState>, r: Result<(HostHead, Vec<u8>), String>) -> u64 {
let f = match r {
Ok((head, body)) => frame::encode_json(&head, &body),
Err(e) => frame::encode_json(&HostHead::err(e), &[]),
};
write_to_wasm(caller, &f)
}
/// **The clock a guest reads** through its `host_clock` door: the wall clock, in Unix ms (0 if
/// the system clock is before 1970). The host's own check of a guest product's actor ticket
/// (monetize-server `site_ticket`) reads this same clock, so a ticket's life is judged as the
/// guest judged it.
pub fn host_clock_unix_ms() -> u64 {
std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or_default()
}
/// **Define exactly the granted doors and nothing else.**
///
/// **The doors BLOCK their caller's thread** (code-smell review 2026-10-10,
/// row 20): a host function is synchronous, so `host_sleep` is a
/// `thread::sleep` and the HTTP, gRPC and stream doors `block_on` the doors'
/// own runtime until the far side answers — a cloud's lay, minutes. A guest
/// call is therefore made ONLY from a thread that may block: a plain thread or
/// `tokio::task::spawn_blocking` (the server's ticks, the estate loop, the
/// callback). Never from an async task on a runtime's worker: tokio refuses a
/// `block_on` there with a panic, and even where it did not, a worker held for
/// minutes stalls every other task on it.
pub fn register(linker: &mut Linker<CallState>, granted: &[&str]) -> Result<(), String> {
let wrap = |e: wasmtime::Error| e.to_string();
for name in granted {
match *name {
imports::CLOCK => {
linker
.func_wrap(imports::MODULE, imports::CLOCK, |_caller: Caller<'_, CallState>| -> u64 { host_clock_unix_ms() })
.map_err(wrap)?;
}
imports::LOG => {
linker
.func_wrap(imports::MODULE, imports::LOG, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| {
if let Some(line) = read_arg(&mut caller, ptr, len, imports::LOG) {
eprintln!("[guest {}] {}", caller.data().doors.product, String::from_utf8_lossy(&line));
}
})
.map_err(wrap)?;
}
imports::SLEEP => {
// Bounded by the call's own deadline: a guest cannot wait on the
// host longer than it would be allowed to run anyway.
linker
.func_wrap(imports::MODULE, imports::SLEEP, |caller: Caller<'_, CallState>, ms: u64| {
let cap = caller.data().doors.call_timeout;
std::thread::sleep(Duration::from_millis(ms).min(cap));
})
.map_err(wrap)?;
}
imports::GRPC_UNARY => {
linker
.func_wrap(imports::MODULE, imports::GRPC_UNARY, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::GRPC_UNARY) else { return 0 };
let (head, body): (GrpcHead, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::GRPC_UNARY)),
};
let doors = Arc::clone(&caller.data().doors);
let Some(door) = doors.grpc.as_ref() else {
return fault(&mut caller, format!("{}: no gRPC door is open for this guest", imports::GRPC_UNARY));
};
let body = body.to_vec();
let r = doors.rt.block_on(door.unary(&head.path, body, &head.headers, head.endpoint.as_deref()));
let r = match r {
Ok(bytes) => Ok((HostHead::ok(), bytes)),
// A status is an ANSWER, not a fault: the guest turns it into
// the product's own Refused/Unavailable.
Err(g) => Ok((HostHead { ok: false, error: Some(g.message.clone()), grpc: Some(g), ..Default::default() }, Vec::new())),
};
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::HTTP => {
linker
.func_wrap(imports::MODULE, imports::HTTP, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::HTTP) else { return 0 };
let (head, body): (HttpHead, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::HTTP)),
};
let doors = Arc::clone(&caller.data().doors);
let Some(door) = doors.http.as_ref() else {
return fault(&mut caller, format!("{}: no HTTP door is open for this guest", imports::HTTP));
};
let body = body.to_vec();
let r = doors.rt.block_on(door.call(&head, body));
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::GRPC_STREAM_OPEN => {
linker
.func_wrap(imports::MODULE, imports::GRPC_STREAM_OPEN, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::GRPC_STREAM_OPEN) else { return 0 };
let (head, _): (StreamOpen, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::GRPC_STREAM_OPEN)),
};
let (doors, call) = (Arc::clone(&caller.data().doors), caller.data().call);
let r = doors.open_stream_for(call, &head);
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::GRPC_STREAM_SEND => {
linker
.func_wrap(imports::MODULE, imports::GRPC_STREAM_SEND, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::GRPC_STREAM_SEND) else { return 0 };
let (head, chunk): (StreamRef, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::GRPC_STREAM_SEND)),
};
let doors = Arc::clone(&caller.data().doors);
let r = doors.send_stream(head.stream, chunk.to_vec());
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::GRPC_STREAM_CLOSE => {
linker
.func_wrap(imports::MODULE, imports::GRPC_STREAM_CLOSE, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::GRPC_STREAM_CLOSE) else { return 0 };
let (head, _): (StreamRef, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::GRPC_STREAM_CLOSE)),
};
let doors = Arc::clone(&caller.data().doors);
let r = doors.close_stream(head.stream);
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::GRPC_STREAM_CANCEL => {
linker
.func_wrap(imports::MODULE, imports::GRPC_STREAM_CANCEL, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::GRPC_STREAM_CANCEL) else { return 0 };
let (head, _): (StreamRef, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::GRPC_STREAM_CANCEL)),
};
let doors = Arc::clone(&caller.data().doors);
doors.cancel_stream(head.stream);
answer(&mut caller, Ok((HostHead::ok(), Vec::new())))
})
.map_err(wrap)?;
}
imports::GRPC_SERVER_STREAM => {
linker
.func_wrap(imports::MODULE, imports::GRPC_SERVER_STREAM, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::GRPC_SERVER_STREAM) else { return 0 };
let (head, request): (StreamOpen, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::GRPC_SERVER_STREAM)),
};
let (doors, call) = (Arc::clone(&caller.data().doors), caller.data().call);
let r = doors.open_server_stream_for(call, &head, request.to_vec());
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::GRPC_STREAM_RECV => {
linker
.func_wrap(imports::MODULE, imports::GRPC_STREAM_RECV, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::GRPC_STREAM_RECV) else { return 0 };
let (head, _): (StreamRef, &[u8]) = match frame::decode_json(&raw) {
Ok(x) => x,
Err(e) => return fault(&mut caller, format!("{}: {e}", imports::GRPC_STREAM_RECV)),
};
let doors = Arc::clone(&caller.data().doors);
let r = doors.recv_stream(head.stream);
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::STATE_GET => {
linker
.func_wrap(imports::MODULE, imports::STATE_GET, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::STATE_GET) else { return 0 };
let r = match frame::decode_json::<StateKey>(&raw) {
Ok((head, _)) => caller.data().doors.state_get(&head.key),
Err(e) => Err(format!("{}: {e}", imports::STATE_GET)),
};
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::STATE_PUT => {
linker
.func_wrap(imports::MODULE, imports::STATE_PUT, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::STATE_PUT) else { return 0 };
let r = match frame::decode_json::<StateKey>(&raw) {
Ok((head, value)) => caller.data().doors.state_put_for(caller.data().call, &head.key, value.to_vec()),
Err(e) => Err(format!("{}: {e}", imports::STATE_PUT)),
};
answer(&mut caller, r)
})
.map_err(wrap)?;
}
imports::HMAC_VERIFY => {
linker
.func_wrap(imports::MODULE, imports::HMAC_VERIFY, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::HMAC_VERIFY) else { return 0 };
let r = match frame::decode_json::<HmacHead>(&raw) {
Ok((head, message)) => caller.data().doors.hmac_verify(&head, message),
Err(e) => Err(format!("{}: {e}", imports::HMAC_VERIFY)),
};
answer(&mut caller, r)
})
.map_err(wrap)?;
}
// The typed door (owner decision 6): no name in, the block's address out. It cannot
// be asked for anything else, so it has no allow-list to get wrong.
imports::ENDPOINT => {
linker
.func_wrap(imports::MODULE, imports::ENDPOINT, |mut caller: Caller<'_, CallState>, ptr: u32, len: u32| -> u64 {
let Some(raw) = read_arg(&mut caller, ptr, len, imports::ENDPOINT) else { return 0 };
let r = match frame::decode(&raw) {
Ok(_) => Ok((HostHead { value: Some(caller.data().doors.endpoint.clone()), ..HostHead::ok() }, Vec::new())),
Err(e) => Err(format!("{}: {e}", imports::ENDPOINT)),
};
answer(&mut caller, r)
})
.map_err(wrap)?;
}
other => return Err(format!("`{other}` is not a host door; the vocabulary is {}", imports::ALL.join(", "))),
}
}
Ok(())
}
/// reqwest 0.13 with reqwest 0.12 `rustls-tls` behaviour (2026-10-02): the ring
/// provider and the bundled Mozilla roots (webpki), not 0.13's aws-lc-rs + OS
/// trust store. Roots added later on the builder are merged in, as in 0.12.
fn tls_roots(b: reqwest::ClientBuilder) -> reqwest::ClientBuilder {
let _ = rustls::crypto::ring::default_provider().install_default();
b.tls_certs_only(
webpki_root_certs::TLS_SERVER_ROOT_CERTS
.iter()
.filter_map(|c| reqwest::Certificate::from_der(c).ok()),
)
}
impl std::fmt::Debug for Doors {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Doors")
.field("product", &self.product)
.field("grpc", &self.grpc.as_ref().map(|g| g.endpoint.as_str()))
.field("http_hosts", &self.http.as_ref().map(|h| h.hosts.clone()))
.field("granted_imports", &self.granted_imports)
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_guests_clock_is_the_wall_clock() {
let wall = || std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_millis() as u64;
let (before, clock, after) = (wall(), host_clock_unix_ms(), wall());
assert!(before <= clock && clock <= after, "{before} <= {clock} <= {after}");
}
/// **A busy gunnar is not a dead one** (R5.1-G, 2026-10-07). gunnar's
/// control plane answered a PING from the same workers that ran its slow
/// synchronous calls, so a burst of them left a PING unanswered for
/// seconds; with five seconds the door killed the connection and the
/// `Entitlement.Set` in flight on it came back as an `Internal` stream
/// reset — a paid fact filed refused. gunnar now answers PINGs from a
/// runtime no call can hold, and the door gives a PING 20–30 s anyway, for
/// the gunnars that are not upgraded yet and for a loaded box. A gunnar
/// that is really gone is still noticed: its RST at once, silence after
/// `KEEPALIVE_EVERY + KEEPALIVE_TIMEOUT`.
#[test]
fn a_busy_gunnar_is_given_time_to_answer_a_ping() {
assert!(
KEEPALIVE_TIMEOUT >= Duration::from_secs(20) && KEEPALIVE_TIMEOUT <= Duration::from_secs(30),
"the door gives a PING {KEEPALIVE_TIMEOUT:?}: under 20 s a busy gunnar's connection is killed with \
every call on it, over 30 s a gunnar that is gone in silence is noticed too late"
);
}
/// ★ **A call's end sweeps only the streams THAT call opened** (round 3,
/// 2026-10-06, MEASURED estate-box dev on a loaded dwarf: a recharged
/// main's restore — two streams, the twin read and the primary filled —
/// broke six times with "stream N is not open", 384 s instead of ~35 s:
/// the product guest's calls are not serialized, and the dead-main watch's
/// `restore_status`, every 5 s, ended and swept EVERY open stream, the
/// restore's included). A call that opened nothing sweeps nothing.
#[test]
fn a_calls_end_sweeps_only_the_streams_that_call_opened() {
let doors = http_doors(&[], Duration::from_secs(5));
let (restore, twin) = (doors.test_stream(7), doors.test_stream(7));
let other = doors.test_stream(8);
assert!(restore != twin && twin != other);
assert_eq!(doors.sweep_streams_of(9), 0, "a call that opened nothing swept another call's streams");
assert_eq!(doors.open_count(), 3);
assert_eq!(doors.sweep_streams_of(7), 2, "the restore's own two");
assert_eq!(doors.open_count(), 1, "call 8's stream is still open");
assert_eq!(doors.sweep_streams(), 1, "the whole sweep (a serialized host) still takes everything");
}
/// **A key written THROUGH is never remembered**: while a sink is installed, a
/// put of its key reaches the sink and not the state map; a sink that fails
/// fails the put, naming the key; another key is remembered as ever; with the
/// slot empty the same key is an ordinary state write again.
#[test]
fn a_write_through_key_reaches_the_sink_and_never_the_state() {
let doors = Doors::open("cloud", "", None, None, None, &["state".to_string()], &[], &[], Duration::from_secs(5)).unwrap();
let got = std::sync::Arc::new(std::sync::Mutex::new(Vec::<Vec<u8>>::new()));
let g2 = std::sync::Arc::clone(&got);
doors.write_through.lock().unwrap().insert(0, WriteThrough {
key: "grow-journal".into(),
sink: Box::new(move |b: &[u8]| if b == b"refuse" { Err("disk full".into()) } else { g2.lock().unwrap().push(b.to_vec()); Ok(()) }),
});
doors.state_put("grow-journal", b"one".to_vec()).unwrap();
doors.state_put("grow-journal", b"two".to_vec()).unwrap();
let e = doors.state_put("grow-journal", b"refuse".to_vec()).unwrap_err();
assert!(e.contains("grow-journal") && e.contains("disk full"), "{e}");
doors.state_put("pool", b"ledger".to_vec()).unwrap();
assert_eq!(*got.lock().unwrap(), vec![b"one".to_vec(), b"two".to_vec()]);
assert!(doors.state_get("grow-journal").unwrap().0.end, "a written-through key is not remembered");
assert_eq!(doors.state_get("pool").unwrap().1, b"ledger");
doors.state_put_for(7, "grow-journal", b"another call's".to_vec()).unwrap();
assert_eq!(doors.state_get("grow-journal").unwrap().1, b"another call's", "another call's put never reaches this call's sink");
assert_eq!(got.lock().unwrap().len(), 2);
doors.write_through.lock().unwrap().remove(&0);
doors.state_put("grow-journal", b"plain".to_vec()).unwrap();
assert_eq!(doors.state_get("grow-journal").unwrap().1, b"plain");
}
#[test]
fn the_http_allow_list_matches_host_scheme_and_port_exactly() {
let hosts = vec!["api.example.se".to_string(), "http://lab:8080".to_string()];
let ok = |u: &str| HttpDoor::allowed(&hosts, &reqwest::Url::parse(u).unwrap());
ok("https://api.example.se/v1/x").unwrap();
ok("https://API.example.se:443/").unwrap();
ok("http://lab:8080/api").unwrap();
assert!(ok("http://api.example.se/").is_err(), "plain http to an https entry");
assert!(ok("https://lab:8080/").is_err(), "https to an http entry");
assert!(ok("http://lab/").is_err(), "wrong port");
assert!(ok("https://evil.example.se/").unwrap_err().contains("http_hosts"));
}
/// A loopback HTTP/1.1 server: each connection gets the next canned
/// answer, built from the port it listens on. Answers the port.
fn canned(answers: impl FnOnce(u16) -> Vec<Vec<u8>> + Send + 'static) -> u16 {
use std::io::{Read as _, Write as _};
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
for a in answers(port) {
let Ok((mut s, _)) = l.accept() else { return };
let mut buf = [0u8; 4096];
let _ = s.read(&mut buf);
let _ = s.write_all(&a);
}
});
port
}
fn get(door: &HttpDoor, url: &str) -> Result<(HostHead, Vec<u8>), String> {
let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
rt.block_on(door.call(&HttpHead { method: "GET".into(), url: url.into(), headers: vec![] }, vec![]))
}
fn redirect(to: String) -> Vec<u8> {
format!("HTTP/1.1 302 Found\r\nLocation: {to}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").into_bytes()
}
/// **A redirect is checked hop by hop**: to a host on the list it is
/// followed; off the list it is an error and the other host is never
/// dialled.
#[test]
fn a_redirect_off_the_allow_list_is_refused_and_one_on_it_is_followed() {
// `localhost` is not the `127.0.0.1` on the list.
let port = canned(|p| vec![redirect(format!("http://localhost:{p}/x"))]);
let door = HttpDoor::new(vec![format!("http://127.0.0.1:{port}")], Duration::from_secs(5), HTTP_BODY_MAX).unwrap();
let err = get(&door, &format!("http://127.0.0.1:{port}/start")).unwrap_err();
assert!(err.contains("leaves the allow-list") && err.contains("http_hosts"), "{err}");
let port = canned(|p| vec![redirect(format!("http://127.0.0.1:{p}/final")), b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_vec()]);
let door = HttpDoor::new(vec![format!("http://127.0.0.1:{port}")], Duration::from_secs(5), HTTP_BODY_MAX).unwrap();
let (h, body) = get(&door, &format!("http://127.0.0.1:{port}/start")).unwrap();
assert_eq!((h.status, body.as_slice()), (Some(200), &b"ok"[..]), "a hop on the list is followed");
}
/// **`http_resolve`**: a URL naming a host NAME is dialled at the address the
/// block gives it (the port stays the URL's), the allow-list still judging the
/// URL as written; a malformed entry is refused by name. The bench's vm1 uses it
/// to reach its mock cloud, which the cloud guest must name as loopback.
#[test]
fn a_resolved_name_is_dialled_at_its_address_and_a_bad_entry_is_refused() {
let port = canned(|_| vec![b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_vec()]);
let host = "mock-cloud.invalid";
let door = HttpDoor::with_resolve(vec![format!("http://{host}:{port}")], Duration::from_secs(5), HTTP_BODY_MAX, vec![format!("{host}=127.0.0.1")]).unwrap();
let (h, body) = get(&door, &format!("http://{host}:{port}/1.3/price")).unwrap();
assert_eq!((h.status, body.as_slice()), (Some(200), &b"ok"[..]));
assert!(get(&door, &format!("http://other.invalid:{port}/")).unwrap_err().contains("http_hosts"), "the allow-list still judges the URL");
assert!(resolve_pairs(&["127.0.0.1=10.0.2.100".into()]).unwrap_err().contains("host NAME"));
assert!(resolve_pairs(&["localhost=nowhere".into()]).unwrap_err().contains("IP address"));
assert!(resolve_pairs(&["localhost".into()]).unwrap_err().contains("<name>=<ip>"));
assert_eq!(resolve_pairs(&["LocalHost = 10.0.2.100".into()]).unwrap()[0].0, "localhost");
}
/// **A body over the cap is refused**, whether the server announces its
/// length or streams it without one.
#[test]
fn a_response_body_over_the_cap_is_refused() {
let big = vec![b'x'; 4096];
let mut announced = b"HTTP/1.1 200 OK\r\nContent-Length: 4096\r\nConnection: close\r\n\r\n".to_vec();
announced.extend_from_slice(&big);
let mut unannounced = b"HTTP/1.1 200 OK\r\nConnection: close\r\n\r\n".to_vec();
unannounced.extend_from_slice(&big);
let small = b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\nConnection: close\r\n\r\nabc".to_vec();
let port = canned(move |_| vec![announced, unannounced, small]);
let door = HttpDoor::new(vec![format!("http://127.0.0.1:{port}")], Duration::from_secs(5), 1024).unwrap();
let url = format!("http://127.0.0.1:{port}/");
assert!(get(&door, &url).unwrap_err().contains("over the door's cap"), "announced length");
assert!(get(&door, &url).unwrap_err().contains("over the door's cap"), "streamed");
assert_eq!(get(&door, &url).unwrap().1, b"abc", "under the cap is read whole");
}
#[test]
fn a_body_placeholder_is_filled_and_an_echoed_secret_is_scrubbed() {
let ph = monetize_product::body_secret_placeholder("klarna_callback");
assert_eq!(ph, "{{secret:klarna_callback}}");
let body = format!("{{\"push\":\"https://{ph}@pay.example/cb\"}}");
let filled = replace_all(body.as_bytes(), ph.as_bytes(), b"u1:p2");
assert_eq!(filled, br#"{"push":"https://u1:p2@pay.example/cb"}"#);
let mut echo = b"x u1:p2 y u1:p2".to_vec();
scrub(&mut echo, b"u1:p2");
assert_eq!(echo, b"x ***** y *****");
assert!(ct_eq(b"abc", b"abc") && !ct_eq(b"abc", b"abd") && !ct_eq(b"abc", b"ab"));
assert_eq!(hex_decode("0aFf"), Some(vec![0x0a, 0xff]));
assert_eq!(hex_decode("0g"), None);
}
/// Lengths that differ by a multiple of 256 are still different lengths:
/// a zero-padded tail must not make a short secret equal a long message.
#[test]
fn lengths_that_differ_by_a_multiple_of_256_never_compare_equal() {
assert!(!ct_eq(&[0u8; 256], &[]));
assert!(!ct_eq(&[], &[0u8; 256]));
assert!(!ct_eq(b"abc", &[b"abc".as_slice(), &[0u8; 256]].concat()));
assert!(!ct_eq(&[0u8; 512], &[0u8; 0]));
}
#[test]
fn an_unknown_grant_is_refused_by_name() {
let err = Doors::open("p", "127.0.0.1:1", None, None, None, &["fs".to_string()], &[], &[], Duration::from_secs(1)).unwrap_err();
assert!(err.contains("`fs`") && err.contains("grpc"), "{err}");
}
#[test]
fn a_ca_without_tls_is_refused_and_a_tls_endpoint_opens() {
// A CA named for a PLAINTEXT endpoint: refused by name, because a block
// that wrote it believes its traffic is verified.
let err = Doors::open(
"p",
"127.0.0.1:1",
None,
None,
Some(Path::new("/etc/monetize/ca.pem")),
&["grpc".to_string()],
&[],
&[],
Duration::from_secs(1),
)
.unwrap_err();
assert!(err.contains("PLAINTEXT") && err.contains("ca.pem"), "{err}");
// ...and a `https://` endpoint now OPENS, against the compiled-in
// roots. It is lazy, so nothing is dialled here.
Doors::open("p", "https://x:1", None, None, None, &["grpc".to_string()], &[], &[], Duration::from_secs(1))
.expect("a TLS endpoint is dialled, not refused");
}
/// Owner decision 6 (2026-10-26): **the doors keep the block's address for the typed door,
/// and nothing for a generic one**: a block with no grants opens only the grant-free doors.
#[test]
fn the_doors_keep_the_address_and_open_no_generic_door() {
let d = Doors::open("p", "127.0.0.1:1", Some("tok"), Some("10.0.0.2:50051"), None, &[], &[], &[], Duration::from_secs(1)).unwrap();
assert_eq!(d.granted_imports, [imports::CLOCK, imports::LOG, imports::SLEEP]);
assert_eq!(d.endpoint, "127.0.0.1:1");
let d = Doors::open("p", "127.0.0.1:1", None, None, None, &[grants::ENDPOINT.to_string()], &[], &[], Duration::from_secs(1)).unwrap();
assert_eq!(d.granted_imports, [imports::CLOCK, imports::LOG, imports::SLEEP, imports::ENDPOINT]);
let err = Doors::open("p", "127.0.0.1:1", None, None, None, &["env".to_string()], &[], &[], Duration::from_secs(1)).unwrap_err();
assert!(err.contains("`env`") && err.contains("not a host door"), "{err}");
}
// ── what a call through the HTTP door says, and how it connects ──────
//
// test2 2026-09-30: `POST /auth/signup/start` took 25-33 s and once never
// answered, and front's log had NOTHING between the capacity line and the
// minted checkout — no line per call to the cloud or to Stripe, so the
// time could not be put anywhere.
type Lines = Arc<std::sync::Mutex<Vec<String>>>;
type Say = Arc<dyn Fn(&str) + Send + Sync>;
fn collector() -> (Lines, Say) {
let lines = Arc::new(std::sync::Mutex::new(Vec::<String>::new()));
let l2 = Arc::clone(&lines);
(lines, Arc::new(move |s: &str| l2.lock().unwrap().push(s.to_string())))
}
/// **One line per call**: the guest's name, the method, the endpoint WITHOUT
/// its query (a query is where a careless guest would put a token), the
/// status and the time.
#[test]
fn every_call_through_the_http_door_says_one_line_with_endpoint_status_and_time() {
let port = canned(|_| vec![b"HTTP/1.1 201 Created\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_vec()]);
let (lines, say) = collector();
let door = HttpDoor::new(vec![format!("http://127.0.0.1:{port}")], Duration::from_secs(5), HTTP_BODY_MAX).unwrap().with_log("vendors.stripe.gunnar", say);
let (head, _) = get(&door, &format!("http://127.0.0.1:{port}/v1/checkout/sessions?key=sk_live_notreally")).unwrap();
assert_eq!(head.status, Some(201));
let lines = lines.lock().unwrap().clone();
assert_eq!(lines.len(), 1, "{lines:?}");
let l = &lines[0];
assert!(l.contains("[vendors.stripe.gunnar]") && l.contains(&format!("GET http://127.0.0.1:{port}/v1/checkout/sessions ")), "{l}");
assert!(l.contains("-> 201 in ") && l.contains(" ms"), "{l}");
assert!(!l.contains("sk_live") && !l.contains("key="), "the query reached the log: {l}");
}
/// ★ **PROD-FIX 3 (prod 2026-10-26): a poll's reads are not said on every poll** — the
/// box's 2000-line log ring held about 3 minutes, filled by `GET /1.3/storage/<uuid> -> 200`
/// several times a second, and the trail of the owner's real purchase was pushed out. The
/// same quiet read (a fast GET answered 2xx) is said once per endpoint and status in
/// [`QUIET_READ_EVERY`]; a write, a failure, another status or a slow read is said every time.
#[test]
fn the_same_quiet_read_is_said_once_but_every_write_and_change_is_said() {
let port = canned(|_| (0..14).map(|i| if i == 12 { b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n".to_vec() } else { b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok".to_vec() }).collect());
let (lines, say) = collector();
let door = HttpDoor::new(vec![format!("http://127.0.0.1:{port}")], Duration::from_secs(5), HTTP_BODY_MAX).unwrap().with_log("clouds.upcloud", say);
let storage = format!("http://127.0.0.1:{port}/1.3/storage/0121576b-f6e9-458d-b161-50021dbe3bad");
for _ in 0..10 {
get(&door, &storage).unwrap();
}
let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
for _ in 0..2 {
rt.block_on(door.call(&HttpHead { method: "POST".into(), url: format!("http://127.0.0.1:{port}/1.3/server"), headers: vec![] }, vec![])).unwrap();
}
get(&door, &storage).unwrap(); // 404: another answer
get(&door, &format!("http://127.0.0.1:{port}/1.3/storage/other")).unwrap();
let lines = lines.lock().unwrap().clone();
let n = |pat: &str| lines.iter().filter(|l| l.contains(pat)).count();
assert_eq!(n("GET") , 3, "ten polls of one storage are said once, its 404 and another storage once each: {lines:#?}");
assert_eq!(n("POST"), 2, "every write is said: {lines:#?}");
}
/// **A call that meets the door's deadline is said at WARN** — the front
/// raises a captured line to WARN when it says FAILED — with how long it
/// waited and that it was the deadline, not the peer, that ended it.
#[test]
fn a_call_that_meets_the_deadline_is_said_as_failed_with_the_time() {
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
std::thread::spawn(move || {
// Accept and never answer: a peer that swallowed the request.
let held: Vec<_> = l.incoming().take(1).collect();
std::thread::sleep(Duration::from_secs(5));
drop(held);
});
let (lines, say) = collector();
let door = HttpDoor::new(vec![format!("http://127.0.0.1:{port}")], Duration::from_millis(400), HTTP_BODY_MAX).unwrap().with_log("clouds.upcloud", say);
let began = std::time::Instant::now();
let err = get(&door, &format!("http://127.0.0.1:{port}/1.3/price")).unwrap_err();
assert!(began.elapsed() < Duration::from_secs(3), "the deadline did not hold: {:?}", began.elapsed());
let lines = lines.lock().unwrap().clone();
assert_eq!(lines.len(), 1, "{lines:?} ({err})");
let l = &lines[0];
assert!(l.contains("[clouds.upcloud]") && l.contains("/1.3/price") && l.contains("FAILED after "), "{l}");
assert!(l.contains("deadline"), "the line does not say the deadline ended it: {l}");
}
/// A keep-alive peer that counts the connections it accepted; each one is
/// closed after `idle` without a request (a peer's keep-alive timeout).
fn keep_alive(idle: Duration) -> (u16, Arc<std::sync::atomic::AtomicUsize>) {
use std::io::{Read as _, Write as _};
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = l.local_addr().unwrap().port();
let accepted = Arc::new(std::sync::atomic::AtomicUsize::new(0));
let a2 = Arc::clone(&accepted);
std::thread::spawn(move || {
for s in l.incoming() {
let Ok(mut s) = s else { return };
a2.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
std::thread::spawn(move || {
s.set_read_timeout(Some(idle)).unwrap();
let mut buf = Vec::new();
let mut chunk = [0u8; 4096];
loop {
match s.read(&mut chunk) {
Ok(0) | Err(_) => return, // the peer's idle timeout: close
Ok(n) => buf.extend_from_slice(&chunk[..n]),
}
while let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") {
buf.drain(..end + 4);
if s.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok").is_err() {
return;
}
}
}
});
}
});
(port, accepted)
}
/// The doors a vendor guest gets: HTTP only, to `hosts`.
fn http_doors(hosts: &[String], call_timeout: Duration) -> Doors {
Doors::open_with(DoorSpec {
product: "stripe",
endpoint: "",
bearer: None,
twin: None,
tls_ca: None,
tls_client: None,
grants: &["http".to_string()],
http_hosts: hosts,
grpc_hosts: &[],
call_timeout,
http_auth: &[],
hmac_secrets: &[],
secrets_dir: None,
http_resolve: &[],
})
.unwrap()
}
/// **The door keeps its connection** across a guest's calls: one client per
/// door, driven by the door's own runtime, so a vendor's second request
/// does not pay a second TCP (and on https a second TLS) handshake.
#[test]
fn the_door_reuses_one_connection_across_calls() {
let (port, accepted) = keep_alive(Duration::from_secs(10));
let doors = Doors::open_with(DoorSpec {
product: "stripe",
endpoint: "",
bearer: None,
twin: None,
tls_ca: None,
tls_client: None,
grants: &["http".to_string()],
http_hosts: &[format!("http://127.0.0.1:{port}")],
grpc_hosts: &[],
call_timeout: Duration::from_secs(5),
http_auth: &[],
hmac_secrets: &[],
secrets_dir: None,
http_resolve: &[],
})
.unwrap();
let http = doors.http.as_ref().unwrap();
for _ in 0..3 {
let (h, _) = doors.rt.block_on(http.call(&HttpHead { method: "GET".into(), url: format!("http://127.0.0.1:{port}/v1/x"), headers: vec![] }, vec![])).unwrap();
assert_eq!(h.status, Some(200));
}
assert_eq!(accepted.load(std::sync::atomic::Ordering::SeqCst), 1, "three calls, one connection");
}
/// **A pooled connection the PEER closed while the door was idle is not
/// handed to the next call.** The door's runtime is a current-thread one
/// that only runs inside a call, so nothing reads the peer's FIN between
/// calls; a request written into that socket is lost ("connection closed
/// before message completed") — a signup that fails after the shop sat
/// idle. The door's pool drops a connection before a peer's keep-alive
/// would.
#[test]
fn a_connection_the_peer_closed_while_idle_is_not_reused() {
let (port, accepted) = keep_alive(Duration::from_millis(300));
let doors = http_doors(&[format!("http://127.0.0.1:{port}")], Duration::from_secs(5));
let door = doors.http.as_ref().unwrap();
let call = || doors.rt.block_on(door.call(&HttpHead { method: "POST".into(), url: format!("http://127.0.0.1:{port}/v1/checkout/sessions"), headers: vec![] }, b"a=1".to_vec()));
call().unwrap();
std::thread::sleep(Duration::from_millis(900));
let second = call();
assert!(second.is_ok(), "the second call, after the peer closed the idle connection, failed: {second:?}");
assert!(accepted.load(std::sync::atomic::Ordering::SeqCst) >= 1);
}
/// **Two threads on one guest's runtime.** The door's runtime is ONE
/// current-thread runtime shared by every call to the guest — the
/// signup's thread and the poller's (`paid_since`) alike. While one thread
/// holds it inside a slow call, a second thread's call must still be
/// driven (its IO, its timer) and answer in its own time — neither wait
/// for the first, nor lose its deadline.
#[test]
fn a_slow_call_on_one_thread_does_not_hold_a_fast_call_on_another() {
use std::io::{Read as _, Write as _};
let slow = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let slow_port = slow.local_addr().unwrap().port();
std::thread::spawn(move || {
for s in slow.incoming() {
let Ok(mut s) = s else { return };
std::thread::spawn(move || {
let mut b = [0u8; 4096];
let _ = s.read(&mut b);
std::thread::sleep(Duration::from_secs(4));
let _ = s.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\nConnection: close\r\n\r\nslow");
});
}
});
let (fast_port, _) = keep_alive(Duration::from_secs(10));
let doors = Arc::new(
Doors::open_with(DoorSpec {
product: "stripe",
endpoint: "",
bearer: None,
twin: None,
tls_ca: None,
tls_client: None,
grants: &["http".to_string()],
http_hosts: &[format!("http://127.0.0.1:{slow_port}"), format!("http://127.0.0.1:{fast_port}")],
grpc_hosts: &[],
call_timeout: Duration::from_secs(2),
http_auth: &[],
hmac_secrets: &[],
secrets_dir: None,
http_resolve: &[],
})
.unwrap(),
);
let d1 = Arc::clone(&doors);
let poller = std::thread::spawn(move || {
let began = std::time::Instant::now();
let r = d1.rt.block_on(d1.http.as_ref().unwrap().call(&HttpHead { method: "GET".into(), url: format!("http://127.0.0.1:{slow_port}/v1/checkout/sessions"), headers: vec![] }, vec![]));
(began.elapsed(), r.is_ok())
});
std::thread::sleep(Duration::from_millis(200));
let began = std::time::Instant::now();
let r = doors.rt.block_on(doors.http.as_ref().unwrap().call(&HttpHead { method: "POST".into(), url: format!("http://127.0.0.1:{fast_port}/v1/checkout/sessions"), headers: vec![] }, b"a=1".to_vec()));
let fast = began.elapsed();
let (slow_took, slow_ok) = poller.join().unwrap();
assert!(r.is_ok(), "{r:?}");
assert!(fast < Duration::from_millis(1000), "the fast call waited {fast:?} behind the slow one");
// And the slow one met ITS deadline (2 s), driven while the other thread ran.
assert!(!slow_ok && slow_took < Duration::from_millis(3500), "the slow call's deadline was not driven: {slow_took:?}");
}
}