server/lib/tests/paired_holgers_e2e.rs
//! **Two holgers that know each other — over a real TLS port.**
//!
//! `mirror_mtls_e2e.rs` proves a mirror can AUTHENTICATE to an upstream. This
//! proves the other half: once the two are PAIRED, each side knows who the other
//! is.
//!
//! The case: the cloud holger serves `prod`; the on-prem holger's mirror pulls it
//! with a client certificate. Paired, that pull is `onprem-1`, drawn by name in
//! the "paired downstream holgers" box, and kept out of the anonymous visitor
//! count and the hashed puller band. Unpaired, the same pull is a visitor and
//! nothing else, exactly as before.
//!
//! What is proved here, on the bytes the door serves:
//!
//! 1. a paired pull is drawn BY NAME, with its key, its last pull and today's
//! numbers, and its repository has an arrow into the box;
//! 2. it is NOT counted as an anonymous consumer;
//! 3. an unpaired pull IS counted as an anonymous consumer and is named by
//! nothing;
//! 4. a paired downstream asking for a repository its pairing does not list is
//! refused by name (`holger-peer-2`);
//! 5. no client address appears anywhere in the picture;
//! 6. the downstream's pin check reads the upstream's REAL certificate over a
//! real handshake, and a different certificate is `key-mismatch`;
//! 7. only the paired key may report a sync point (`holger-peer-4` otherwise),
//! and a REAL paired mirror, after one cycle, has told the upstream how far
//! it synced — read back on the admin-read `Downstreams` door and kept under
//! the upstream's data root.
// A server's test, never a gateway build's (lane HOL-CLEANUP): a gateway build is a
// downstream, and several Holgers in one process, and a downstream's egress pin is process-wide.
#![cfg(not(feature = "gateway"))]
use std::io::Write;
use std::sync::Arc;
use traits::test_support::hold_port;
use holger_server_lib::grpc::holger_proto::repository_service_client::RepositoryServiceClient;
use holger_server_lib::grpc::holger_proto::{ArtifactId as PbId, FetchArtifactRequest};
use holger_server_lib::{peers, read_ron_config, wire_holger, Holger};
use traits::{ArtifactId, RepositoryBackendTrait};
// ───────────────────────────── a throwaway PKI ─────────────────────────────
// The same shape `mirror_mtls_e2e.rs` uses: one CA, one server certificate, one
// client certificate per party. rcgen is already this crate's dev-dependency.
struct Pki {
ca_pem: String,
/// rcgen 0.14 (2026-10-02, W3): the CA's params + key sign the leaves.
issuer: rcgen::Issuer<'static, rcgen::KeyPair>,
}
struct Issued {
cert_pem: String,
key_pem: String,
}
impl Pki {
fn new() -> Self {
let mut params = rcgen::CertificateParams::new(Vec::<String>::new()).unwrap();
params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
params
.distinguished_name
.push(rcgen::DnType::CommonName, "holger-pairing-ca");
let ca_key = rcgen::KeyPair::generate().unwrap();
let ca_cert = params.self_signed(&ca_key).unwrap();
Self {
ca_pem: ca_cert.pem(),
issuer: rcgen::Issuer::new(params, ca_key),
}
}
fn server(&self, cn: &str) -> Issued {
let mut params =
rcgen::CertificateParams::new(vec!["localhost".to_string(), "127.0.0.1".to_string()])
.unwrap();
params
.distinguished_name
.push(rcgen::DnType::CommonName, cn);
params.use_authority_key_identifier_extension = true;
let key = rcgen::KeyPair::generate().unwrap();
let cert = params.signed_by(&key, &self.issuer).unwrap();
Issued {
cert_pem: cert.pem(),
key_pem: key.serialize_pem(),
}
}
fn client(&self, cn: &str) -> Issued {
let mut params = rcgen::CertificateParams::new(Vec::<String>::new()).unwrap();
params
.distinguished_name
.push(rcgen::DnType::CommonName, cn);
params.use_authority_key_identifier_extension = true;
params.extended_key_usages = vec![rcgen::ExtendedKeyUsagePurpose::ClientAuth];
let key = rcgen::KeyPair::generate().unwrap();
let cert = params.signed_by(&key, &self.issuer).unwrap();
Issued {
cert_pem: cert.pem(),
key_pem: key.serialize_pem(),
}
}
}
fn write_to(dir: &std::path::Path, name: &str, body: &str) -> String {
let p = dir.join(name);
std::fs::write(&p, body).unwrap();
std::fs::set_permissions(
&p,
<std::fs::Permissions as std::os::unix::fs::PermissionsExt>::from_mode(0o600),
)
.unwrap();
p.to_string_lossy().into_owned()
}
struct Cloud {
_data: tempfile::TempDir,
_certs: tempfile::TempDir,
_cfg: tempfile::NamedTempFile,
_holger: Holger,
endpoint: String,
server_fingerprint: String,
server_cert_pem: String,
data_root: std::path::PathBuf,
}
/// A cloud holger with `prod` and `staging`, a TLS gRPC door that accepts (but
/// does not demand) a client certificate, audit ON — and a PAIRING with
/// `onprem-1` for `prod` alone, recorded before the box starts, exactly as the
/// `pair` verb writes it.
fn start_cloud(pki: &Pki, downstream: &Issued) -> Cloud {
let data = tempfile::tempdir().unwrap();
let certs = tempfile::tempdir().unwrap();
let srv = pki.server("cloud.holger.test");
let cert_path = write_to(certs.path(), "server.crt", &srv.cert_pem);
let key_path = write_to(certs.path(), "server.key", &srv.key_pem);
let ca_path = write_to(certs.path(), "ca.crt", &pki.ca_pem);
for d in ["prod", "staging"] {
std::fs::create_dir_all(data.path().join(d)).unwrap();
}
let grpc = hold_port();
let config = format!(
r#"(
repositories: [
(ron_name: "prod", ron_repo_type: "rust", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "doors")),
ron_data_dir: Some("{prod}")),
(ron_name: "staging", ron_repo_type: "rust", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "doors")),
ron_data_dir: Some("{staging}")),
],
// Several boxes share this test process, and the egress guard is per process: the box
// that names no upstream says `network: Online` so its neighbours may dial (B2).
network: Online,
exposed_endpoints: [
(ron_name: "doors", ron_url: "127.0.0.1:{grpc}",
ron_tls: Some((cert: "{cert}", key: "{key}", client_ca: Some("{ca}"), client_auth_optional: true))),
],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "file", ron_path: "{root}") ],
audit: ( dir: "{root}/holger-audit", enabled: true ),
)"#,
prod = data.path().join("prod").display(),
staging = data.path().join("staging").display(),
root = data.path().display(),
grpc = grpc,
cert = cert_path,
key = key_path,
ca = ca_path,
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(config.as_bytes()).unwrap();
let mut holger: Holger = read_ron_config(cfg.path()).unwrap();
holger.instantiate_backends().unwrap();
wire_holger(&mut holger).unwrap();
// THE PAIRING, written the way the `pair downstream` verb writes it: under
// the data root, before the box starts, for `prod` only.
peers::pair_downstream(
&holger.peers_root(),
"onprem-1",
downstream.cert_pem.as_bytes(),
&["prod".to_string()],
"test",
1,
)
.expect("the pairing is recorded");
for repo in ["prod", "staging"] {
let id = ArtifactId {
namespace: None,
name: "thing".into(),
version: "1.0.0".into(),
};
let backend: Arc<dyn RepositoryBackendTrait> = holger.backend_named(repo).unwrap();
backend.put(&id, b"BYTES").unwrap();
}
holger.start().unwrap();
let data_root = holger.peers_root();
Cloud {
_data: data,
_certs: certs,
_cfg: cfg,
_holger: holger,
endpoint: format!("https://localhost:{grpc}"),
server_fingerprint: peers::fingerprint_pem(srv.cert_pem.as_bytes()).unwrap(),
server_cert_pem: srv.cert_pem.clone(),
data_root,
}
}
/// A gRPC channel to the cloud, with a client certificate or without one.
async fn channel(cloud: &Cloud, pki: &Pki, identity: Option<&Issued>) -> tonic::transport::Channel {
let mut tls = tonic::transport::ClientTlsConfig::new()
.ca_certificate(tonic::transport::Certificate::from_pem(pki.ca_pem.clone()));
if let Some(id) = identity {
tls = tls.identity(tonic::transport::Identity::from_pem(
&id.cert_pem,
&id.key_pem,
));
}
for _ in 0..100 {
match tonic::transport::Channel::from_shared(cloud.endpoint.clone())
.unwrap()
.tls_config(tls.clone())
.unwrap()
.connect()
.await
{
Ok(c) => return c,
Err(_) => tokio::time::sleep(std::time::Duration::from_millis(50)).await,
}
}
panic!("the cloud holger never answered on {}", cloud.endpoint);
}
async fn fetch(ch: &tonic::transport::Channel, repo: &str) -> Result<usize, tonic::Status> {
fetch_as(ch, repo, None).await
}
/// A fetch that arrives through a reverse proxy, which is how a real client
/// reaches holger.rs: the transport peer is loopback and the CLIENT address is
/// the `x-forwarded-for` entry. Without one, a loopback caller is the box
/// talking to itself and is counted as nobody — see `crate::visitors`.
async fn fetch_as(
ch: &tonic::transport::Channel,
repo: &str,
forwarded_for: Option<&str>,
) -> Result<usize, tonic::Status> {
let mut c = RepositoryServiceClient::new(ch.clone());
let mut req = tonic::Request::new(FetchArtifactRequest {
repository: repo.to_string(),
id: Some(PbId {
namespace: String::new(),
name: "thing".into(),
version: "1.0.0".into(),
}),
});
if let Some(xff) = forwarded_for {
req.metadata_mut()
.insert("x-forwarded-for", xff.parse().unwrap());
}
let reply = c.fetch_artifact(req).await?;
Ok(reply.into_inner().data.len())
}
#[tokio::test(flavor = "multi_thread")]
async fn a_paired_downstream_is_held_to_its_pairing() {
let pki = Pki::new();
let onprem = pki.client("onprem-1");
let cloud = start_cloud(&pki, &onprem);
let paired = channel(&cloud, &pki, Some(&onprem)).await;
assert_eq!(
fetch_as(&paired, "prod", Some("198.51.100.4"))
.await
.expect("the paired pull works"),
5
);
// A repository the pairing does not list is refused BY NAME.
let refused = fetch(&paired, "staging")
.await
.expect_err("staging is not paired");
assert_eq!(refused.code(), tonic::Code::PermissionDenied);
assert!(
refused.message().contains("holger-peer-2"),
"{}",
refused.message()
);
}
/// ★ **An UNPAIRED puller is a visitor**: only a PAIRED key is held to its list.
#[tokio::test(flavor = "multi_thread")]
async fn an_unpaired_puller_may_read_every_open_repository() {
let pki = Pki::new();
let onprem = pki.client("onprem-1");
let cloud = start_cloud(&pki, &onprem);
let stranger = pki.client("somebody-else");
let ch = channel(&cloud, &pki, Some(&stranger)).await;
assert_eq!(
fetch_as(&ch, "prod", Some("203.0.113.9"))
.await
.expect("an open read"),
5
);
assert_eq!(
fetch_as(&ch, "staging", Some("203.0.113.9"))
.await
.expect("an open read"),
5
);
}
/// ★ **The downstream's pin check reads the upstream's REAL certificate**, over
/// a real handshake — and another certificate is `key-mismatch`, by name.
#[tokio::test(flavor = "multi_thread")]
async fn the_pin_check_reads_the_upstreams_real_certificate() {
let pki = Pki::new();
let onprem = pki.client("onprem-1");
let cloud = start_cloud(&pki, &onprem);
// Let the door come up.
let _ = channel(&cloud, &pki, Some(&onprem)).await;
let seen = peers::probe_upstream_key(
&cloud.endpoint,
Some((
onprem.cert_pem.clone().into_bytes(),
onprem.key_pem.clone().into_bytes(),
)),
std::time::Duration::from_secs(5),
)
.await
.expect("the upstream answered the probe");
assert_eq!(
seen, cloud.server_fingerprint,
"the probe read the server's own certificate"
);
let pinned = peers::UpstreamRecord {
name: "cloud".into(),
fingerprint: cloud.server_fingerprint.clone(),
source: "cloud".into(),
paired_at: 0,
paired_by: "test".into(),
};
assert_eq!(
peers::verdict(&pinned, Some(&seen)),
peers::LinkState::Paired
);
let impostor = peers::UpstreamRecord {
fingerprint: peers::fingerprint_pem(pki.server("impostor").cert_pem.as_bytes()).unwrap(),
..pinned
};
assert_eq!(
peers::verdict(&impostor, Some(&seen)),
peers::LinkState::KeyMismatch
);
assert_eq!(peers::LinkState::KeyMismatch.word(), "key-mismatch");
}
// ───────────────────────── 7. how far a downstream synced ─────────────────────────
use grow::grow_service_client::GrowServiceClient;
use holger_server_lib::grpc::holger_grow_proto as grow;
fn report_req(repo: &str) -> grow::DownstreamReportRequest {
grow::DownstreamReportRequest {
points: vec![grow::DownstreamPoint {
repository: repo.into(),
applied_boot: 1,
applied_seq: 1,
last_complete_unix_ms: 1,
}],
every_secs: 300,
summary: None,
summary_proof: String::new(),
console_proof: String::new(),
}
}
/// ★ **Only the paired key may say how far it synced**: an unpaired client
/// certificate and no certificate at all are refused by name (`holger-peer-4`),
/// the paired one is answered with the upstream's position.
#[tokio::test(flavor = "multi_thread")]
async fn a_sync_report_needs_the_paired_key() {
let pki = Pki::new();
let onprem = pki.client("onprem-1");
let cloud = start_cloud(&pki, &onprem);
for who in [Some(pki.client("somebody-else")), None] {
let ch = channel(&cloud, &pki, who.as_ref()).await;
let refused = GrowServiceClient::new(ch)
.downstream_report(report_req("prod"))
.await
.expect_err("not paired");
assert_eq!(refused.code(), tonic::Code::PermissionDenied);
assert!(
refused.message().contains("holger-peer-4"),
"{}",
refused.message()
);
}
let paired = channel(&cloud, &pki, Some(&onprem)).await;
let off_list = GrowServiceClient::new(paired.clone())
.downstream_report(report_req("staging"))
.await
.expect_err("staging is not paired");
assert!(
off_list.message().contains("holger-peer-2"),
"{}",
off_list.message()
);
let at = GrowServiceClient::new(paired.clone())
.downstream_report(report_req("prod"))
.await
.expect("the paired key reports")
.into_inner();
assert!(at.boot > 0 && at.head_seq > 0, "{at:?}");
let rows = GrowServiceClient::new(paired)
.downstreams(grow::DownstreamsRequest {})
.await
.unwrap()
.into_inner();
let row = rows
.downstreams
.iter()
.find(|d| d.name == "onprem-1")
.expect("the paired downstream is drawn");
// Applied 1 of ANOTHER boot: no position here, so the whole head is the lag.
assert_eq!(row.repositories[0].repository, "prod");
assert_eq!(row.repositories[0].lag_entries, rows.head_seq);
assert_eq!(row.repositories[0].last_complete_unix_ms, 1);
}
/// ★ **A REAL paired mirror reports how far it synced.** The on-prem holger's
/// mirror (mTLS with the paired certificate, the upstream's key pinned) runs its
/// timer against the cloud; after its first whole cycle the cloud's
/// `Downstreams` reads `prod` applied at the cloud's head — lag 0, `ok`, a
/// completion time — and the record is on the cloud's disk.
#[tokio::test(flavor = "multi_thread")]
async fn a_paired_mirror_tells_its_upstream_how_far_it_synced() {
let pki = Pki::new();
let onprem = pki.client("onprem-1");
let cloud = start_cloud(&pki, &onprem);
let _ = channel(&cloud, &pki, Some(&onprem)).await;
let data = tempfile::tempdir().unwrap();
let certs = tempfile::tempdir().unwrap();
let c = write_to(certs.path(), "mirror.crt", &onprem.cert_pem);
let k = write_to(certs.path(), "mirror.key", &onprem.key_pem);
let ca = write_to(certs.path(), "ca.crt", &pki.ca_pem);
std::fs::create_dir_all(data.path().join("prod")).unwrap();
let config = format!(
r#"(
repositories: [
(ron_name: "prod", ron_repo_type: "rust", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "default")),
ron_data_dir: Some("{prod}")),
],
// Several boxes share this test process, and the egress guard is per process: the box
// that names no upstream says `network: Online` so its neighbours may dial (B2).
network: Online,
exposed_endpoints: [ (ron_name: "default", ron_url: "127.0.0.1:{grpc}") ],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "file", ron_path: "{root}") ],
audit: ( dir: "{root}/holger-audit", enabled: true ),
mirror: Some((
ron_state_dir: "holger-mirror",
ron_sources: [
(
ron_name: "cloud",
ron_kind: Holger,
ron_endpoint: "{endpoint}",
ron_credential: Mtls( ron_client_cert: "{c}", ron_client_key: "{k}", ron_ca_cert: Some("{ca}") ),
ron_every_secs: 300,
ron_repositories: [ ( ron_from: "prod", ron_to: "prod" ) ],
),
],
)),
)"#,
prod = data.path().join("prod").display(),
root = data.path().display(),
grpc = hold_port(),
endpoint = cloud.endpoint,
);
let mut cfg = tempfile::NamedTempFile::new().unwrap();
cfg.write_all(config.as_bytes()).unwrap();
let mut onprem_box: Holger = read_ron_config(cfg.path()).unwrap();
onprem_box.instantiate_backends().unwrap();
wire_holger(&mut onprem_box).unwrap();
assert_eq!(
onprem_box.peers_root(),
data.path(),
"the on-prem box keeps its records in its own data root"
);
// The downstream's half of the pairing: the cloud's key, on source `cloud`.
peers::pair_upstream(
&onprem_box.peers_root(),
"cloud",
"cloud-holger",
cloud.server_cert_pem.as_bytes(),
"test",
1,
)
.expect("the upstream pairing is recorded");
let plans = holger_server_lib::mirror_cycle::plan(&onprem_box, Some(data.path())).unwrap();
assert!(plans[0].pinned.is_some(), "the source is paired");
let _timers = holger_server_lib::mirror_cycle::spawn_timers(plans);
let admin = channel(&cloud, &pki, None).await;
let mut seen = None;
for _ in 0..200 {
let rows = GrowServiceClient::new(admin.clone())
.downstreams(grow::DownstreamsRequest {})
.await
.unwrap()
.into_inner();
let row = rows
.downstreams
.into_iter()
.find(|d| d.name == "onprem-1")
.unwrap();
if row
.repositories
.first()
.is_some_and(|r| r.last_complete_unix_ms > 0)
{
seen = Some((row, rows.head_seq));
break;
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
let (row, head) = seen.expect("the mirror reported a completed cycle to its upstream");
let prod = &row.repositories[0];
assert_eq!(
(
prod.repository.as_str(),
prod.state.as_str(),
prod.lag_entries
),
("prod", "ok", 0),
"{row:?}"
);
assert_eq!(prod.applied_seq, head);
assert_eq!(
(row.state.as_str(), row.stalled, row.every_secs),
("ok", false, 300)
);
assert!(
row.requests_today > 0,
"the cycle's pulls are counted on the same row"
);
assert!(
cloud
.data_root
.join(peers::PEERS_DIR)
.join("sync-onprem-1.ron")
.is_file(),
"the last sync point is kept under the upstream's data root"
);
// The bytes landed too.
let id = ArtifactId {
namespace: None,
name: "thing".into(),
version: "1.0.0".into(),
};
assert_eq!(
onprem_box
.backend_named("prod")
.unwrap()
.fetch(&id)
.unwrap()
.as_deref(),
Some(&b"BYTES"[..])
);
}