server/cli/tests/one_way_downstream.rs
//! **The one-way downstream** (lane HOL-CDS, owner 2026-10-30: "safer if we have to add
//! code for fetching").
//!
//! * The UPSTREAM's `export: (push_to: Some(<dir>), every: <interval>)` writes each new
//! signed bundle of its AVAILABLE set into the diode's send folder, atomically (a hidden
//! `.part` file, then a rename): `holger export-push` runs one push, the started server
//! runs it every interval. `holger export-push --trust-policy-out` writes the pinned trust
//! policy the secure side verifies those bundles with.
//! * The DOWNSTREAM is the default `holger-downstream` build (`--no-default-features
//! --features downstream`, no fetch code in it). Its packages come in ONLY through its
//! inbox. The first import is an admin's action on this side (`holger-downstream bundle
//! allow-first-import`, for one vendor, product and sequence): until then the inbox refuses
//! the bundle (`holger-import-9`), a `FIRST-IMPORT` marker in the inbox is refused
//! (`holger-import-10`) and so is a caller's `first_import` flag. Then it verifies the
//! release signature, every hash, the attestations and the floor, and serves the wheel to
//! pip with its `domain:` label in `X-Holger-Domain`.
//! * `boundary: OneWay` (the default) refuses to start on a `mirror:` block or an `export:`
//! door, each by its line (`holger-boundary-1`); `boundary: Fetch` is refused by a build
//! without the fetch code (`holger-boundary-2`).
//!
//! The fetching downstream (`--features fetch`, `boundary: Fetch`) is
//! `downstream_serves_the_build_systems.rs`'s pull test.
//!
//! The downstream binary: `HOLGER_DOWNSTREAM_BIN`, or `holger-downstream` beside the test's
//! `holger`, built ONE-WAY (`cargo build -p holger --no-default-features --features
//! downstream --bin holger-downstream` into the same target directory first).
use std::collections::BTreeMap;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
use std::sync::Arc;
use std::time::{Duration, Instant};
use server_lib::promote::intake::{ArtifactState, IntakeChain, TransitionRequest};
use server_lib::scanner::{
ScanOutcome, ScanPipeline, ScanSubject, Scanner, ScannerKind, ScannerSet,
};
use server_lib::ArtifactId;
use traits::test_support::hold_port;
const BIN: &str = env!("CARGO_BIN_EXE_holger");
const WHEEL: &str = "demo_pkg-1.0.0-py3-none-any.whl";
const WHEEL2: &str = "demo_pkg-1.1.0-py3-none-any.whl";
const DOMAIN: &str = "onprem-lab";
fn text(o: &Output) -> String {
format!(
"{}{}",
String::from_utf8_lossy(&o.stdout),
String::from_utf8_lossy(&o.stderr)
)
}
fn sha256(b: &[u8]) -> String {
update_bundle::digest::sha256_hex(b)
}
fn wheel(version: &str) -> Vec<u8> {
let meta = format!(
"Metadata-Version: 2.1\nName: demo-pkg\nVersion: {version}\nLicense-Expression: MIT\n"
);
zip_entry::fixture::zip(&[
("demo_pkg/__init__.py", format!("__version__ = \"{version}\"\n").as_bytes()),
(&format!("demo_pkg-{version}.dist-info/METADATA"), meta.as_bytes()),
(&format!("demo_pkg-{version}.dist-info/WHEEL"), b"Wheel-Version: 1.0\nGenerator: holger-test\nRoot-Is-Purelib: true\nTag: py3-none-any\n"),
(&format!("demo_pkg-{version}.dist-info/RECORD"), b""),
])
}
struct CleanVulnerabilityScan;
impl Scanner for CleanVulnerabilityScan {
fn id(&self) -> &str {
"upstream-vulnerability-scan"
}
fn version(&self) -> String {
"test 1".into()
}
fn kind(&self) -> ScannerKind {
ScannerKind::Vulnerability
}
fn scan(&self, subject: &ScanSubject<'_>) -> anyhow::Result<ScanOutcome> {
Ok(ScanOutcome::pass(format!(
"{}: no advisory",
subject.file_name
)))
}
}
struct SbomOfTheUpstream;
impl Scanner for SbomOfTheUpstream {
fn id(&self) -> &str {
"upstream-sbom"
}
fn version(&self) -> String {
"test 1".into()
}
fn kind(&self) -> ScannerKind {
ScannerKind::Sbom
}
fn scan(&self, subject: &ScanSubject<'_>) -> anyhow::Result<ScanOutcome> {
let doc = serde_json::json!({
"bomFormat": "CycloneDX", "specVersion": "1.5", "version": 1,
"metadata": { "component": { "type": "library", "name": subject.file_name } },
"components": []
});
Ok(ScanOutcome::pass(format!("{}: an SBOM", subject.file_name)).with_document(doc))
}
}
fn intake_repos(d: &Path, extra: &str) -> String {
let repo = |name: &str| {
format!(
r#"(ron_name: "{name}", ron_repo_type: "raw", ron_upstreams: [], ron_in: None,
ron_out: Some((ron_storage_endpoint: "storage", ron_exposed_endpoint: "door")),
ron_data_dir: Some("{}/{name}")),"#,
d.display()
)
};
format!(
"{} {} {} {} {extra}",
repo("incoming"),
repo("quarantine"),
repo("approved"),
repo("production")
)
}
fn ladder(rules: &str, trust_policy: Option<&Path>) -> String {
let tp = trust_policy
.map(|p| format!(r#"trust_policy: Some("{}"),"#, p.display()))
.unwrap_or_default();
format!(
r#"promotion: Some((
ladder_name: "intake",
stages: [
(name: "incoming", repo: "incoming"),
(name: "quarantine", repo: "quarantine"),
(name: "approved", repo: "approved"),
(name: "production", repo: "production"),
],
intake: Some(({rules}
incoming: "incoming", quarantine: "quarantine",
approved: "approved", production: "production",
{tp}
)),
)),"#
)
}
struct Upstream {
dir: tempfile::TempDir,
config: PathBuf,
pki: bench_scenarios::secure_update::AttestPki,
send: PathBuf,
http: u16,
}
/// The upstream with its push into the diode's send folder (`export:`), and one wheel walked
/// to AVAILABLE with signed scan evidence and an approval by a person who did not import it.
fn upstream(every: &str) -> Upstream {
let dir = tempfile::tempdir().unwrap();
let d = dir.path().join("up");
let send = dir.path().join("diode-send");
std::fs::create_dir_all(&send).unwrap();
let pki = bench_scenarios::secure_update::mint_attestation_pki(&d.join("pki"), "upstream-scan")
.unwrap();
let (grpc, http) = (hold_port(), hold_port());
let ron = format!(
r#"(
repositories: [ {repos} ],
exposed_endpoints: [ (ron_name: "door", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}")) ],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{d}") ],
audit: (dir: "{d}/audit", enabled: true),
attestation: Some((
trust_roots: ["{root}"],
signer_key: Some("{key}"),
signer_cert: Some("{cert}"),
signer_alg: Some("ecdsa-p256"),
)),
{ladder}
export: Some((push_to: Some("{send}"), every: "{every}", vendor: "upstream", product: "on-prem")),
)"#,
repos = intake_repos(&d, ""),
d = d.display(),
root = pki.root_der.display(),
key = pki.key_pkcs8.display(),
cert = pki.cert_der.display(),
ladder = ladder("rules: (sandbox: off),", None),
send = send.display(),
);
let config = dir.path().join("upstream.ron");
std::fs::write(&config, ron).unwrap();
let up = Upstream {
dir,
config,
pki,
send,
http,
};
up.approve(WHEEL, &wheel("1.0.0"));
up
}
impl Upstream {
/// Walk one package to AVAILABLE through the upstream's intake chain.
fn approve(&self, name: &str, bytes: &[u8]) {
let mut h = server_lib::read_ron_config(&self.config).unwrap();
h.instantiate_backends().unwrap();
let audit = h.audit_log().unwrap();
let settings = h
.promotion
.as_ref()
.and_then(|p| p.intake.clone())
.expect("the intake block");
let incoming = h.get_repo("incoming").unwrap().clone();
let pipeline = ScanPipeline::new(
ScannerSet::new()
.with(Arc::new(CleanVulnerabilityScan))
.with(Arc::new(SbomOfTheUpstream))
.require(ScannerKind::Vulnerability)
.require(ScannerKind::Sbom),
h.scan_evidence_store(audit.clone()),
)
.with_attestation(
h.attestation_store().unwrap().expect("a store"),
h.attestation_signer().unwrap().expect("a signer"),
);
let chain = IntakeChain::new(
settings,
h.promotion_engine().unwrap(),
h.data_root().join("holger-prod-store"),
audit,
incoming.clone(),
)
.with_scanner(Arc::new(pipeline));
let pid = ArtifactId {
namespace: None,
name: name.into(),
version: String::new(),
};
chain
.ingest("ida", "test", "incoming", &pid, bytes, |a| {
a.store(&*incoming, bytes)
})
.unwrap_or_else(|e| panic!("ingest {name}: {e}"));
for (to, who) in [
(ArtifactState::Scanned, "ida"),
(ArtifactState::PolicyChecked, "ida"),
(ArtifactState::Approved, "bob"),
(ArtifactState::Sealed, "bob"),
(ArtifactState::Promoted, "bob"),
(ArtifactState::Available, "bob"),
] {
chain
.advance(TransitionRequest::new(pid.clone(), to, who))
.unwrap_or_else(|e| panic!("{name} → {to}: {e}"));
}
}
fn push(&self) -> Output {
Command::new(BIN)
.args(["export-push", "--config"])
.arg(&self.config)
.output()
.unwrap()
}
fn trust_policy(&self) -> PathBuf {
let p = self.dir.path().join("pushed.policy.ron");
let out = Command::new(BIN)
.args(["export-push", "--config"])
.arg(&self.config)
.arg("--trust-policy-out")
.arg(&p)
.output()
.unwrap();
assert!(
out.status.success(),
"the trust policy of the pushes: {}",
text(&out)
);
p
}
/// The bundles in the send folder, by name.
fn sent(&self) -> Vec<String> {
let mut v: Vec<String> = std::fs::read_dir(&self.send)
.unwrap()
.flatten()
.map(|e| e.file_name().to_string_lossy().into_owned())
.collect();
v.sort();
v
}
}
// ── the downstream ──────────────────────────────────────────────────────────
fn downstream_bin() -> PathBuf {
if let Ok(p) = std::env::var("HOLGER_DOWNSTREAM_BIN") {
return PathBuf::from(p);
}
let beside = Path::new(BIN).with_file_name("holger-downstream");
assert!(
beside.exists(),
"no downstream binary: build it first with `cargo build --manifest-path holger-downstream/Cargo.toml \
--features sandbox-off --bin holger-downstream` (into this target directory), or set HOLGER_DOWNSTREAM_BIN"
);
beside
}
#[path = "support/connect_trap.rs"]
mod connect_trap;
use connect_trap::{dialled, trapped};
struct Downstream {
dir: PathBuf,
config: PathBuf,
http: u16,
child: Option<Child>,
}
impl Drop for Downstream {
fn drop(&mut self) {
if let Some(mut c) = self.child.take() {
let _ = c.kill();
let _ = c.wait();
}
}
}
/// A one-way downstream: the intake chain (no scanner), the upstream's attestation roots,
/// the pushes' trust policy, its `domain:` label and `extra` top-level lines.
fn downstream(up: &Upstream, name: &str, trust_policy: Option<&Path>, extra: &str) -> Downstream {
let dir = up.dir.path().join(name);
std::fs::create_dir_all(&dir).unwrap();
let (http, grpc) = (hold_port(), hold_port());
let ron = format!(
r#"(
domain: "{DOMAIN}",
repositories: [ {repos} ],
exposed_endpoints: [ (ron_name: "door", ron_url: "127.0.0.1:{grpc}", ron_http_url: Some("127.0.0.1:{http}")) ],
storage_endpoints: [ (ron_name: "storage", ron_storage_type: "znippy", ron_path: "{d}") ],
audit: (dir: "{d}/audit", enabled: true),
attestation: Some((
trust_roots: ["{root}"],
enrolled_signers: ["{cert}"],
)),
trust: (first_import: Admin),
{ladder}
{extra}
)"#,
repos = intake_repos(&dir, ""),
d = dir.display(),
root = up.pki.root_der.display(),
cert = up.pki.cert_der.display(),
// The binary under test is the toolchain image's: no container engine, so its
// verify-bundle jobs run in-process (`holger-downstream --features sandbox-off`).
ladder = ladder("rules: (sandbox: off),", trust_policy),
);
let config = up.dir.path().join(format!("{name}.ron"));
std::fs::write(&config, ron).unwrap();
Downstream {
dir,
config,
http,
child: None,
}
}
impl Downstream {
fn connects(&self) -> PathBuf {
self.dir.join("connects.log")
}
fn inbox(&self) -> PathBuf {
self.dir.join("bundle.inbox")
}
fn cli(&self, args: &[&std::ffi::OsStr]) -> Output {
trapped(&mut Command::new(downstream_bin()), &self.connects())
.args(args)
.output()
.unwrap()
}
fn start(&mut self) {
let log = self.dir.join("server.log");
let log_file = std::fs::File::create(&log).unwrap();
let mut c = Command::new(downstream_bin());
trapped(&mut c, &self.connects());
let child = c
.args(["start", "--config"])
.arg(&self.config)
.stdout(Stdio::from(log_file.try_clone().unwrap()))
.stderr(Stdio::from(log_file))
.spawn()
.unwrap();
let t = Instant::now();
while std::net::TcpStream::connect(("127.0.0.1", self.http)).is_err() {
assert!(
t.elapsed() < Duration::from_secs(60),
"the downstream never listened:\n{}",
self.log()
);
std::thread::sleep(Duration::from_millis(100));
}
self.child = Some(child);
}
fn stop(&mut self) {
if let Some(mut c) = self.child.take() {
let _ = c.kill();
let _ = c.wait();
}
}
fn log(&self) -> String {
std::fs::read_to_string(self.dir.join("server.log")).unwrap_or_default()
}
/// The start that must not happen: its exit and its words.
fn start_refused(&self) -> Output {
let mut c = Command::new(downstream_bin());
let mut child = c
.args(["start", "--config"])
.arg(&self.config)
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap();
let t = Instant::now();
loop {
if child.try_wait().unwrap().is_some() {
return child.wait_with_output().unwrap();
}
if t.elapsed() > Duration::from_secs(30) {
let _ = child.kill();
let out = child.wait_with_output().unwrap();
panic!(
"the downstream started although it must refuse:\n{}",
text(&out)
);
}
std::thread::sleep(Duration::from_millis(100));
}
}
/// Wait until the inbox has moved a file whose name ends in `name` into `sub`.
/// The `.why.txt` beside a refused file, once the door has written it whole: the door moves
/// the file into `refused/` first and writes its why after (CI-RED, flaky under load: a read
/// right after the move found no file yet).
fn wait_why(&self, refused: &Path) -> String {
let why = refused.with_extension("why.txt");
let t = Instant::now();
loop {
if let Ok(s) = std::fs::read_to_string(&why) {
if s.ends_with('\n') {
return s;
}
}
assert!(
t.elapsed() < Duration::from_secs(60),
"{} never got its why.txt:\n{}",
refused.display(),
self.log()
);
std::thread::sleep(Duration::from_millis(100));
}
}
fn wait_moved(&self, sub: &str, name: &str) -> PathBuf {
let t = Instant::now();
loop {
if let Ok(d) = std::fs::read_dir(self.inbox().join(sub)) {
if let Some(e) = d
.flatten()
.find(|e| e.file_name().to_string_lossy().ends_with(name))
{
return e.path();
}
}
assert!(
t.elapsed() < Duration::from_secs(60),
"the inbox never moved {name} into {sub}/:\n{}",
self.log()
);
std::thread::sleep(Duration::from_millis(200));
}
}
}
struct Reply {
status: u16,
headers: BTreeMap<String, String>,
body: Vec<u8>,
}
impl Reply {
fn text(&self) -> String {
String::from_utf8_lossy(&self.body).to_string()
}
}
fn request(port: u16, method: &str, path: &str, body: &[u8]) -> Reply {
let mut s = std::net::TcpStream::connect(("127.0.0.1", port)).unwrap();
let head = format!("{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\nContent-Length: {}\r\n\r\n", body.len());
s.write_all(head.as_bytes()).unwrap();
s.write_all(body).unwrap();
let mut raw = Vec::new();
s.read_to_end(&mut raw).unwrap();
let split = raw
.windows(4)
.position(|w| w == b"\r\n\r\n")
.expect("an HTTP answer");
let head = String::from_utf8_lossy(&raw[..split]).to_string();
let mut lines = head.lines();
let status = lines
.next()
.unwrap()
.split_whitespace()
.nth(1)
.unwrap()
.parse()
.unwrap();
let headers = lines
.filter_map(|l| l.split_once(':'))
.map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string()))
.collect::<BTreeMap<_, _>>();
let mut body = raw[split + 4..].to_vec();
if headers
.get("transfer-encoding")
.is_some_and(|v| v.eq_ignore_ascii_case("chunked"))
{
let mut out = Vec::new();
let mut rest = &body[..];
loop {
let nl = rest.windows(2).position(|w| w == b"\r\n").unwrap();
let n = usize::from_str_radix(String::from_utf8_lossy(&rest[..nl]).trim(), 16).unwrap();
if n == 0 {
break;
}
out.extend_from_slice(&rest[nl + 2..nl + 2 + n]);
rest = &rest[nl + 2 + n + 2..];
}
body = out;
}
Reply {
status,
headers,
body,
}
}
/// The pushed bundle's sequence, from the push's own line `(vendor V product P sequence N, …)`.
fn pushed_sequence(out: &Output) -> String {
let t = text(out);
t.split("sequence ")
.nth(1)
.and_then(|s| s.split(|c: char| !c.is_ascii_digit()).next())
.map(String::from)
.unwrap_or_else(|| panic!("the push names no sequence: {t}"))
}
/// ★ **The diode, end to end**: the upstream pushes, the one-way downstream imports through
/// its inbox after the admin's first-import authorization, and serves pip.
#[test]
fn the_upstream_pushes_into_the_diode_folder_and_a_one_way_downstream_imports_and_serves_it() {
let up = upstream("1h");
let policy = up.trust_policy();
// One push: one bundle, written atomically; nothing half-written is left.
let out = up.push();
assert!(out.status.success(), "the push: {}", text(&out));
assert!(
text(&out).contains("vendor upstream product on-prem"),
"the push names vendor and product: {}",
text(&out)
);
let seq = pushed_sequence(&out);
let sent = up.sent();
assert_eq!(sent.len(), 1, "one bundle in the send folder: {sent:?}");
assert!(
sent[0].ends_with(".holger.znippy") && !sent[0].starts_with('.'),
"{sent:?}"
);
// Nothing new: no second bundle.
let again = up.push();
assert!(again.status.success(), "{}", text(&again));
assert!(text(&again).contains("nothing new"), "{}", text(&again));
assert_eq!(up.sent().len(), 1, "a push with nothing new writes nothing");
let bundle = up.send.join(&sent[0]);
let mut ds = downstream(&up, "onprem", Some(&policy), "");
std::fs::create_dir_all(ds.inbox()).unwrap();
ds.start();
assert!(
ds.log().contains(&format!("domain: {DOMAIN}")),
"the start names its domain:\n{}",
ds.log()
);
assert!(
ds.log().contains("boundary: OneWay"),
"the start names its boundary:\n{}",
ds.log()
);
// No floor and no admin authorization: the inbox refuses the bundle, by name.
std::fs::copy(&bundle, ds.inbox().join("first.holger.znippy")).unwrap();
let refused = ds.wait_moved("refused", "first.holger.znippy");
let why = ds.wait_why(&refused);
assert!(
why.contains("[holger-import-9]") && why.contains("allow-first-import"),
"{why}"
);
// A FIRST-IMPORT marker in the inbox is no authorization: it is refused itself.
std::fs::write(ds.inbox().join("FIRST-IMPORT"), b"").unwrap();
std::fs::copy(&bundle, ds.inbox().join("second.holger.znippy")).unwrap();
let refused = ds.wait_moved("refused", "second.holger.znippy");
let why = ds.wait_why(&refused);
assert!(
why.contains("[holger-import-9]"),
"a marker establishes nothing: {why}"
);
let marker = ds.wait_moved("refused", "FIRST-IMPORT");
let why = ds.wait_why(&marker);
assert!(why.contains("[holger-import-10]"), "{why}");
// A caller's first_import flag is gone, on the CLI and on the HTTP door.
let flag = ds.cli(&[
"bundle".as_ref(),
"import".as_ref(),
"--config".as_ref(),
ds.config.as_os_str(),
"--bundle".as_ref(),
bundle.as_os_str(),
"--first-import".as_ref(),
]);
assert!(!flag.status.success(), "{}", text(&flag));
assert!(
text(&flag).contains("--first-import"),
"the CLI names the flag it does not take: {}",
text(&flag)
);
let http_flag = request(
ds.http,
"POST",
"/-/bundle/import?first_import=true",
&std::fs::read(&bundle).unwrap(),
);
assert_eq!(http_flag.status, 400, "{}", http_flag.text());
assert!(
http_flag.text().contains("[holger-import-10]"),
"{}",
http_flag.text()
);
// ★ The admin's action on the secure side: one vendor, product and sequence.
let wrong = ds.cli(
&[
"bundle",
"allow-first-import",
"--config",
ds.config.to_str().unwrap(),
"--vendor",
"upstream",
"--product",
"on-prem",
"--sequence",
"0",
]
.map(std::ffi::OsStr::new),
);
assert!(
!wrong.status.success() && text(&wrong).contains("[holger-import-11]"),
"sequence 0 is refused: {}",
text(&wrong)
);
let grant = ds.cli(
&[
"bundle",
"allow-first-import",
"--config",
ds.config.to_str().unwrap(),
"--vendor",
"upstream",
"--product",
"on-prem",
"--sequence",
&seq,
]
.map(std::ffi::OsStr::new),
);
assert!(grant.status.success(), "{}", text(&grant));
assert!(
text(&grant).contains(&format!("domain: {DOMAIN}")),
"the CLI names its domain: {}",
text(&grant)
);
std::fs::copy(&bundle, ds.inbox().join("third.holger.znippy")).unwrap();
ds.wait_moved("archive", "third.holger.znippy");
// Served to pip, the domain on the answer.
let page = request(ds.http, "GET", "/production/simple/demo-pkg/", &[]);
assert_eq!(page.status, 200, "{}\n{}", page.text(), ds.log());
assert!(
page.text()
.contains(&format!("#sha256={}", sha256(&wheel("1.0.0")))),
"{}",
page.text()
);
assert_eq!(
page.headers.get("x-holger-domain").map(String::as_str),
Some(DOMAIN),
"{:?}",
page.headers
);
assert_eq!(
request(ds.http, "GET", &format!("/production/{WHEEL}"), &[]).body,
wheel("1.0.0")
);
let miss = request(ds.http, "GET", "/no/such/thing", &[]);
assert_eq!(
miss.headers.get("x-holger-domain").map(String::as_str),
Some(DOMAIN),
"every answer carries the domain"
);
// The audit names who authorized the first import.
let audit: String = std::fs::read_dir(ds.dir.join("audit"))
.unwrap()
.flatten()
.map(|e| std::fs::read_to_string(e.path()).unwrap_or_default())
.collect();
assert!(
audit.contains("first-import authorized"),
"the authorization is audited"
);
// ★ A second approved package: the next push carries only it, and the floor (now standing)
// takes it without any admin step.
up.approve(WHEEL2, &wheel("1.1.0"));
let next = up.push();
assert!(next.status.success(), "{}", text(&next));
let sent = up.sent();
assert_eq!(sent.len(), 2, "{sent:?}");
let newest = sent
.iter()
.find(|n| up.send.join(n) != bundle)
.cloned()
.unwrap();
std::fs::copy(
up.send.join(&newest),
ds.inbox().join("fourth.holger.znippy"),
)
.unwrap();
ds.wait_moved("archive", "fourth.holger.znippy");
let page = request(ds.http, "GET", "/production/simple/demo-pkg/", &[]);
assert!(
page.text()
.contains(&format!("#sha256={}", sha256(&wheel("1.1.0")))),
"{}",
page.text()
);
// ★ One-way at run time: nothing was dialled.
ds.stop();
let dials = dialled(&ds.connects());
assert!(
dials.is_empty(),
"a one-way downstream dials nothing, yet it did:\n{dials:#?}"
);
}
/// ★ The started upstream pushes on its own, every `every`.
#[test]
fn the_started_upstream_pushes_every_interval() {
let up = upstream("1s");
let log = up.dir.path().join("upstream.log");
let f = std::fs::File::create(&log).unwrap();
let mut server = Command::new(BIN)
.args(["start", "--config"])
.arg(&up.config)
.stdout(Stdio::from(f.try_clone().unwrap()))
.stderr(Stdio::from(f))
.spawn()
.unwrap();
let t = Instant::now();
while std::net::TcpStream::connect(("127.0.0.1", up.http)).is_err() {
assert!(
t.elapsed() < Duration::from_secs(60),
"the upstream never listened:\n{}",
std::fs::read_to_string(&log).unwrap_or_default()
);
std::thread::sleep(Duration::from_millis(100));
}
loop {
let sent: Vec<String> = up
.sent()
.into_iter()
.filter(|n| n.ends_with(".holger.znippy") && !n.starts_with('.'))
.collect();
if !sent.is_empty() {
break;
}
assert!(
t.elapsed() < Duration::from_secs(60),
"the started upstream never pushed:\n{}",
std::fs::read_to_string(&log).unwrap_or_default()
);
std::thread::sleep(Duration::from_millis(250));
}
let _ = server.kill();
let _ = server.wait();
}
/// ★ `boundary: OneWay` refuses a way out, each by its line; `Fetch` needs the fetch build.
#[test]
fn a_one_way_downstream_refuses_a_mirror_or_an_export_door_by_line_and_code() {
let up = upstream("1h");
let ds = downstream(
&up,
"mirrored",
None,
r#" mirror: Some((ron_state_dir: "/tmp/never", ron_sources: [ (ron_name: "upstream", ron_kind: Holger, ron_endpoint: "http://127.0.0.1:1",
ron_credential: Anonymous, ron_every_secs: 1, ron_retry_secs: 1,
ron_repositories: [ (ron_from: "bundles", ron_to: "incoming", ron_unproven: TrustSize) ]) ])),
export: Some((push_to: Some("/tmp/never-send"), every: "1m")),"#,
);
let ron = std::fs::read_to_string(&ds.config).unwrap();
let line_of = |key: &str| {
ron.lines()
.position(|l| l.trim_start().starts_with(key))
.unwrap()
+ 1
};
let out = ds.start_refused();
let t = text(&out);
assert!(!out.status.success(), "{t}");
assert!(t.contains("[holger-boundary-1]"), "{t}");
assert!(
t.contains(&format!("line {}: the `mirror:` block", line_of("mirror:"))),
"the refusal names the mirror's line: {t}"
);
assert!(
t.contains(&format!("line {}: the `export:` block", line_of("export:"))),
"the refusal names the export door's line: {t}"
);
let fetch = downstream(&up, "fetching", None, " boundary: Fetch,");
let out = fetch.start_refused();
let t = text(&out);
assert!(
!out.status.success() && t.contains("[holger-boundary-2]"),
"a one-way build refuses `boundary: Fetch`: {t}"
);
assert!(t.contains("--features fetch"), "{t}");
}
/// PEM of a DER certificate (a medium's staged parameters are text).
fn pem(der: &[u8]) -> String {
use base64::Engine as _;
let b = base64::engine::general_purpose::STANDARD.encode(der);
let lines: Vec<&str> = b
.as_bytes()
.chunks(64)
.map(|c| std::str::from_utf8(c).unwrap())
.collect();
format!(
"-----BEGIN CERTIFICATE-----\n{}\n-----END CERTIFICATE-----\n",
lines.join("\n")
)
}
/// **The KVM proof's fixture** (lane HOL-CDS-5, `cargo xtask downstream-oneway-proof`): an
/// upstream's pushed bundle with one approved wheel, the pinned trust policy of its pushes, the
/// upstream's attestation root and evidence signer (PEM, what the one-way medium carries) and
/// `fixture.json` (vendor, product, sequence, the wheel and its sha256). Written into
/// `HOLGER_ONEWAY_FIXTURE`.
#[test]
#[ignore = "writes the KVM proof's fixture: HOLGER_ONEWAY_FIXTURE=<dir> cargo test --test one_way_downstream -- --ignored write_the_one_way_appliance_fixture"]
fn write_the_one_way_appliance_fixture() {
let out = PathBuf::from(
std::env::var("HOLGER_ONEWAY_FIXTURE")
.expect("HOLGER_ONEWAY_FIXTURE names the fixture's directory"),
);
std::fs::create_dir_all(&out).unwrap();
let up = upstream("1h");
let policy = up.trust_policy();
let push = up.push();
assert!(push.status.success(), "{}", text(&push));
let seq = pushed_sequence(&push);
let sent = up.sent();
assert_eq!(sent.len(), 1, "{sent:?}");
std::fs::copy(up.send.join(&sent[0]), out.join(&sent[0])).unwrap();
std::fs::copy(&policy, out.join("trust-policy.ron")).unwrap();
std::fs::write(
out.join("upstream-root.pem"),
pem(&std::fs::read(&up.pki.root_der).unwrap()),
)
.unwrap();
std::fs::write(
out.join("upstream-signer.pem"),
pem(&std::fs::read(&up.pki.cert_der).unwrap()),
)
.unwrap();
let fixture = serde_json::json!({
"bundle": sent[0], "vendor": "upstream", "product": "on-prem", "sequence": seq.parse::<u64>().unwrap(),
"project": "demo-pkg", "wheel": WHEEL, "wheel_sha256": sha256(&wheel("1.0.0")),
});
std::fs::write(
out.join("fixture.json"),
serde_json::to_vec_pretty(&fixture).unwrap(),
)
.unwrap();
println!("FIXTURE {}", out.display());
}