crates/monetize-shop-pairing/src/sale_check.rs
//! **The sale check's transport** (S4, the owner 2026-10-11): before a sale is captured, the
//! sold instance's PUBLIC front — `https://<company>.<apex>`, what the buyer reaches, never the
//! estate's WireGuard or utility net — proves it serves what was sold. WHAT it must prove is the
//! product's own sale proof ([`crate::sale_proof`], from its facts); this module is HOW the front
//! is asked: a [`Front`], and [`Https`], the real one (HTTP/1.1 over rustls, the web's roots
//! plus the certificate the shop itself minted for the front, [`FRONT_CERT`]).
//! The invite and the bearer are never in a word it says.
use std::io::{Read as _, Write as _};
use std::net::{SocketAddr, TcpStream, ToSocketAddrs as _};
use std::path::Path;
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use monetize_server::estate_order::SaleCheckAt;
/// The certificate the shop minted for the sold front (`shop.mint_certificate`), kept beside
/// the order's run at the install: the sale check trusts it for that front alone.
pub const FRONT_CERT: &str = "front-cert.pem";
/// How long one request to the front may take, connect and answer each.
pub const PATIENCE: Duration = Duration::from_secs(20);
/// The most one request to the front may take in all, its connect, its TLS and its whole
/// answer (A22 #6, F-T-09): [`PATIENCE`] bounds each read, and a front that drips its answer
/// a byte at a time answers every read.
pub const ANSWER_WITHIN: Duration = Duration::from_secs(30);
/// The most of an answer read.
pub const MAX_ANSWER: usize = 8 << 20;
/// **The sold front, as the buyer reaches it**: `GET path_and_query`, with a bearer or none,
/// answered with its status and body.
pub trait Front {
fn get(&self, path_and_query: &str, bearer: Option<&str>) -> Result<(u16, String), String>;
/// `POST path` with a JSON `body` (holger's ceremony doors take JSON). A front that cannot
/// says so.
fn post(&self, path: &str, body: &str, bearer: Option<&str>) -> Result<(u16, String), String> {
let _ = (body, bearer);
Err(format!("this front cannot POST {path}"))
}
}
/// **Where the front is reached** when not by the host's own DNS (the UpCloud mock forwards a
/// guest's :443 to a loopback port). MOCK ONLY: a shop resolves `<host>` as its buyer does.
pub type FrontAt = fn(&SaleCheckAt<'_>) -> Result<SocketAddr, String>;
static FRONT_AT: OnceLock<FrontAt> = OnceLock::new();
/// Reach every sold front at `at` from now on (the bench over the UpCloud mock). Once.
pub fn front_reached_at(at: FrontAt) -> bool {
FRONT_AT.set(at).is_ok()
}
/// **Where the sold front is reached**: where [`front_reached_at`] says, else by DNS.
pub fn front_addr(at: &SaleCheckAt<'_>) -> Result<SocketAddr, String> {
match FRONT_AT.get() {
Some(f) => f(at),
None => by_dns(at.host),
}
}
/// `host:443` by DNS, as the buyer's browser finds it.
pub fn by_dns(host: &str) -> Result<SocketAddr, String> {
(host, 443u16).to_socket_addrs().map_err(|e| format!("{host} does not resolve: {e}"))?.next().ok_or_else(|| format!("{host} resolves to no address"))
}
/// A body, short and on one line, for a word.
pub(crate) fn excerpt(body: &str) -> String {
let one: String = body.split_whitespace().collect::<Vec<_>>().join(" ");
match one.char_indices().nth(160) {
Some((i, _)) => format!("{}…", &one[..i]),
None => one,
}
}
/// `text` with every non-empty `secret` in it withheld.
pub(crate) fn scrub(text: &str, secrets: &[&str]) -> String {
secrets.iter().filter(|s| s.len() >= 4).fold(text.to_string(), |t, s| t.replace(s, "<withheld>"))
}
/// Percent-encoded for a query value (RFC 3986 unreserved kept).
pub(crate) fn query(v: &str) -> String {
v.bytes().map(|b| if b.is_ascii_alphanumeric() || b"-._~".contains(&b) { (b as char).to_string() } else { format!("%{b:02X}") }).collect()
}
/// **The front over HTTPS**: one HTTP/1.1 request per connection to `addr`, the TLS name
/// `host`, trusted to the web's roots and to `pin` (the certificate the shop minted for it).
pub struct Https {
host: String,
/// The `Host` the request names when it is not the TLS name (OX-WGDS3: the repository host
/// `repo.<host>`, reached under the front's own certificate, which names `<host>`).
addr: SocketAddr,
tls: Arc<rustls::ClientConfig>,
/// [`ANSWER_WITHIN`] (a test's shorter one).
within: Duration,
}
impl Https {
pub fn new(host: &str, addr: SocketAddr, pin: Option<&[u8]>) -> Result<Https, String> {
use rustls::pki_types::pem::PemObject as _;
let mut roots = rustls::RootCertStore { roots: webpki_roots::TLS_SERVER_ROOTS.to_vec() };
if let Some(pem) = pin {
for cert in rustls::pki_types::CertificateDer::pem_slice_iter(pem) {
let cert = cert.map_err(|e| format!("{FRONT_CERT}: {e}"))?;
roots.add(cert).map_err(|e| format!("{FRONT_CERT}: {e}"))?;
}
}
let tls = rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
.with_safe_default_protocol_versions()
.map_err(|e| format!("the sale check's TLS: {e}"))?
.with_root_certificates(roots)
.with_no_client_auth();
Ok(Https { host: host.to_string(), addr, tls: Arc::new(tls), within: ANSWER_WITHIN })
}
}
impl Front for Https {
fn get(&self, path_and_query: &str, bearer: Option<&str>) -> Result<(u16, String), String> {
self.ask("GET", path_and_query, None, bearer)
}
fn post(&self, path: &str, body: &str, bearer: Option<&str>) -> Result<(u16, String), String> {
self.ask("POST", path, Some(body), bearer)
}
}
impl Https {
/// One HTTP/1.1 request on its own connection; a JSON body when `body` is given.
fn ask(&self, method: &str, path_and_query: &str, body: Option<&str>, bearer: Option<&str>) -> Result<(u16, String), String> {
let raw = self.exchange_www(method, path_and_query, body.map(str::as_bytes), body.map(|_| "application/json"), bearer)?;
parse_answer(&raw)
}
/// [`Self::exchange`], and when the answer is the bare host's one hop to `www.<host>`
/// ([`www_hop`]) the same request once more under the www name (its TLS name and `Host`), at
/// the same address. A front asked under another `Host` (`repo.<host>`) is never moved.
fn exchange_www(&self, method: &str, path_and_query: &str, body: Option<&[u8]>, content_type: Option<&str>, bearer: Option<&str>) -> Result<Vec<u8>, String> {
let raw = self.exchange(method, path_and_query, body, content_type, bearer)?;
let status = parse_answer_bytes(&raw).map(|(s, _)| s).unwrap_or(0);
if www_hop(&self.host, path_and_query, status, header_in(&raw, "location").as_deref()) {
let www = Https { host: format!("www.{}", self.host), addr: self.addr, tls: Arc::clone(&self.tls), within: self.within };
return www.exchange(method, path_and_query, body, content_type, bearer);
}
Ok(raw)
}
/// One HTTP/1.1 exchange on its own connection: the whole raw answer, within its deadline.
fn exchange(&self, method: &str, path_and_query: &str, body: Option<&[u8]>, content_type: Option<&str>, bearer: Option<&str>) -> Result<Vec<u8>, String> {
let name = rustls::pki_types::ServerName::try_from(self.host.clone()).map_err(|e| format!("{}: {e}", self.host))?;
let conn = rustls::ClientConnection::new(Arc::clone(&self.tls), name).map_err(|e| format!("TLS: {e}"))?;
let until = std::time::Instant::now() + within_for(self.within, body.map_or(0, <[u8]>::len));
let left = || until.saturating_duration_since(std::time::Instant::now()).min(PATIENCE);
let late = || format!("{} did not answer {method} {path_and_query} within {} s", self.addr, within_for(self.within, body.map_or(0, <[u8]>::len)).as_secs_f64());
let tcp = TcpStream::connect_timeout(&self.addr, left()).map_err(|e| format!("{}: {e}", self.addr))?;
tcp.set_read_timeout(Some(left())).ok();
tcp.set_write_timeout(Some(left())).ok();
let mut tls = rustls::StreamOwned::new(conn, tcp);
let auth = bearer.map(|b| format!("Authorization: Bearer {b}\r\n")).unwrap_or_default();
let typed = content_type.map(|t| format!("Content-Type: {t}\r\n")).unwrap_or_default();
let sent = match body {
Some(b) => format!("{typed}Content-Length: {}\r\n", b.len()),
None if matches!(method, "POST" | "PUT") => "Content-Length: 0\r\n".to_string(),
None => String::new(),
};
let host = &self.host;
let head = format!("{method} {path_and_query} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: monetize-sale-check\r\nAccept: */*\r\n{auth}{sent}Connection: close\r\n\r\n");
tls.write_all(head.as_bytes()).and_then(|()| tls.flush()).map_err(|e| format!("TLS to {}: {e}", self.addr))?;
// The body in pieces, the front's answer looked for before each: a front (or the box
// behind it) that answers BEFORE it reads the whole body (a refusal) is heard, and
// nothing more is sent (OX-WGDS4; a 3 MB bundle said only "Broken pipe").
let mut early = Vec::new();
for piece in body.unwrap_or_default().chunks(BODY_PIECE) {
match answered(&mut tls.conn, &mut tls.sock, &mut early) {
Ok(true) => break,
Ok(false) => {}
Err(e) => return Err(format!("TLS from {}: {e}", self.addr)),
}
if let Err(e) = tls.write_all(piece).and_then(|()| tls.flush()) {
// The pipe broke under the body: the answer the front sent before it closed
// is already here, and it is the answer.
let _ = early_answer(&mut tls.conn, &mut tls.sock, &mut early, left());
if parse_answer_bytes(&early).is_ok() {
return Ok(early);
}
return Err(format!("TLS to {}: {e}", self.addr));
}
}
let mut raw = early;
let mut buf = [0u8; 16 * 1024];
loop {
// The whole answer within its deadline: each read waits only what is left of it.
let wait = left();
if wait.is_zero() {
return Err(late());
}
tls.sock.set_read_timeout(Some(wait)).ok();
match tls.read(&mut buf) {
Ok(0) => break,
Ok(n) => {
raw.extend_from_slice(&buf[..n]);
if raw.len() > MAX_ANSWER {
return Err(format!("an answer over {MAX_ANSWER} bytes"));
}
}
// A peer that closes without TLS's close_notify: the framing below decides.
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof && !raw.is_empty() => break,
Err(_) if left().is_zero() => return Err(late()),
Err(e) => return Err(format!("TLS from {}: {e}", self.addr)),
}
}
Ok(raw)
}
}
/// Pure: **the whole exchange's deadline for a body of `len` bytes**: `within`, and a second
/// more for every 256 KiB sent (an artifact through the front over a slow link is no stall).
pub fn within_for(within: Duration, len: usize) -> Duration {
within + Duration::from_secs((len / (256 << 10)) as u64)
}
/// How much of a request's body is written between two looks for an early answer.
pub const BODY_PIECE: usize = 64 << 10;
/// **Has the peer answered already?** What it sent so far is read without waiting (TLS
/// records decrypted, the plaintext appended to `out`): `true` once there is an answer's
/// first byte, or the peer closed.
fn answered(conn: &mut rustls::ClientConnection, sock: &mut TcpStream, out: &mut Vec<u8>) -> std::io::Result<bool> {
sock.set_nonblocking(true)?;
let mut closed = false;
let read = loop {
match conn.read_tls(sock) {
Ok(0) => {
closed = true;
break Ok(());
}
Ok(_) => {
if let Err(e) = conn.process_new_packets() {
break Err(std::io::Error::new(std::io::ErrorKind::InvalidData, e));
}
}
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break Ok(()),
Err(e) => break Err(e),
}
};
sock.set_nonblocking(false)?;
read?;
let mut buf = [0u8; 16 * 1024];
loop {
match conn.reader().read(&mut buf) {
Ok(0) => return Ok(true),
Ok(n) => out.extend_from_slice(&buf[..n]),
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => return Ok(closed || !out.is_empty()),
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(true),
Err(e) => return Err(e),
}
}
}
/// **What the peer already sent**, read after our write broke: the TLS records in the
/// socket's receive queue, decrypted, without writing anything more (the connection's pending
/// writes would break again). Ends at the peer's close, or when `within` passed.
fn early_answer(conn: &mut rustls::ClientConnection, sock: &mut TcpStream, out: &mut Vec<u8>, within: Duration) -> std::io::Result<()> {
let until = std::time::Instant::now() + within;
let mut buf = [0u8; 16 * 1024];
loop {
match conn.reader().read(&mut buf) {
Ok(0) => return Ok(()),
Ok(n) => {
out.extend_from_slice(&buf[..n]);
if out.len() > MAX_ANSWER {
return Ok(());
}
continue;
}
Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {}
Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => return Ok(()),
Err(e) => return Err(e),
}
let left = until.saturating_duration_since(std::time::Instant::now());
if left.is_zero() {
return Ok(());
}
sock.set_read_timeout(Some(left)).ok();
match conn.read_tls(sock) {
Ok(0) => return Ok(()),
Ok(_) => {
conn.process_new_packets().map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
}
// The reset that followed the answer: what was read before it stands.
Err(_) if !out.is_empty() => return Ok(()),
Err(e) => return Err(e),
}
}
}
/// Pure: an HTTP/1.1 answer — its status and its whole body (`Content-Length` or chunked;
/// cut short is an error).
pub fn parse_answer(raw: &[u8]) -> Result<(u16, String), String> {
parse_answer_bytes(raw).map(|(s, b)| (s, String::from_utf8_lossy(&b).into_owned()))
}
/// Pure (WWW-CANON, owner 2026-10-30: the sites are `www.<site>`): **is this answer the bare host's
/// one hop to its www name?** edda estate-front 0.3.3 answers a request to a served bare `<host>` whose
/// `www.<host>` is served with a 301 (GET, HEAD) or a 308 (any other method) to `https://www.<host>`
/// and the same path and query. Only that hop is followed, once; any other redirect is the answer.
pub(crate) fn www_hop(host: &str, path_and_query: &str, status: u16, location: Option<&str>) -> bool {
matches!(status, 301 | 308) && !host.starts_with("www.") && location == Some(format!("https://www.{host}{path_and_query}").as_str())
}
/// The value of `name` in an HTTP answer's head, if any.
fn header_in(raw: &[u8], name: &str) -> Option<String> {
let end = raw.windows(4).position(|w| w == b"\r\n\r\n")?;
let head = std::str::from_utf8(&raw[..end]).ok()?;
head.split("\r\n").skip(1).find_map(|l| l.split_once(':').filter(|(k, _)| k.trim().eq_ignore_ascii_case(name)).map(|(_, v)| v.trim().to_string()))
}
/// Pure: [`parse_answer`], the body as its bytes.
pub fn parse_answer_bytes(raw: &[u8]) -> Result<(u16, Vec<u8>), String> {
let end = raw.windows(4).position(|w| w == b"\r\n\r\n").ok_or("no HTTP answer")?;
let head = std::str::from_utf8(&raw[..end]).map_err(|_| "an HTTP head that is not text")?;
let mut lines = head.split("\r\n");
let status_line = lines.next().unwrap_or_default();
let status: u16 = status_line.split_whitespace().nth(1).and_then(|s| s.parse().ok()).filter(|_| status_line.starts_with("HTTP/1.")).ok_or_else(|| format!("not an HTTP answer: {}", excerpt(status_line)))?;
let header = |name: &str| lines.clone().find_map(|l| l.split_once(':').filter(|(k, _)| k.trim().eq_ignore_ascii_case(name)).map(|(_, v)| v.trim().to_string()));
let rest = &raw[end + 4..];
let body = if header("transfer-encoding").is_some_and(|v| v.to_ascii_lowercase().contains("chunked")) {
dechunk(rest)?
} else if let Some(len) = header("content-length") {
let len: usize = len.parse().map_err(|_| format!("Content-Length {len:?}"))?;
rest.get(..len).ok_or_else(|| format!("an answer cut short ({} of {len} bytes)", rest.len()))?.to_vec()
} else {
rest.to_vec()
};
Ok((status, body))
}
fn dechunk(mut rest: &[u8]) -> Result<Vec<u8>, String> {
let mut out = Vec::new();
loop {
let eol = rest.windows(2).position(|w| w == b"\r\n").ok_or("a chunked answer cut short")?;
let size = std::str::from_utf8(&rest[..eol]).ok().and_then(|s| usize::from_str_radix(s.split(';').next().unwrap_or_default().trim(), 16).ok()).ok_or("a chunk size that is not hex")?;
rest = &rest[eol + 2..];
if size == 0 {
return Ok(out);
}
out.extend_from_slice(rest.get(..size).ok_or("a chunked answer cut short")?);
rest = rest.get(size + 2..).ok_or("a chunked answer cut short")?;
}
}
/// **Keep the front's minted certificate for the sale check** (`<run>/`[`FRONT_CERT`]) when
/// the shop minted one for it (`tls.<host>.cert` in the front's feed); else none is kept and
/// the front is trusted to the web's roots alone.
pub fn keep_front_cert(run: &Path, host: &str, front_feed: &std::collections::BTreeMap<String, String>) -> Result<(), String> {
let at = run.join(FRONT_CERT);
match front_feed.get(&format!("tls.{host}.cert")) {
Some(pem) => crate::write_atomic(run, FRONT_CERT, pem.as_bytes()).map_err(|e| format!("keeping the front's certificate at {}: {e}", at.display())),
None => match std::fs::remove_file(&at) {
Ok(()) => Ok(()),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(format!("{}: {e}", at.display())),
},
}
}
#[cfg(test)]
mod tests;
#[cfg(test)]
mod holger_tests;