crates/fast-edge/src/main.rs
//! `fast-edge --listen 0.0.0.0:443 --to 127.0.0.1:50061 --cert cert.pem --key key.pem`: TLS in,
//! the plaintext bytes out. monetize-admin's `server` role serves gRPC h2c and the HTTP callback
//! on one plain port; its PID 1 serves TLS on :443 in front of it. The fast test runs the
//! `server` role, so this is its :443: a sold box's pull channel (holger-fleet, `fleet_at` with
//! `fleet_ca`) and its front's shop door dial it exactly as they dial the real edge.
use std::sync::Arc;
use rustls_pki_types::pem::{self, PemObject};
use rustls_pki_types::{CertificateDer, PrivateKeyDer};
use tokio::net::{TcpListener, TcpStream};
use tokio_rustls::rustls::ServerConfig;
use tokio_rustls::TlsAcceptor;
fn arg(args: &[String], flag: &str) -> Option<String> {
args.iter()
.position(|a| a == flag)
.and_then(|i| args.get(i + 1))
.cloned()
}
/// The server's TLS: the certificate chain and key from PEM files, ALPN h2 then http/1.1 (the
/// pull channel is gRPC, the callback HTTP/1.1). Every refusal names the file it is about.
fn tls_config(cert: &str, key: &str) -> Result<ServerConfig, String> {
let certs: Vec<CertificateDer<'static>> = CertificateDer::pem_file_iter(cert)
.map_err(|e| format!("{cert}: {e}"))?
.filter_map(Result::ok)
.collect();
if certs.is_empty() {
return Err(format!("{cert}: no certificate in the PEM"));
}
let key: PrivateKeyDer<'static> = match PrivateKeyDer::from_pem_file(key) {
Ok(k) => k,
Err(pem::Error::NoItemsFound) => return Err(format!("{key}: no private key in the PEM")),
Err(e) => return Err(format!("{key}: {e}")),
};
let mut config = ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certs, key)
.map_err(|e| format!("the certificate and key: {e}"))?;
config.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
Ok(config)
}
/// The edge, ready: its listener bound, its TLS loaded, where it copies to. A missing `--cert`
/// or `--key` is refused before anything is read or bound.
async fn start(args: &[String]) -> Result<(TcpListener, TlsAcceptor, String), String> {
let listen = arg(args, "--listen").unwrap_or_else(|| "0.0.0.0:443".into());
let to = arg(args, "--to").unwrap_or_else(|| "127.0.0.1:50061".into());
let (Some(cert), Some(key)) = (arg(args, "--cert"), arg(args, "--key")) else {
return Err("--cert and --key (PEM) are required".into());
};
let acceptor = TlsAcceptor::from(Arc::new(tls_config(&cert, &key)?));
let listener = TcpListener::bind(&listen)
.await
.map_err(|e| format!("{listen}: {e}"))?;
Ok((listener, acceptor, to))
}
/// One connection: the TLS handshake, then the plaintext both ways to `to`. A failed handshake
/// never dials `to`; an unreachable `to` closes the TLS side.
async fn bridge(acceptor: TlsAcceptor, tcp: TcpStream, peer: std::net::SocketAddr, to: String) {
let mut tls = match acceptor.accept(tcp).await {
Ok(t) => t,
Err(e) => {
eprintln!("fast-edge: {peer}: TLS: {e}");
return;
}
};
let mut inner = match TcpStream::connect(&to).await {
Ok(s) => s,
Err(e) => {
eprintln!("fast-edge: {to}: {e}");
return;
}
};
let _ = tokio::io::copy_bidirectional(&mut tls, &mut inner).await;
}
/// Accept for ever, one task per connection.
async fn serve(listener: TcpListener, acceptor: TlsAcceptor, to: String) {
loop {
let Ok((tcp, peer)) = listener.accept().await else {
continue;
};
tokio::spawn(bridge(acceptor.clone(), tcp, peer, to.clone()));
}
}
#[tokio::main(flavor = "multi_thread", worker_threads = 2)]
async fn main() {
let args: Vec<String> = std::env::args().skip(1).collect();
let (listener, acceptor, to) = match start(&args).await {
Ok(s) => s,
Err(e) => {
eprintln!("fast-edge: {e}");
std::process::exit(2);
}
};
match listener.local_addr() {
Ok(at) => println!("fast-edge: TLS on {at} -> {to}"),
Err(_) => println!("fast-edge: TLS -> {to}"),
}
serve(listener, acceptor, to).await
}
#[cfg(test)]
mod tests {
//! The edge's few words (lane TEST-SWEEP-MONETIZE, 2026-10-24), and since MON-OSSF
//! (2026-10-31) the edge itself: its TLS refusals and a real TLS client through it to a
//! plain upstream. tests/fixtures holds a throwaway P-256 test CA (ca.pem) and the leaf it
//! signed for `localhost` / 127.0.0.1 (cert.pem, key.pem); its CA key was not kept. Nothing secret.
use super::*;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio_rustls::rustls::pki_types::ServerName;
use tokio_rustls::rustls::{ClientConfig, RootCertStore};
fn words(w: &[&str]) -> Vec<String> {
w.iter().map(|s| s.to_string()).collect()
}
fn fixture(name: &str) -> String {
format!("{}/tests/fixtures/{name}", env!("CARGO_MANIFEST_DIR"))
}
#[test]
fn a_flag_is_the_word_after_it_and_a_missing_or_last_flag_is_none() {
let a = words(&[
"--listen",
"0.0.0.0:443",
"--to",
"127.0.0.1:50061",
"--cert",
]);
assert_eq!(arg(&a, "--listen").as_deref(), Some("0.0.0.0:443"));
assert_eq!(arg(&a, "--to").as_deref(), Some("127.0.0.1:50061"));
assert_eq!(
arg(&a, "--cert"),
None,
"a flag with no word after it names nothing"
);
assert_eq!(arg(&a, "--key"), None);
assert_eq!(arg(&[], "--listen"), None);
}
#[test]
fn the_first_spelling_of_a_flag_wins() {
let a = words(&["--to", "a:1", "--to", "b:2"]);
assert_eq!(arg(&a, "--to").as_deref(), Some("a:1"));
}
#[test]
fn the_tls_config_takes_the_pair_and_offers_h2_then_http1() {
let c = tls_config(&fixture("cert.pem"), &fixture("key.pem")).expect("the fixture pair");
assert_eq!(c.alpn_protocols, vec![b"h2".to_vec(), b"http/1.1".to_vec()]);
}
#[test]
fn the_tls_config_refuses_by_name_what_it_cannot_serve() {
let missing = tls_config("/nonexistent/cert.pem", &fixture("key.pem")).unwrap_err();
assert!(missing.starts_with("/nonexistent/cert.pem:"), "{missing}");
// a key file holding only a certificate: no private key in it
let no_key = tls_config(&fixture("cert.pem"), &fixture("cert.pem")).unwrap_err();
assert!(no_key.contains("no private key in the PEM"), "{no_key}");
// a certificate file holding only a key: no certificate in it
let no_cert = tls_config(&fixture("key.pem"), &fixture("key.pem")).unwrap_err();
assert!(no_cert.contains("no certificate in the PEM"), "{no_cert}");
// a key that is not the certificate's
let other = tls_config(&fixture("cert.pem"), &fixture("other-key.pem")).unwrap_err();
assert!(other.starts_with("the certificate and key:"), "{other}");
}
#[tokio::test]
async fn start_refuses_without_the_pair_and_before_it_binds() {
let e = start(&words(&["--listen", "127.0.0.1:0", "--cert", "c.pem"]))
.await
.err()
.expect("refused");
assert_eq!(e, "--cert and --key (PEM) are required");
let e = start(&words(&[
"--listen",
"256.0.0.1:1",
"--cert",
&fixture("cert.pem"),
"--key",
&fixture("key.pem"),
]))
.await
.err()
.expect("refused");
assert!(
e.starts_with("256.0.0.1:1:"),
"an address that does not bind is named: {e}"
);
}
fn client() -> tokio_rustls::TlsConnector {
let mut roots = RootCertStore::empty();
for c in CertificateDer::pem_file_iter(fixture("ca.pem")).unwrap() {
roots.add(c.unwrap()).unwrap();
}
let mut cfg = ClientConfig::builder()
.with_root_certificates(roots)
.with_no_client_auth();
cfg.alpn_protocols = vec![b"h2".to_vec()];
tokio_rustls::TlsConnector::from(Arc::new(cfg))
}
/// The edge on 127.0.0.1:0 in front of `to`; answers its address.
async fn edge(to: String) -> std::net::SocketAddr {
let (listener, acceptor, to) = start(&words(&[
"--listen",
"127.0.0.1:0",
"--to",
&to,
"--cert",
&fixture("cert.pem"),
"--key",
&fixture("key.pem"),
]))
.await
.unwrap();
let at = listener.local_addr().unwrap();
tokio::spawn(serve(listener, acceptor, to));
at
}
#[tokio::test]
async fn a_tls_client_reaches_the_plain_upstream_through_the_edge_both_ways() {
// the upstream: plaintext, answers each read with its bytes upper-cased
let up = TcpListener::bind("127.0.0.1:0").await.unwrap();
let up_at = up.local_addr().unwrap();
tokio::spawn(async move {
let (mut s, _) = up.accept().await.unwrap();
let mut buf = [0u8; 64];
let n = s.read(&mut buf).await.unwrap();
s.write_all(&buf[..n].to_ascii_uppercase()).await.unwrap();
});
let at = edge(up_at.to_string()).await;
let tcp = TcpStream::connect(at).await.unwrap();
let mut tls = client()
.connect(ServerName::try_from("localhost").unwrap(), tcp)
.await
.expect("the edge's certificate verifies");
assert_eq!(
tls.get_ref().1.alpn_protocol(),
Some(&b"h2"[..]),
"h2 is offered, so gRPC rides it"
);
tls.write_all(b"fleet call").await.unwrap();
let mut got = vec![0u8; 10];
tls.read_exact(&mut got).await.unwrap();
assert_eq!(
got, b"FLEET CALL",
"the bytes went to the upstream in plain and came back through TLS"
);
}
#[tokio::test]
async fn a_client_that_speaks_no_tls_never_reaches_the_upstream() {
let up = TcpListener::bind("127.0.0.1:0").await.unwrap();
let up_at = up.local_addr().unwrap();
let at = edge(up_at.to_string()).await;
let mut plain = TcpStream::connect(at).await.unwrap();
plain
.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n")
.await
.unwrap();
let mut sink = Vec::new();
let _ = tokio::time::timeout(
std::time::Duration::from_secs(5),
plain.read_to_end(&mut sink),
)
.await
.expect("the edge closes the connection");
let dialled =
tokio::time::timeout(std::time::Duration::from_millis(300), up.accept()).await;
assert!(
dialled.is_err(),
"a failed handshake never dials the upstream"
);
}
#[tokio::test]
async fn an_unreachable_upstream_closes_the_tls_side() {
// a port that was bound and let go: nothing listens there
let gone = TcpListener::bind("127.0.0.1:0")
.await
.unwrap()
.local_addr()
.unwrap();
let at = edge(gone.to_string()).await;
let tcp = TcpStream::connect(at).await.unwrap();
let mut tls = client()
.connect(ServerName::try_from("localhost").unwrap(), tcp)
.await
.expect("the handshake still completes");
let mut buf = Vec::new();
let n = tokio::time::timeout(std::time::Duration::from_secs(5), tls.read_to_end(&mut buf))
.await
.expect("closed, not hung");
assert!(
n.is_err() || buf.is_empty(),
"nothing comes back from an upstream that is not there: {n:?} {buf:?}"
);
}
}