crates/installer/nordisk-twin/src/net.rs
//! **The door's accept loop and the reader's dialler** (feature `tokio`).
//!
//! Both are shaped for tonic without naming it: [`incoming`] is a stream of
//! finished TLS 1.3 streams for `Server::serve_with_incoming` (tonic reads the
//! peer's certificate off a `tokio_rustls::server::TlsStream`, so
//! `Request::peer_certs` works as on its own TLS), and [`Dialer`] is a
//! `tower::Service<Uri>` for `Endpoint::connect_with_connector`. The Uri the
//! channel is built with is a label (`http://<door name>:<port>`, `http` so
//! tonic does not wrap a second TLS around ours); the dialler ignores its host.
use std::future::Future;
use std::net::SocketAddr;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use std::task::{Context, Poll};
use std::time::Duration;
use tokio::net::{TcpListener, TcpStream};
use crate::credential::Credential;
use crate::Result;
/// How long a door may take to answer the reader's TCP probe before the next
/// recorded host is tried — gunnar's `pick_replication_dial` figure.
pub const PROBE: Duration = Duration::from_secs(3);
/// How long a handshake may take on the door before the connection is dropped
/// (and named `handshake`).
pub const HANDSHAKE: Duration = Duration::from_secs(10);
/// **The first door that answers a TCP probe within `probe`**, in order — the
/// tunnel address before the utility one when a source records both.
pub async fn pick(doors: &[String], probe: Duration) -> Option<SocketAddr> {
for d in doors {
let Ok(addr) = d.parse::<SocketAddr>() else { continue };
if let Ok(Ok(_)) = tokio::time::timeout(probe, TcpStream::connect(addr)).await {
return Some(addr);
}
}
None
}
/// **The reader's dialler**: pick a door, connect, TLS 1.3 with the pinned CA
/// and the reader's own certificate, checked against the door's fixed name.
#[derive(Clone)]
pub struct Dialer {
cfg: Arc<rustls::ClientConfig>,
name: rustls::pki_types::ServerName<'static>,
doors: Vec<String>,
probe: Duration,
last: Arc<Mutex<Option<SocketAddr>>>,
}
impl Dialer {
pub fn new(cred: &Credential, server_name: &str, probe: Duration) -> Result<Dialer> {
Ok(Dialer { cfg: cred.client_config()?, name: crate::tls::server_name(server_name)?, doors: cred.doors(), probe, last: Arc::new(Mutex::new(None)) })
}
/// The door the last dial reached, for a log line.
pub fn last(&self) -> Option<SocketAddr> {
*self.last.lock().unwrap_or_else(|p| p.into_inner())
}
/// The doors this dialler tries, in order.
pub fn doors(&self) -> &[String] {
&self.doors
}
}
impl tower_service::Service<http::Uri> for Dialer {
type Response = hyper_util::rt::TokioIo<tokio_rustls::client::TlsStream<TcpStream>>;
type Error = std::io::Error;
type Future = Pin<Box<dyn Future<Output = std::io::Result<Self::Response>> + Send>>;
fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, _label: http::Uri) -> Self::Future {
let me = self.clone();
Box::pin(async move {
let addr = pick(&me.doors, me.probe).await.ok_or_else(|| {
std::io::Error::new(std::io::ErrorKind::ConnectionRefused, format!("no replication door answered a {}s probe: {}", me.probe.as_secs(), me.doors.join(", ")))
})?;
*me.last.lock().unwrap_or_else(|p| p.into_inner()) = Some(addr);
let tcp = TcpStream::connect(addr).await?;
let _ = tcp.set_nodelay(true);
let tls = tokio_rustls::TlsConnector::from(me.cfg).connect(me.name, tcp).await?;
Ok(hyper_util::rt::TokioIo::new(tls))
})
}
}
/// Told of every refused handshake: the peer, [`crate::tls::refusal_name`] (or
/// `closed` for a peer that hung up before a byte — a reader's TCP probe), and
/// rustls' own words.
pub type Refused = Arc<dyn Fn(SocketAddr, &'static str, String) + Send + Sync>;
/// A finished TLS 1.3 connection with a verified client certificate.
pub type Accepted = tokio_rustls::server::TlsStream<TcpStream>;
/// **The door's accept loop.** Each handshake runs on its own task (a slow or
/// hostile peer holds up nobody); only a connection whose client certificate
/// chained to the issuer comes out of the stream. The loop ends when the stream
/// is dropped.
pub fn incoming(listener: TcpListener, cfg: Arc<rustls::ServerConfig>, refused: Refused) -> tokio_stream::wrappers::ReceiverStream<std::io::Result<Accepted>> {
let (tx, rx) = tokio::sync::mpsc::channel(64);
let acceptor = tokio_rustls::TlsAcceptor::from(cfg);
tokio::spawn(async move {
loop {
let (tcp, peer) = tokio::select! {
got = listener.accept() => match got {
Ok(x) => x,
Err(_) => {
// EMFILE and friends: breathe, do not spin.
tokio::time::sleep(Duration::from_millis(100)).await;
continue;
}
},
_ = tx.closed() => return,
};
let (acceptor, tx, refused) = (acceptor.clone(), tx.clone(), refused.clone());
tokio::spawn(async move {
match tokio::time::timeout(HANDSHAKE, acceptor.accept(tcp)).await {
Ok(Ok(s)) => {
let _ = tx.send(Ok(s)).await;
}
Ok(Err(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof && e.get_ref().is_none() => refused(peer, "closed", e.to_string()),
Ok(Err(e)) => refused(peer, crate::tls::refusal_name(&e), e.to_string()),
Err(_) => refused(peer, "handshake", format!("no handshake within {}s", HANDSHAKE.as_secs())),
}
});
}
});
tokio_stream::wrappers::ReceiverStream::new(rx)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::credential;
use crate::pki::{mint_request, Issuer, DEFAULT_LEAF_DAYS};
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
/// The door takes the issuer's twin, names a certless peer and another CA's
/// peer, and the dialler skips a dead host for the first that answers.
#[tokio::test]
async fn the_door_admits_the_issuers_twin_only_and_names_every_refusal() {
let root = std::env::temp_dir().join(format!("nordisk-twin-net-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
let issuer = Issuer::load_or_mint(&root.join("p"), "issuer").unwrap();
let (cert, key) = issuer.server_identity("holger-replication", &["127.0.0.1".into()], 30).unwrap();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let said: Arc<Mutex<Vec<&'static str>>> = Arc::default();
let s2 = said.clone();
let mut stream = incoming(listener, crate::tls::server_config(&cert, &key, &issuer.cert_pem).unwrap(), Arc::new(move |_, name, _| s2.lock().unwrap().push(name)));
tokio::spawn(async move {
use tokio_stream::StreamExt as _;
while let Some(Ok(mut s)) = stream.next().await {
let _ = s.write_all(b"ok").await;
let _ = s.shutdown().await;
}
});
let twin = root.join("t");
let csr = credential::request(&twin, "holger-twin").unwrap();
let leaf = issuer.issue_from_csr("holger-twin", &csr, DEFAULT_LEAF_DAYS).unwrap();
// 127.0.0.9 answers nothing on this port → the dialler moves on.
credential::install(&twin, &leaf.cert_pem, &issuer.cert_pem, port, &["127.0.0.9".into(), "127.0.0.1".into()]).unwrap();
let cred = credential::load(&twin).unwrap().unwrap();
let mut d = Dialer::new(&cred, "holger-replication", Duration::from_millis(300)).unwrap();
use tower_service::Service as _;
let io = d.call(http::Uri::from_static("http://holger-replication")).await.expect("the twin gets in");
let mut tls = io.into_inner();
let mut buf = Vec::new();
tls.read_to_end(&mut buf).await.unwrap();
assert_eq!(buf, b"ok");
assert_eq!(d.last().unwrap().ip().to_string(), "127.0.0.1");
// Another CA's certificate: refused at the handshake, by name.
let other = Issuer::load_or_mint(&root.join("o"), "other").unwrap();
let (okey, ocsr) = mint_request("holger-twin").unwrap();
let oleaf = other.issue_from_csr("holger-twin", &ocsr, 30).unwrap();
let ocfg = crate::tls::client_config(&issuer.cert_pem, &oleaf.cert_pem, &okey).unwrap();
let tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
let mut o = tokio_rustls::TlsConnector::from(ocfg).connect(crate::tls::server_name("holger-replication").unwrap(), tcp).await.unwrap();
let mut b = Vec::new();
assert!(o.read_to_end(&mut b).await.is_err() || b.is_empty(), "no byte reaches another CA's peer");
// No certificate at all.
let bare = rustls::ClientConfig::builder_with_provider(crate::tls::tls13_provider())
.with_protocol_versions(&[&rustls::version::TLS13])
.unwrap()
.with_root_certificates({
let mut r = rustls::RootCertStore::empty();
r.add(rustls::pki_types::CertificateDer::from(crate::pki::pem_to_der(&issuer.cert_pem).unwrap())).unwrap();
r
})
.with_no_client_auth();
let tcp = TcpStream::connect(("127.0.0.1", port)).await.unwrap();
let mut n = tokio_rustls::TlsConnector::from(Arc::new(bare)).connect(crate::tls::server_name("holger-replication").unwrap(), tcp).await.unwrap();
let mut b = Vec::new();
assert!(n.read_to_end(&mut b).await.is_err() || b.is_empty());
for _ in 0..50 {
let s = said.lock().unwrap().clone();
if s.contains(&"no-client-certificate") && s.contains(&"not-from-this-issuer") {
let _ = std::fs::remove_dir_all(&root);
return;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
panic!("the door did not name both refusals: {:?}", said.lock().unwrap());
}
/// ★ One twin per principal (Gunnar 273960cb), at the door: once twin B is
/// certified, twin A's leaf (still 365 days valid, still chained) is refused at
/// the handshake and named `superseded`; B gets in.
#[tokio::test]
async fn the_door_refuses_a_superseded_twin_by_name_and_admits_its_replacement() {
let root = std::env::temp_dir().join(format!("nordisk-twin-net-sup-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&root);
let pins = root.join("pins");
let issuer = Issuer::load_or_mint(&root.join("p"), "issuer").unwrap();
let (cert, key) = issuer.server_identity("holger-replication", &["127.0.0.1".into()], 30).unwrap();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let port = listener.local_addr().unwrap().port();
let said: Arc<Mutex<Vec<&'static str>>> = Arc::default();
let s2 = said.clone();
let mut stream = incoming(listener, crate::tls::server_config_current(&cert, &key, &issuer.cert_pem, &pins).unwrap(), Arc::new(move |_, name, _| s2.lock().unwrap().push(name)));
tokio::spawn(async move {
use tokio_stream::StreamExt as _;
while let Some(Ok(mut s)) = stream.next().await {
let _ = s.write_all(b"ok").await;
let _ = s.shutdown().await;
}
});
let twin = |tag: &str| {
let dir = root.join(tag);
let csr = credential::request(&dir, "holger-twin").unwrap();
let leaf = issuer.issue_superseding(&pins, "holger-twin", &csr, DEFAULT_LEAF_DAYS).unwrap();
credential::install(&dir, &leaf.cert_pem, &issuer.cert_pem, port, &["127.0.0.1".into()]).unwrap();
credential::load(&dir).unwrap().unwrap()
};
let dial = |cred: credential::Credential| async move {
use tower_service::Service as _;
let mut d = Dialer::new(&cred, "holger-replication", Duration::from_millis(300)).unwrap();
let mut buf = Vec::new();
if let Ok(io) = d.call(http::Uri::from_static("http://holger-replication")).await {
let _ = io.into_inner().read_to_end(&mut buf).await;
}
buf
};
let a = twin("a");
assert_eq!(dial(a.clone()).await, b"ok", "twin A holds the principal");
let b = twin("b");
assert_eq!(dial(b).await, b"ok", "twin B, certified since, gets in");
assert!(dial(a).await.is_empty(), "no byte reaches the superseded twin A");
for _ in 0..50 {
if said.lock().unwrap().contains(&"superseded") {
let _ = std::fs::remove_dir_all(&root);
return;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
panic!("the door did not name the superseded twin: {:?}", said.lock().unwrap());
}
}