crates/gunnar-server/tests/twin_mtls_pairing.rs
//! **A twin from the one medium, paired at bootstrap, follows over mutual TLS.**
//!
//! Found by T15's bench on a medium carrying `agent/wg-mtls`: the twin refused
//! to follow in the clear (correct) and NOTHING gave it a certificate, so no
//! twin installed from the one medium could ever follow. These tests pin the
//! fix — the certificate minted at pairing, never on the medium — end to end
//! over a real TLS 1.3 listener on loopback:
//!
//! * P1 — paired over the pairing doors (`twin_tls::request` → the primary's
//! `issue_for_twin` → `twin_tls::install`), the twin's `GrpcSource` dials the
//! primary's MUTUAL replication listener and applies the primary's head. The
//! plaintext control plane is not even running in this test: the only door is
//! the one that demands a certificate at the handshake.
//! * P2 — that door really does demand one: a TLS dial with no client
//! certificate gets no answer.
//! * P3 — a twin with no certificate, pointed at another machine, refuses BY
//! NAME rather than following in the clear.
//! * P4 — the install refuses a certificate for someone else's key, and a
//! second issuer (a twin is never re-pointed); the same issuer again (a
//! renewal) is taken.
//! * P5 — the primary refuses to certify a principal that is not in group
//! `twin`, and a primary with no replication listener refuses by name.
use std::net::SocketAddr;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use gunnar_core::{NewRepo, Principal, PrincipalKind, RepoRef};
use gunnar_grpc::{EventBus, ServerConfig, ServiceContext};
use gunnar_server::stores::ServerStores;
use gunnar_server::twin::{self, Follower, GrpcSource};
use gunnar_server::{twin_tls, CoreControlPlane};
struct Primary {
stores: Arc<ServerStores>,
rt: tokio::runtime::Runtime,
replication: SocketAddr,
stop: Option<tokio::sync::oneshot::Sender<()>>,
_dir: tempfile::TempDir,
}
impl Primary {
/// A primary with an issuer, a `twin-box` machine principal in group `twin`,
/// and ONLY its mutual replication listener up — built the way `serve.rs`
/// builds it: leaf from the box's own issuer under the fixed service name.
fn start() -> Primary {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("stores"));
stores
.catalog()
.create_namespace("alice")
.expect("namespace");
stores
.create_repo(
NewRepo::new(RepoRef::new("alice", "one").expect("legal")),
false,
)
.expect("repo");
// The host key, minted at start as `serve.rs` mints it on every boot. Left
// to the twin's first SIGNED `LogHead` (twin::answering_key), the mint is a
// framed secret that grew the primary's log under a read: 49 vs 48, red 49
// in 50 against the exact check.
gunnar_core::ensure_host_key(&stores.secrets().expect("secrets")).expect("host key");
let pki = stores.pki().expect("pki");
let issuer = pki
.create("gunnar issuer", gunnar_core::DEFAULT_ISSUER_DAYS)
.expect("issuer");
let auth = stores.auth();
auth.add_principal(Principal::new("twin-box", "twin-box", PrincipalKind::Machine).unwrap())
.expect("twin principal");
auth.add_principal(Principal::new("twin", "twin", PrincipalKind::Group).unwrap())
.expect("twin group");
auth.add_to_group("twin", "twin-box").expect("membership");
auth.add_principal(Principal::new("stranger", "stranger", PrincipalKind::Machine).unwrap())
.expect("stranger");
let hosts = vec![
gunnar_grpc::client_signer::REPLICATION_SERVER_NAME.to_owned(),
"localhost".to_owned(),
"127.0.0.1".to_owned(),
];
let leaf = pki
.issue_server_identity(&hosts, gunnar_core::DEFAULT_LEAF_DAYS)
.expect("leaf");
let trust = gunnar_grpc::Trust {
client_ca: issuer.cert_pem.clone().into_bytes(),
cert: format!("{}{}", leaf.cert_pem, issuer.cert_pem).into_bytes(),
key: leaf.key_pem.expect("key").into_bytes(),
};
let plane: Arc<dyn gunnar_grpc::ControlPlane> =
Arc::new(CoreControlPlane::new(Arc::clone(&stores)));
// Allow-all at the GATE, so this test is about the TRANSPORT: what it
// proves is that the handshake demanded a certificate from this issuer.
// Who may read the log is the gate's business and is covered elsewhere.
let ctx = ServiceContext::new(
plane,
Arc::new(EventBus::new(16)),
ServerConfig::default().insecure_allow_all_for_local_development(),
);
let rt = tokio::runtime::Builder::new_multi_thread()
.worker_threads(2)
.enable_all()
.build()
.expect("rt");
let std_listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let replication = std_listener.local_addr().expect("addr");
std_listener.set_nonblocking(true).expect("nonblocking");
twin_tls::record_replication_port(stores.root(), Some(replication.port()))
.expect("record port");
let (tx, rx) = tokio::sync::oneshot::channel::<()>();
rt.spawn(async move {
let listener = tokio::net::TcpListener::from_std(std_listener).expect("tokio listener");
let _ = gunnar_grpc::router_with_trust(ctx, || 0, Some(&trust))
.expect("mutual router")
.serve_with_incoming_shutdown(
tokio_stream::wrappers::TcpListenerStream::new(listener),
async {
let _ = rx.await;
},
)
.await;
});
Primary {
stores,
rt,
replication,
stop: Some(tx),
_dir: dir,
}
}
}
impl Drop for Primary {
fn drop(&mut self) {
if let Some(tx) = self.stop.take() {
let _ = tx.send(());
}
}
}
/// A fresh box made a twin of `primary_at` — the address as the TWIN reaches
/// the primary's control plane. Paired, the twin dials that HOST at the
/// replication port the primary reported, whatever port this names.
fn fresh_twin(primary_at: &str) -> (Arc<ServerStores>, tempfile::TempDir) {
let dir = tempfile::tempdir().expect("tempdir");
let stores = Arc::new(ServerStores::open(dir.path()).expect("twin stores"));
twin::become_twin(&stores, primary_at).expect("a fresh box becomes a twin");
(stores, dir)
}
fn pair(primary: &Primary, twin_root: &std::path::Path) -> gunnar_core::expiry::Expiry {
let csr = twin_tls::request(twin_root).expect("the twin mints its key and answers a request");
let (cert, ca, port) = twin_tls::issue_for_twin(&primary.stores, "twin-box", &csr)
.expect("the primary certifies its twin");
assert_eq!(
port,
primary.replication.port(),
"the primary must hand out the port it actually serves on"
);
twin_tls::install(twin_root, &cert, &ca, port)
.expect("the twin takes a certificate for its own key")
}
/// How long P1 gives the twin to take the primary's log. An OUTER bound only:
/// the wait is on the twin's own polls, so a healthy run ends the moment the
/// twin asks again after applying the head, and this is reached only when the
/// follower is truly stuck (it says what it had applied).
const FOLLOW_WITHIN: Duration = Duration::from_secs(120);
/// The twin's source, telling the test what the twin has applied each time it
/// asks the primary for its head. `Follower::run` asks again AT ONCE after a
/// poll that applied (T1), so the ask that follows the last apply is the event
/// P1 waits on — not a clock.
struct Told {
inner: GrpcSource,
twin: Arc<ServerStores>,
tell: std::sync::mpsc::Sender<u64>,
}
impl twin::LogSource for Told {
fn alternatives(&self) -> bool {
self.inner.alternatives()
}
fn head(&self) -> anyhow::Result<(u64, u64)> {
self.inner.head()
}
fn head_and_identity(&self) -> anyhow::Result<(u64, u64, Option<String>)> {
// `Follower::applied_seq`: a step appends and applies before the next ask.
let _ = self.tell.send(self.twin.txlog().head_seq());
self.inner.head_and_identity()
}
fn fetch(
&self,
from: u64,
to: u64,
max_bytes: u64,
) -> anyhow::Result<Vec<gunnar_core::txlog::Entry>> {
self.inner.fetch(from, to, max_bytes)
}
fn snapshot(&self, into: &mut twin::snapshot::Stager) -> anyhow::Result<bool> {
self.inner.snapshot(into)
}
}
/// P1 — the whole point: paired at bootstrap, the twin FOLLOWS, over mTLS.
#[test]
fn a_twin_paired_over_bootstrap_follows_over_mutual_tls() {
// No operator environment may be doing the work the pairing is meant to do.
for var in [
gunnar_grpc::client_signer::CLIENT_CERT_ENV,
gunnar_grpc::client_signer::CLIENT_KEY_ENV,
gunnar_grpc::client_signer::CA_ENV,
] {
assert!(
std::env::var_os(var).is_none(),
"{var} is set in the test environment; this test proves the PAIRED credential"
);
}
let primary = Primary::start();
// The twin names the primary's CONTROL plane on port 1 — nothing listens
// there. Following can only succeed through the port the pairing handed over.
let (twin_stores, _twin_dir) = fresh_twin("127.0.0.1:1");
let expiry = pair(&primary, twin_stores.root());
assert_eq!(
expiry.grade,
gunnar_core::expiry::Grade::Fine,
"a fresh certificate is good for its life: {expiry:?}"
);
assert!(twin_stores.root().join(twin_tls::TLS_CERT).exists());
// The primary's head is read BEFORE the twin exists: nothing writes to the
// primary in this test once it is built, so this is the whole log to follow.
let want = primary.stores.txlog().head_seq();
assert!(want > 0, "the primary has a log to follow");
let key = twin::twin_private_key(twin_stores.root()).expect("the twin's key");
let source = GrpcSource::connect("127.0.0.1:1", key, twin_stores.root())
.expect("a paired twin builds its mutual channel");
let stop = Arc::new(AtomicBool::new(false));
let (s, st) = (Arc::clone(&stop), Arc::clone(&twin_stores));
let (tell, told) = std::sync::mpsc::channel::<u64>();
let follower = std::thread::spawn(move || {
let mut source = Some(source);
loop {
let inner = source.take().unwrap_or_else(|| {
let key = twin::twin_private_key(st.root()).expect("the twin's key");
GrpcSource::connect("127.0.0.1:1", key, st.root())
.expect("the paired twin re-dials its mutual channel")
});
let told = Told {
inner,
twin: Arc::clone(&st),
tell: tell.clone(),
};
match Follower::new(Arc::clone(&st), Box::new(told)).run(Duration::from_millis(100), &s)
{
// What the supervisor does (twin::start_follower_every): a hand-back — a
// LogHead slower than HEAD_WITHIN on a loaded box — is dialled again at once.
twin::Ended::Redial => continue,
ended => return ended,
}
}
});
let deadline = Instant::now() + FOLLOW_WITHIN;
let mut applied = 0;
while applied < want {
match told.recv_timeout(deadline.saturating_duration_since(Instant::now())) {
Ok(seq) => applied = seq,
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
panic!("the paired twin's follower stopped ({:?}) having applied {applied} of {want}", follower.join())
}
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => panic!(
"the paired twin did not follow the primary over mutual TLS within {FOLLOW_WITHIN:?}: applied {} of {want}",
twin_stores.txlog().head_seq()
),
}
}
stop.store(true, Ordering::SeqCst);
assert_eq!(follower.join().expect("follower"), twin::Ended::Stopped);
// A twin's poll is a READ: the primary's log must not move under it. When it
// does, "the twin holds the primary's log" has no fixed answer — the twin may
// or may not have taken the extra frame by the time it is stopped (T1 re-polls
// at once after an apply), and the check can only be `>=`.
assert_eq!(
primary.stores.txlog().head_seq(),
want,
"the primary's log moved under its twin's poll"
);
// Then the twin holds EXACTLY the primary's log: every frame, and nothing else.
assert_eq!(
twin_stores.txlog().head_seq(),
want,
"the twin holds the primary's log"
);
drop(primary);
}
/// P2 — the replication door demands a certificate at the handshake. A TLS
/// client that pins the right issuer but presents NO certificate gets nothing.
#[test]
fn the_replication_listener_refuses_a_peer_with_no_certificate() {
let primary = Primary::start();
let ca = primary
.stores
.pki()
.unwrap()
.issuer()
.unwrap()
.unwrap()
.cert_pem;
let at = format!("https://127.0.0.1:{}", primary.replication.port());
let refused = primary.rt.block_on(async move {
gunnar_grpc::ensure_crypto_provider();
let tls = tonic::transport::ClientTlsConfig::new()
.domain_name(gunnar_grpc::client_signer::REPLICATION_SERVER_NAME)
.ca_certificate(tonic::transport::Certificate::from_pem(ca));
let channel = match tonic::transport::Channel::from_shared(at)
.unwrap()
.tls_config(tls)
.unwrap()
.connect()
.await
{
Ok(channel) => channel,
Err(_) => return true,
};
let mut client = gunnar_proto::mirror_client::MirrorClient::new(channel);
client
.log_head(gunnar_proto::LogHeadRequest::default())
.await
.is_err()
});
assert!(
refused,
"a peer with no client certificate was answered on the mutual replication listener"
);
}
/// P3 — no certificate, a primary on another machine: refused BY NAME.
#[test]
fn a_twin_without_a_certificate_refuses_by_name() {
let (twin_stores, _dir) = fresh_twin("10.0.2.2:50051");
let key = twin::twin_private_key(twin_stores.root()).expect("key");
let err = match GrpcSource::connect("10.0.2.2:50051", key, twin_stores.root()) {
Ok(_) => panic!("a twin with no certificate built a channel to another machine"),
Err(err) => format!("{err:#}"),
};
assert!(err.contains("no client certificate is configured"), "{err}");
assert!(err.contains("It will not"), "{err}");
assert!(
err.contains("KEK"),
"the refusal must say what is at stake: {err}"
);
}
/// P4 — the install takes only a certificate for THIS twin's key, from ONE issuer.
#[test]
fn the_install_refuses_a_foreign_key_and_a_second_issuer_and_takes_a_renewal() {
let primary = Primary::start();
let (twin_stores, _dir) = fresh_twin("127.0.0.1:1");
let root = twin_stores.root();
twin_tls::request(root).expect("request");
// A certificate for SOMEONE ELSE's key, from the right issuer.
let (_other_key, other_csr) =
gunnar_core::issuer::mint_client_request("someone-else").expect("other csr");
let (foreign, ca, port) =
twin_tls::issue_for_twin(&primary.stores, "twin-box", &other_csr).expect("issued");
let err = format!(
"{:#}",
twin_tls::install(root, &foreign, &ca, port).expect_err("a certificate for another key")
);
assert!(err.contains("DIFFERENT key"), "{err}");
assert!(
!root.join(twin_tls::TLS_CERT).exists(),
"nothing may be installed on a refusal"
);
// The real pairing, then a renewal under the same issuer.
pair(&primary, root);
pair(&primary, root);
// A second primary's issuer: a twin is never re-pointed.
let other = Primary::start();
let csr = twin_tls::request(root).expect("same request");
let (cert, other_ca, port) =
twin_tls::issue_for_twin(&other.stores, "twin-box", &csr).expect("issued by the other");
let err = format!(
"{:#}",
twin_tls::install(root, &cert, &other_ca, port).expect_err("a second issuer")
);
assert!(err.contains("DIFFERENT issuer"), "{err}");
}
/// P5 — the primary certifies only a twin, and only through a door it serves.
#[test]
fn the_primary_certifies_only_a_member_of_twin_and_only_with_a_listener() {
let primary = Primary::start();
let (twin_stores, _dir) = fresh_twin("127.0.0.1:1");
let csr = twin_tls::request(twin_stores.root()).expect("request");
let err = format!(
"{:#}",
twin_tls::issue_for_twin(&primary.stores, "stranger", &csr).expect_err("not in group twin")
);
assert!(err.contains("not a member of group `twin`"), "{err}");
twin_tls::record_replication_port(primary.stores.root(), None).expect("listener off");
let err = format!(
"{:#}",
twin_tls::issue_for_twin(&primary.stores, "twin-box", &csr).expect_err("no listener")
);
assert!(
err.contains("serves no mutual replication listener"),
"{err}"
);
}