server/lib/src/sync/reports.rs
//! **A sync downstream's mirror report, counted by its enrolled name** (lane OX-WGDS2).
//!
//! A downstream's mirror tells its upstream how far it has synced after every cycle
//! (`holger.grow.v1.GrowService/DownstreamReport`). A PAIRED downstream is named by its
//! mutual-TLS certificate (`crate::peers`). A SYNC downstream, the RON's or the console's
//! (`via: Front` or `via: Wireguard`), signs the report's text with its sync key and sends
//! the proof beside it ([`REPORT_HEADER`]); main opens it under its LIVE downstreams (a
//! revoked one is refused), checks the name the front put on a WireGuard caller, and keeps
//! the report as `holger-sync/reports/<name>.ron`. The sync board and the estate board's
//! summary line read it: "synced N ago", "N behind", instead of "never".
//!
//! ```text
//! downstream mirror ── DownstreamReport + x-holger-sync-report ──► front (public door, or
//! holger-ds + x-holger-downstream)
//! ──► main: heard() → reports/<name>.ron
//! board(): last_report_at, entries_behind
//! ```
use std::collections::BTreeMap;
use base64::Engine as _;
use holger_sync::ledger::DownstreamRow;
use super::downstreams::{read_ron_dir, write_ron};
use super::*;
use crate::grpc::holger_grow_proto::DownstreamReportRequest;
/// The gRPC metadata the proof travels in (base64 of the signed envelope).
pub const REPORT_HEADER: &str = crate::mirror_cycle::REPORT_HEADER;
/// The kept reports, under [`STATE_DIR`].
pub const REPORTS_DIR: &str = "reports";
/// **What main keeps of a sync downstream's last report.**
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct Heard {
pub name: String,
/// Its mirror's declared cycle, seconds.
pub every_secs: u32,
/// When main heard it (Unix ms).
pub last_seen_ms: i64,
/// How many of main's writes it had not applied at that report (the most behind
/// repository; another boot of main's counts as all of them).
pub behind: u64,
/// Per repository of main: `(applied boot, applied write sequence, last complete cycle ms)`.
pub repos: BTreeMap<String, (u64, u64, i64)>,
/// What it said of its own RON (HOL-FLOW-SMART); a report before it says nothing.
#[serde(default)]
pub summary: Option<crate::flow::RonSummary>,
}
/// Pure: **the report's text**, the bytes its proof signs: the cycle, then one line per
/// repository, `repository \t boot \t sequence \t last complete ms`. The same text in every
/// holger version (HOL-SYNC-SKEW): an upstream and its downstreams upgrade at different
/// times, so nothing a later version adds to the report goes into it; the RON summary has
/// its own proof over [`summary_text`].
pub fn report_text(req: &DownstreamReportRequest) -> String {
let mut t = format!(
"holger.downstream-report-text/v1\nevery {}\n",
req.every_secs
);
for p in &req.points {
t.push_str(&format!(
"{}\t{}\t{}\t{}\n",
p.repository, p.applied_boot, p.applied_seq, p.last_complete_unix_ms
));
}
t
}
/// Pure: **the RON summary's text**, the bytes its own proof (`summary_proof`) signs
/// (HOL-SYNC-SKEW): its schema line, the whole [`report_text`] (so the summary belongs to
/// this one report), then one `summary \t field \t value` line per field. `None` when the
/// report carries no summary.
pub fn summary_text(req: &DownstreamReportRequest) -> Option<String> {
let s = req.summary.as_ref()?;
let one_line = |v: &str| v.replace(['\n', '\t'], " ");
let mut t = format!("holger.downstream-summary-text/v1\n{}", report_text(req));
for (field, value) in [
("boundary", s.boundary.clone()),
("repositories", s.repositories.join(",")),
("inbox", s.inbox.clone()),
("first_import", s.first_import.clone()),
("domain", s.domain.clone()),
("downstreams", s.downstreams.to_string()),
] {
t.push_str(&format!("summary\t{field}\t{}\n", one_line(&value)));
}
Some(t)
}
/// Pure: **the console address's text**, the bytes its own proof (`console_proof`) signs
/// (HOL-FLOW-FRONT): its schema line, the whole [`report_text`], then the address. `None` when
/// the report's summary names no console. Apart from [`summary_text`], so an upstream that
/// checks the summary as holger 3.1.0 signs it still opens it.
pub fn console_text(req: &DownstreamReportRequest) -> Option<String> {
let url = req
.summary
.as_ref()
.map(|s| s.console_url.trim())
.filter(|u| !u.is_empty())?;
Some(format!(
"holger.downstream-console-text/v1\n{}console\t{}\n",
report_text(req),
url.replace(['\n', '\t'], " ")
))
}
/// Pure: **how many of main's writes `(boot, head)` the report had not applied**, the most
/// behind of its repositories. A point of another boot is behind by all of them.
pub fn behind(req: &DownstreamReportRequest, boot: u64, head: u64) -> u64 {
req.points
.iter()
.map(|p| {
if p.applied_boot == boot {
head.saturating_sub(p.applied_seq)
} else {
head
}
})
.max()
.unwrap_or(0)
}
/// Pure: **a board row with its mirror's report** (lane OX-WGDS2). The report says when
/// it last synced and how far behind it was. A downstream that has never acknowledged a
/// generation (its acks come by stick, or not at all) is judged by its mirror: `current`
/// or `behind`, never `never`; its generations are not its lag. An alarm stays an alarm.
pub fn with_report(row: &mut DownstreamRow, heard: &Heard) {
row.last_report_at = heard.last_seen_ms / 1000;
row.entries_behind = heard.behind;
if row.acks == 0 && !row.alarm {
(row.lag_generations, row.lag_secs) = (0, 0);
if heard.behind > 0 {
row.state = "behind".into();
row.headline = format!(
"its mirror reported {} of this upstream's writes not applied yet",
heard.behind
);
} else {
row.state = "current".into();
row.headline = "its mirror reported it holds every write of this upstream".into();
}
}
}
/// The live hubs of this process, so a mirror planned before its server's intake chain
/// signs with the hub that chain gets ([`proof_for`]).
static LIVE: Mutex<Vec<Weak<SyncHub>>> = Mutex::new(Vec::new());
/// **Keep `hub` as this data root's live hub** (the server's start, when it attaches it).
pub fn register_live(hub: &Arc<SyncHub>) {
let mut live = LIVE.lock_or_recover();
live.retain(|w| w.strong_count() > 0);
live.push(Arc::downgrade(hub));
}
fn live_at(dir: &Path) -> Option<Arc<SyncHub>> {
LIVE.lock_or_recover()
.iter()
.filter_map(Weak::upgrade)
.rfind(|h| h.dir == dir)
}
/// ★ **The proof a mirror source's reports carry**, for the server under `data_root`:
/// signed by its live hub, as the sync downstream it is, to the upstream that source is
/// ([`SyncHub::report_proof`]). `None` from the closure while no hub is live.
pub fn proof_for(data_root: &Path, source: &str) -> crate::mirror_cycle::ReportProof {
let (dir, source) = (data_root.join(STATE_DIR), source.to_string());
Arc::new(move |req: &DownstreamReportRequest| {
live_at(&dir)?.report_proof(&source, &report_text(req))
})
}
/// ★ **The proof a mirror source's RON summary carries** (HOL-SYNC-SKEW, the
/// `summary_proof` field): signed like [`proof_for`], over [`summary_text`]. `None` from the
/// closure while no hub is live, or when the report has no summary.
pub fn summary_proof_for(data_root: &Path, source: &str) -> crate::mirror_cycle::ReportProof {
let (dir, source) = (data_root.join(STATE_DIR), source.to_string());
Arc::new(move |req: &DownstreamReportRequest| {
live_at(&dir)?.report_proof(&source, &summary_text(req)?)
})
}
/// ★ **The proof a mirror source's console address carries** (HOL-FLOW-FRONT, the
/// `console_proof` field): signed like [`proof_for`], over [`console_text`]. `None` from the
/// closure while no hub is live, or when the summary names no console.
pub fn console_proof_for(data_root: &Path, source: &str) -> crate::mirror_cycle::ReportProof {
let (dir, source) = (data_root.join(STATE_DIR), source.to_string());
Arc::new(move |req: &DownstreamReportRequest| {
live_at(&dir)?.report_proof(&source, &console_text(req)?)
})
}
impl SyncHub {
fn reports_dir(&self) -> PathBuf {
self.dir.join(REPORTS_DIR)
}
/// The upstream mirror source `source` reaches: the one of that name, else the only one.
fn upstream_for(&self, source: &str) -> Option<&SyncPeer> {
let ups = &self.cfg.upstreams;
ups.iter()
.find(|u| u.name == source)
.or(if ups.len() == 1 { ups.first() } else { None })
}
/// **Sign a report's text** as this downstream, to the upstream mirror source `source`
/// is: the [`REPORT_HEADER`] value. `None`: this Holger is no downstream of it.
pub fn report_proof(&self, source: &str, text: &str) -> Option<String> {
let up = self.upstream_for(source)?;
match self.key().sign_report(
&self.cfg.name,
&up.name,
text.as_bytes(),
holger_sync::now_secs(),
) {
Ok(s) => Some(base64::engine::general_purpose::STANDARD.encode(s.to_bytes())),
Err(e) => {
log::warn!("sync: the report to '{}' could not be signed: {e}", up.name);
None
}
}
}
/// ★ **A sync downstream's report, heard**: the proof opened under the LIVE downstreams
/// (a revoked one is refused, `holger-sync-5`), for this upstream, over exactly this
/// report, near now; a caller the front named (`x-holger-downstream`, a WireGuard
/// downstream) must be the one that signed. Kept, and answered with what was kept.
pub fn heard(
&self,
proof: &str,
req: &DownstreamReportRequest,
front_named: Option<&str>,
boot: u64,
head: u64,
now_ms: i64,
) -> Result<Heard, SyncFailure> {
let bytes = base64::engine::general_purpose::STANDARD
.decode(proof.trim())
.map_err(|e| {
SyncFailure::from(SyncError::Malformed(format!(
"the report's proof is not base64: {e}"
)))
})?;
let enrolled = self.enrolled()?;
let p = holger_sync::report::open_report(
&bytes,
&enrolled,
&self.cfg.name,
report_text(req).as_bytes(),
now_ms / 1000,
)?;
if let Some(n) = front_named.map(str::trim).filter(|n| !n.is_empty()) {
if n != p.downstream {
return Err(SyncFailure::new(
&holger_errcode::SYNC_DOWNSTREAM_REFUSED,
403,
format!("the front names this caller '{n}', and its report is signed by '{}': a downstream reports as itself", p.downstream),
));
}
}
let heard = Heard {
name: p.downstream.clone(),
every_secs: req.every_secs,
last_seen_ms: now_ms,
behind: behind(req, boot, head),
repos: req
.points
.iter()
.map(|x| {
(
x.repository.clone(),
(x.applied_boot, x.applied_seq, x.last_complete_unix_ms),
)
})
.collect(),
summary: self.proven_summary(req, &enrolled, &p.downstream, now_ms),
};
write_ron(
&self.reports_dir().join(format!("{}.ron", heard.name)),
&heard,
false,
)?;
Ok(heard)
}
/// **The report's RON summary, when its own proof holds** (HOL-SYNC-SKEW): signed by
/// the downstream that signed the report, for this upstream, over [`summary_text`] of
/// exactly this report, near now. A summary without its proof, or with one that does
/// not hold, is not kept (said at `warn`); the report is heard all the same, since its
/// own proof covers the sync points.
fn proven_summary(
&self,
req: &DownstreamReportRequest,
enrolled: &[holger_sync::TrustedPeer],
downstream: &str,
now_ms: i64,
) -> Option<crate::flow::RonSummary> {
let (summary, text) = (req.summary.as_ref()?, summary_text(req)?);
let why = match base64::engine::general_purpose::STANDARD
.decode(req.summary_proof.trim())
.map_err(|e| format!("its proof is not base64: {e}"))
.and_then(|bytes| {
holger_sync::report::open_report(
&bytes,
enrolled,
&self.cfg.name,
text.as_bytes(),
now_ms / 1000,
)
.map_err(|e| e.to_string())
}) {
Ok(q) if q.downstream == downstream => {
let mut kept: crate::flow::RonSummary = summary.into();
if !kept.console_url.is_empty()
&& !self.console_holds(req, enrolled, downstream, now_ms)
{
kept.console_url.clear();
}
return Some(kept);
}
Ok(q) => format!("its proof is signed by '{}'", q.downstream),
Err(_) if req.summary_proof.trim().is_empty() => "it has no proof".to_string(),
Err(e) => e,
};
log::warn!("sync: the RON summary in '{downstream}''s report is not kept: {why}");
None
}
/// **Does the summary's console address hold under its own proof** (HOL-FLOW-FRONT): signed
/// by the downstream that signed the report, for this upstream, over [`console_text`] of
/// exactly this report, near now. A console without its proof is dropped (said at `warn`),
/// the summary kept.
fn console_holds(
&self,
req: &DownstreamReportRequest,
enrolled: &[holger_sync::TrustedPeer],
downstream: &str,
now_ms: i64,
) -> bool {
let Some(text) = console_text(req) else {
return false;
};
let why = match base64::engine::general_purpose::STANDARD
.decode(req.console_proof.trim())
.map_err(|e| format!("its proof is not base64: {e}"))
.and_then(|bytes| {
holger_sync::report::open_report(
&bytes,
enrolled,
&self.cfg.name,
text.as_bytes(),
now_ms / 1000,
)
.map_err(|e| e.to_string())
}) {
Ok(q) if q.downstream == downstream => return true,
Ok(q) => format!("its proof is signed by '{}'", q.downstream),
Err(_) if req.console_proof.trim().is_empty() => "it has no proof".to_string(),
Err(e) => e,
};
log::warn!("sync: the console address in '{downstream}''s summary is not kept: {why}");
false
}
/// Every kept report.
pub fn reports(&self) -> Result<Vec<Heard>, SyncFailure> {
read_ron_dir(&self.reports_dir())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::audit::MemoryAuditLog;
use crate::grpc::holger_grow_proto::DownstreamPoint;
use holger_downstream_rules::via::Via;
const WG_A: &str = "AAECAwQFBgcICQoLDA0ODxAREhMUFRYXGBkaGxwdHh8=";
fn cfg(name: &str, downstreams: Vec<SyncPeer>, upstreams: Vec<SyncPeer>) -> SyncConfig {
SyncConfig {
name: name.into(),
key: None,
downstreams,
upstreams,
ack_deadline_secs: 60,
tick_secs: 30,
}
}
fn hub(dir: &Path, c: SyncConfig) -> SyncHub {
SyncHub::new(c, dir, Arc::new(MemoryAuditLog::new()), Weak::new()).unwrap()
}
fn peer(name: &str, key: String, via: Via) -> SyncPeer {
SyncPeer {
name: name.into(),
key,
url: None,
token_file: None,
via,
invite: None,
}
}
fn req(seq: u64) -> DownstreamReportRequest {
DownstreamReportRequest {
points: vec![DownstreamPoint {
repository: "bundles".into(),
applied_boot: 7,
applied_seq: seq,
last_complete_unix_ms: 5_000,
}],
every_secs: 15,
summary: None,
summary_proof: String::new(),
console_proof: String::new(),
}
}
/// ★ Main counts a sync downstream's mirror report by its enrolled name, for `via:
/// Front` and `via: Wireguard` alike: the board says when it last synced and how far
/// behind it is instead of "never"; a forged, tampered, misnamed or revoked report is
/// refused by name.
#[test]
fn main_counts_a_sync_downstreams_report_by_its_enrolled_name_for_both_vias() {
let (m, d1, d2) = (
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
tempfile::tempdir().unwrap(),
);
let upstream_of = |key: String| vec![peer("cloud", key, Via::Front)];
let main = hub(m.path(), cfg("cloud", vec![], vec![]));
let lab1 = hub(
d1.path(),
cfg("lab-1", vec![], upstream_of(main.public_key())),
);
let lab2 = hub(
d2.path(),
cfg("lab-2", vec![], upstream_of(main.public_key())),
);
main.add_downstream(peer("lab-1", lab1.public_key(), Via::Front), "console", "x")
.unwrap();
main.add_downstream(
peer(
"lab-2",
lab2.public_key(),
Via::Wireguard { key: WG_A.into() },
),
"console",
"x",
)
.unwrap();
let now_ms = holger_sync::now_secs() * 1000;
// Before any report: never.
let b = main.board(now_ms / 1000).unwrap();
assert!(
b.downstreams
.iter()
.all(|r| r.state == "never" && r.last_report_at == 0),
"{:?}",
b.downstreams
);
// lab-1 through the front's public door: no name from the front, its proof names it.
let p1 = lab1
.report_proof("cloud", &report_text(&req(10)))
.expect("lab-1 reports to cloud");
let h1 = main.heard(&p1, &req(10), None, 7, 10, now_ms).unwrap();
assert_eq!((h1.name.as_str(), h1.behind), ("lab-1", 0));
// lab-2 through the WireGuard sync door: the front named it, and it signed as itself.
let p2 = lab2
.report_proof("any-source", &report_text(&req(6)))
.expect("the only upstream");
assert_eq!(
main.heard(&p2, &req(6), Some("lab-2"), 7, 10, now_ms)
.unwrap()
.behind,
4
);
let b = main.board(now_ms / 1000).unwrap();
let row = |n: &str| b.downstreams.iter().find(|r| r.name == n).cloned().unwrap();
let (r1, r2) = (row("lab-1"), row("lab-2"));
assert_eq!(
(
r1.state.as_str(),
r1.via.as_str(),
r1.last_report_at,
r1.entries_behind,
r1.lag_generations
),
("current", "front", now_ms / 1000, 0, 0)
);
assert_eq!(
(r2.state.as_str(), r2.via.as_str(), r2.entries_behind),
("behind", "wireguard", 4)
);
assert!(
r2.headline.contains("4 of this upstream's writes"),
"{}",
r2.headline
);
// The report is kept (it rides the twin with the rest of holger-sync/).
assert_eq!(main.reports().unwrap().len(), 2);
assert!(m
.path()
.join(STATE_DIR)
.join(REPORTS_DIR)
.join("lab-2.ron")
.exists());
// Refused, by name: another report than the one signed, a caller the front named
// otherwise, a proof that is no proof, a stranger, and a revoked downstream.
assert!(main
.heard(&p1, &req(11), None, 7, 11, now_ms)
.unwrap_err()
.tagged()
.contains("not the one its proof was signed for"));
assert!(main
.heard(&p1, &req(10), Some("lab-2"), 7, 10, now_ms)
.unwrap_err()
.tagged()
.contains("signed by 'lab-1'"));
assert!(main
.heard("not base64!", &req(10), None, 7, 10, now_ms)
.unwrap_err()
.tagged()
.contains("holger-sync-1"));
let s = tempfile::tempdir().unwrap();
let stranger = hub(
s.path(),
cfg("lab-9", vec![], upstream_of(main.public_key())),
);
let p9 = stranger
.report_proof("cloud", &report_text(&req(10)))
.unwrap();
assert!(main
.heard(&p9, &req(10), None, 7, 10, now_ms)
.unwrap_err()
.tagged()
.contains("holger-sync-5"));
main.revoke_downstream("lab-1", "console", "x").unwrap();
assert!(
main.heard(&p1, &req(10), None, 7, 10, now_ms)
.unwrap_err()
.tagged()
.contains("holger-sync-5"),
"a revoked downstream is heard no more"
);
// A Holger with no upstream signs no report.
assert_eq!(main.report_proof("cloud", "x"), None);
}
/// The live hub of a data root signs for the mirror planned before it existed.
#[test]
fn a_mirror_planned_before_its_hub_signs_with_the_hub_its_server_attaches() {
let d = tempfile::tempdir().unwrap();
let proof = proof_for(d.path(), "cloud");
assert_eq!(proof(&req(1)), None, "no hub is live yet");
let h = Arc::new(hub(
d.path(),
cfg(
"lab-1",
vec![],
vec![peer("cloud", "a".repeat(64), Via::Front)],
),
));
register_live(&h);
let signed = proof(&req(1)).expect("the live hub signs");
let env = Signed::from_bytes(
&base64::engine::general_purpose::STANDARD
.decode(signed)
.unwrap(),
)
.unwrap();
assert_eq!(
(env.signer.as_str(), env.key.as_str()),
("lab-1", h.public_key().as_str())
);
}
#[test]
fn behind_is_the_most_behind_repository_and_another_boot_is_all_of_it() {
let mut r = req(8);
r.points.push(DownstreamPoint {
repository: "other".into(),
applied_boot: 7,
applied_seq: 3,
last_complete_unix_ms: 0,
});
assert_eq!(behind(&r, 7, 10), 7);
assert_eq!(behind(&r, 8, 10), 10);
assert_eq!(behind(&DownstreamReportRequest::default(), 7, 10), 0);
assert!(report_text(&r).ends_with("bundles\t7\t8\t5000\nother\t7\t3\t0\n"));
}
/// ★ **A downstream's RON summary rides in its report with its own proof**
/// (HOL-FLOW-SMART, HOL-SYNC-SKEW): the report's proof signs the text it always did, the
/// summary's proof signs the summary with that text, main keeps a summary whose proof
/// holds, and a changed summary, or a summary proof of another report, is not kept while
/// the report is still heard.
#[test]
fn a_reports_ron_summary_is_signed_and_kept() {
let (m, d1) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (main, lab1) = cloud_and_lab1(m.path(), d1.path());
let now_ms = holger_sync::now_secs() * 1000;
let plain = req(10);
assert_eq!(
report_text(&plain),
"holger.downstream-report-text/v1\nevery 15\nbundles\t7\t10\t5000\n",
"without a summary, the text it always was"
);
assert_eq!(summary_text(&plain), None);
let mut with = req(10);
with.summary = Some(lab_summary());
assert_eq!(report_text(&with), report_text(&plain), "and with one too");
let text = summary_text(&with).unwrap();
assert!(text.starts_with("holger.downstream-summary-text/v1\n"));
assert!(text.contains(&report_text(&plain)) && text.contains("pypi,oci"));
let proof = lab1.report_proof("cloud", &report_text(&with)).unwrap();
with.summary_proof = lab1.report_proof("cloud", &text).unwrap();
let h = main.heard(&proof, &with, None, 7, 10, now_ms).unwrap();
let kept = h.summary.clone().expect("the summary is kept");
assert_eq!(kept.repositories, vec!["pypi", "oci"]);
assert_eq!(kept.domain, "lab");
assert_eq!(
main.reports().unwrap()[0].summary,
Some(kept),
"on disk too"
);
let mut forged = with.clone();
forged.summary.as_mut().unwrap().domain = "prod".into();
let h = main.heard(&proof, &forged, None, 7, 10, now_ms).unwrap();
assert_eq!(
h.summary, None,
"another summary does not open under its proof"
);
let mut other = req(11);
other.summary = Some(lab_summary());
let other_proof = lab1.report_proof("cloud", &report_text(&other)).unwrap();
other.summary_proof = with.summary_proof.clone();
let h = main
.heard(&other_proof, &other, None, 7, 11, now_ms)
.unwrap();
assert_eq!(h.summary, None, "a summary proof belongs to its own report");
// Through the live hub, the way a mirror signs both.
let lab1 = Arc::new(lab1);
register_live(&lab1);
let mut sent = req(12);
sent.summary = Some(lab_summary());
sent.summary_proof = summary_proof_for(d1.path(), "cloud")(&sent).unwrap();
let report = proof_for(d1.path(), "cloud")(&sent).unwrap();
let h = main.heard(&report, &sent, None, 7, 12, now_ms).unwrap();
assert_eq!(h.summary.map(|s| s.domain), Some("lab".to_string()));
assert_eq!(summary_proof_for(d1.path(), "cloud")(&req(12)), None);
}
// ── HOL-SYNC-SKEW: an older and a newer holger hear each other ─────────────────
/// **The fixture: the report text every holger before HOL-FLOW-SMART signs and checks**
/// (`report_text` at 7abfe814, holger 3.0.0), byte for byte. An upstream and its
/// downstreams upgrade at different times, so this text is the contract.
fn v1_report_text(every_secs: u32, points: &[DownstreamPoint]) -> String {
let mut t = format!("holger.downstream-report-text/v1\nevery {every_secs}\n");
for p in points {
t.push_str(&format!(
"{}\t{}\t{}\t{}\n",
p.repository, p.applied_boot, p.applied_seq, p.last_complete_unix_ms
));
}
t
}
/// The report as an older holger decodes it: fields 1 and 2. A later field is unknown
/// to it, and protobuf drops it.
#[derive(Clone, PartialEq, ::prost::Message)]
struct V1ReportRequest {
#[prost(message, repeated, tag = "1")]
points: Vec<DownstreamPoint>,
#[prost(uint32, tag = "2")]
every_secs: u32,
}
fn lab_summary() -> crate::grpc::holger_grow_proto::DownstreamSummary {
crate::grpc::holger_grow_proto::DownstreamSummary {
boundary: "Fetch".into(),
repositories: vec!["pypi".into(), "oci".into()],
inbox: String::new(),
first_import: "admin".into(),
domain: "lab".into(),
downstreams: 2,
console_url: String::new(),
}
}
/// main ("cloud") with lab-1 enrolled, and lab-1 a downstream of it.
fn cloud_and_lab1(m: &Path, d1: &Path) -> (SyncHub, SyncHub) {
let main = hub(m, cfg("cloud", vec![], vec![]));
let lab1 = hub(
d1,
cfg(
"lab-1",
vec![],
vec![peer("cloud", main.public_key(), Via::Front)],
),
);
main.add_downstream(peer("lab-1", lab1.public_key(), Via::Front), "console", "x")
.unwrap();
(main, lab1)
}
#[test]
fn the_v1_fixture_is_the_text_holger_3_0_signs() {
assert_eq!(
v1_report_text(15, &req(10).points),
"holger.downstream-report-text/v1\nevery 15\nbundles\t7\t10\t5000\n"
);
}
/// ★ **An older upstream hears a newer downstream's report** (HOL-SYNC-SKEW): the newer
/// downstream's report carries its RON summary, the older upstream drops the field it
/// does not know and checks the proof against the v1 text of what is left, and the
/// proof opens. The summary never changes the text the report's proof signs.
#[test]
fn an_older_upstream_hears_a_newer_downstreams_report() {
let (m, d1) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (main, lab1) = cloud_and_lab1(m.path(), d1.path());
let lab1 = Arc::new(lab1);
register_live(&lab1);
let mut newer = req(10);
newer.summary = Some(lab_summary());
// Signed the way a newer downstream's mirror signs every report it sends.
let proof = proof_for(d1.path(), "cloud")(&newer).expect("lab-1 signs");
let wire = ::prost::Message::encode_to_vec(&newer);
let older: V1ReportRequest = ::prost::Message::decode(&wire[..]).unwrap();
let older_text = v1_report_text(older.every_secs, &older.points);
assert_eq!(
report_text(&newer),
older_text,
"the summary stays out of the text the report's proof signs"
);
let bytes = base64::engine::general_purpose::STANDARD
.decode(proof)
.unwrap();
let opened = holger_sync::report::open_report(
&bytes,
&main.enrolled().unwrap(),
"cloud",
older_text.as_bytes(),
holger_sync::now_secs(),
)
.expect("an older upstream opens a newer downstream's proof");
assert_eq!(opened.downstream, "lab-1");
}
/// ★ **A newer upstream hears an older downstream's report, and says it has no summary**
/// (HOL-SYNC-SKEW): the older downstream signs the v1 text and sends no summary; the
/// newer upstream hears it and keeps no summary for it.
#[test]
fn a_newer_upstream_hears_an_older_downstreams_report_without_a_summary() {
let (m, d1) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (main, lab1) = cloud_and_lab1(m.path(), d1.path());
let older = V1ReportRequest {
points: req(10).points,
every_secs: 15,
};
let proof = lab1
.report_proof("cloud", &v1_report_text(older.every_secs, &older.points))
.unwrap();
let wire = ::prost::Message::encode_to_vec(&older);
let heard_as: DownstreamReportRequest = ::prost::Message::decode(&wire[..]).unwrap();
let now_ms = holger_sync::now_secs() * 1000;
let h = main
.heard(&proof, &heard_as, None, 7, 10, now_ms)
.expect("a newer upstream hears an older downstream");
assert_eq!((h.name.as_str(), h.summary.clone()), ("lab-1", None));
assert_eq!(main.reports().unwrap()[0].summary, None, "none kept");
}
/// ★ **A summary's console address is kept only under its own proof** (HOL-FLOW-FRONT):
/// the downstream signs it apart from the summary, so an upstream that checks the summary
/// as holger 3.1.0 signs it still opens the summary (its text never names the address); a
/// newer upstream keeps the address only when `console_proof` holds for this report, and
/// keeps the summary either way.
#[test]
fn a_summarys_console_address_is_kept_only_under_its_own_proof() {
let (m, d1) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (main, lab1) = cloud_and_lab1(m.path(), d1.path());
let lab1 = Arc::new(lab1);
register_live(&lab1);
let now_ms = holger_sync::now_secs() * 1000;
let url = "https://www.lab-1.example";
let signed = |seq: u64, with_console: bool| {
let mut r = req(seq);
let mut s = lab_summary();
s.console_url = url.into();
r.summary = Some(s);
r.summary_proof = summary_proof_for(d1.path(), "cloud")(&r).unwrap();
if with_console {
r.console_proof = console_proof_for(d1.path(), "cloud")(&r).unwrap();
}
let proof = proof_for(d1.path(), "cloud")(&r).unwrap();
(r, proof)
};
let (r, proof) = signed(10, true);
assert!(
!summary_text(&r).unwrap().contains(url),
"the summary's text is holger 3.1.0's: it never names the address"
);
assert!(console_text(&r).unwrap().contains(url));
let h = main.heard(&proof, &r, None, 7, 10, now_ms).unwrap();
assert_eq!(h.summary.map(|s| s.console_url), Some(url.to_string()));
let (r, proof) = signed(11, false);
let h = main.heard(&proof, &r, None, 7, 11, now_ms).unwrap();
let s = h.summary.expect("the summary is kept");
assert_eq!(
(s.console_url.as_str(), s.domain.as_str()),
("", "lab"),
"no proof, no address"
);
let (mut r, proof) = signed(12, true);
r.summary.as_mut().unwrap().console_url = "https://evil.example".into();
let h = main.heard(&proof, &r, None, 7, 12, now_ms);
assert_eq!(
h.unwrap().summary.map(|s| s.console_url),
Some(String::new()),
"another address does not open under its proof"
);
assert_eq!(
console_proof_for(d1.path(), "cloud")(&req(13)),
None,
"no summary, no address, nothing to sign"
);
}
/// ★ **A summary without its own proof is not kept, and the report is still heard**
/// (HOL-SYNC-SKEW): the report's proof covers the sync points only, so a summary that
/// came without a proof of its own is dropped, never believed.
#[test]
fn a_summary_without_its_own_proof_is_dropped_and_the_report_heard() {
let (m, d1) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
let (main, lab1) = cloud_and_lab1(m.path(), d1.path());
let mut with = req(10);
with.summary = Some(lab_summary());
let proof = lab1.report_proof("cloud", &report_text(&with)).unwrap();
let now_ms = holger_sync::now_secs() * 1000;
let h = main
.heard(&proof, &with, None, 7, 10, now_ms)
.expect("the report is heard");
assert_eq!(h.summary, None, "an unproven summary is not kept");
}
}