server/lib/src/sync.rs
//! **The sync protocol, run against this Holger's intake chain** (Holger v2.0 H4;
//! the documents and the ledger are the `holger-sync` crate).
//!
//! ```text
//! UPSTREAM (sync.downstreams) DOWNSTREAM (sync.upstreams)
//! invalidate ──► ledger: generation+1 ──► held here too
//! usb_manifest / pull_manifest / bundle_manifest
//! │ signed by this Holger's sync key
//! ├── USB: holger-sync-<up>-<gen>.json on the stick, or evidence/holger-sync.json in the bundle
//! └── pull: GET /-/sync/manifest?downstream=&since= ──────► apply: verify under sync.upstreams,
//! anti-rollback, audit first,
//! IntakeChain::invalidate each match
//! import_ack ◄── USB: holger-sync-ack-<down>-<gen>.json ◄──────── signed ack (outbox + the stick)
//! ◄── pull: POST /-/sync/ack ◄──────────────────────── pull_once posts it
//! board / check_alarms: per downstream the generation applied, the lag, the invalidations
//! reached; ALARM (audit + board flag + the `holger::alarm` log line) past the deadline.
//! ```
//!
//! The downstream's default is zero outbound: a box with no `url` on its upstream
//! takes syncs from the stick (`holger-server sync apply`, `POST /-/sync/apply`, or a
//! bundle that carries a manifest) and never dials. Pull mode dials ONLY the
//! upstream's own door, through [`crate::egress::check`].
//!
//! The state is two JSON files under `<data_root>/holger-sync/`: the upstream's
//! ledger and the downstream's applied record. Each change is audited before it is
//! written; no audit record, no change.
use crate::lock::LockOrRecover as _;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, Weak};
use holger_errcode::ErrCode;
use serde::{Deserialize, Serialize};
pub use holger_sync as protocol;
/// How each downstream reaches this Holger, enrolled, invited and revoked (lane HOL-WGDS).
pub mod downstreams;
/// The invite an upstream gave this downstream, kept once: WireGuard or direct (HOL-PRODUCTS-7).
pub mod invited;
/// A sync downstream's mirror report, counted by its enrolled name (lane OX-WGDS2).
pub mod reports;
/// A WireGuard downstream's tunnel to its upstream's front (lane HOL-WGDS, `downstream-wg`).
#[cfg(feature = "downstream-wg")]
pub mod tunnel;
use holger_sync::applied::{Admission, Applied};
use holger_sync::ledger::{AckOutcome, Alarm, Ledger, SyncBoard};
use holger_sync::{Invalidation, PackageRef, Reason, Signed, SyncError, SyncKey, TrustedPeer};
#[cfg(feature = "fetch")]
use holger_sync::{SyncAck, ACK_SCHEMA};
use crate::audit::{AuditAction, AuditEvent, AuditLog};
use crate::promote::intake::{IntakeChain, InvalidateOutcome, INVALIDATED_PREFIX};
/// The state directory under the data root.
pub const STATE_DIR: &str = "holger-sync";
/// The upstream's ledger, in [`STATE_DIR`].
pub const LEDGER_FILE: &str = "upstream-ledger.json";
/// The downstream's applied record, in [`STATE_DIR`].
pub const APPLIED_FILE: &str = "downstream-applied.json";
/// Where the downstream keeps every ack it signed, in [`STATE_DIR`].
pub const OUTBOX_DIR: &str = "outbox";
/// The key file when `sync.key` names none, in [`STATE_DIR`].
pub const KEY_FILE: &str = "sync-key.pk8";
/// The log target the operator's alert path reads.
pub const ALARM_TARGET: &str = "holger::alarm";
/// The largest sync document a door or a link reads.
pub const MAX_DOCUMENT: usize = 16 * 1024 * 1024;
fn default_deadline() -> u64 {
24 * 60 * 60
}
fn default_tick() -> u64 {
30
}
/// **The `sync:` block.**
///
/// ```ron
/// sync: Some((
/// name: "dmz", // this Holger, as its peers name it
/// key: Some("/var/lib/holger/sync-key.pk8"), // minted when absent
/// downstreams: [ (name: "onprem", key: "<hex>") ], // upstream role: enrolled ack keys
/// // + `via: Front` (default) or `via: Wireguard(key: "<wg pub>")`
/// ack_deadline_secs: 86400, // the alarm's deadline
/// upstreams: [ (name: "dmz", key: "<hex>", url: Some("https://dmz:8443"), token_file: Some("…")) ],
/// )),
/// ```
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SyncConfig {
pub name: String,
/// The PKCS#8 Ed25519 key file. `None`: `<data_root>/holger-sync/sync-key.pk8`.
#[serde(default)]
pub key: Option<String>,
/// Upstream role: the downstreams whose acks this Holger takes, with their keys.
#[serde(default)]
pub downstreams: Vec<SyncPeer>,
/// Downstream role: the upstreams whose manifests this Holger applies.
#[serde(default)]
pub upstreams: Vec<SyncPeer>,
/// An invalidation not acknowledged within this many seconds raises the alarm.
#[serde(default = "default_deadline")]
pub ack_deadline_secs: u64,
/// How often the server checks the alarms and, in pull mode, pulls.
#[serde(default = "default_tick")]
pub tick_secs: u64,
}
impl SyncConfig {
/// **An upstream named `name`** with no downstream in its RON (an estate box: its
/// console enrols them, `downstreams`), the default deadline and tick.
pub fn upstream_of(name: &str) -> SyncConfig {
SyncConfig {
name: name.to_string(),
key: None,
downstreams: vec![],
upstreams: vec![],
ack_deadline_secs: default_deadline(),
tick_secs: default_tick(),
}
}
}
/// A peer in the `sync:` block.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SyncPeer {
pub name: String,
/// Its Ed25519 public key, hex (`holger-server sync key` prints it).
pub key: String,
/// Pull mode (a downstream's upstream only): the upstream's door. `None`: USB only.
#[serde(default)]
pub url: Option<String>,
/// A file holding the bearer token the pull presents at the upstream's door.
#[serde(default)]
pub token_file: Option<String>,
/// **How this downstream reaches this Holger** (a downstream entry only; lane
/// HOL-WGDS): `Front` (the default, the front's public door) or
/// `Wireguard(key: "<its WireGuard public key>")` (the front's downstream interface,
/// main's sync door only). An upstream's way in is its invite, never this.
#[serde(default)]
pub via: holger_downstream_rules::via::Via,
/// **An upstream reached over WireGuard** (an upstream entry only; lane HOL-WGDS): the
/// signed invite it gave this downstream. Verified once at the first start and kept;
/// the tunnel to the upstream's front comes up from it (server-lib `downstream-wg`).
#[serde(default)]
pub invite: Option<String>,
}
impl SyncPeer {
fn trusted(&self) -> TrustedPeer {
TrustedPeer {
name: self.name.clone(),
key: self.key.trim().to_ascii_lowercase(),
}
}
#[cfg(feature = "fetch")]
fn token(&self) -> Result<Option<String>, SyncFailure> {
match &self.token_file {
None => Ok(None),
Some(f) => std::fs::read_to_string(f)
.map(|t| Some(t.trim().to_string()))
.map_err(|e| {
SyncFailure::new(
&holger_errcode::SYNC_STATE,
500,
format!("the token file {f} for upstream '{}': {e}", self.name),
)
}),
}
}
}
/// Why a sync call did not happen. Nothing changed.
#[derive(Debug, Clone)]
pub struct SyncFailure {
pub code: &'static ErrCode,
pub status: u16,
pub message: String,
}
impl SyncFailure {
pub fn new(code: &'static ErrCode, status: u16, message: impl Into<String>) -> Self {
Self {
code,
status,
message: message.into(),
}
}
/// `[holger-sync-<n>] …`.
pub fn tagged(&self) -> String {
self.code.tag(&self.message)
}
fn not_configured(why: impl Into<String>) -> Self {
Self::new(&holger_errcode::SYNC_NOT_CONFIGURED, 404, why)
}
fn audit(e: impl std::fmt::Display) -> Self {
Self::new(
&holger_errcode::SYNC_AUDIT_FAILED,
500,
format!("the audit record could not be written, so nothing changed: {e}"),
)
}
#[cfg(feature = "fetch")]
fn link(why: impl Into<String>) -> Self {
Self::new(&holger_errcode::SYNC_LINK, 502, why)
}
}
impl From<SyncError> for SyncFailure {
fn from(e: SyncError) -> Self {
let (code, status) = match e {
SyncError::Malformed(_) => (&holger_errcode::SYNC_MALFORMED, 400),
SyncError::UntrustedSigner { .. } => (&holger_errcode::SYNC_UNTRUSTED_SIGNER, 403),
SyncError::BadSignature(_) => (&holger_errcode::SYNC_BAD_SIGNATURE, 403),
SyncError::Rollback { .. } => (&holger_errcode::SYNC_ROLLBACK, 409),
SyncError::UnknownDownstream(_) => (&holger_errcode::SYNC_UNKNOWN_DOWNSTREAM, 403),
SyncError::FutureGeneration { .. } => (&holger_errcode::SYNC_FUTURE_GENERATION, 409),
SyncError::Io(_) => (&holger_errcode::SYNC_STATE, 500),
};
Self::new(code, status, e.to_string())
}
}
impl std::fmt::Display for SyncFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(&self.tagged())
}
}
impl std::error::Error for SyncFailure {}
/// What [`SyncHub::invalidate`] did.
#[derive(Debug, Clone)]
pub struct Invalidated {
pub invalidation: Invalidation,
/// `false`: the same package and reason were invalidated before.
pub new: bool,
/// The coordinates held on THIS Holger's chain.
pub held_here: Vec<String>,
}
/// What [`SyncHub::apply`] did.
#[derive(Debug, Clone)]
pub struct AppliedSync {
pub upstream: String,
pub generation: u64,
/// `new` or `replay`.
pub admission: String,
/// How many packages the manifest adds.
pub added: usize,
/// The coordinates held just now.
pub quarantined: Vec<String>,
/// The coordinates an invalidation names that were held already.
pub already_held: Vec<String>,
/// The invalidations that name no package here.
pub unmatched: Vec<String>,
/// The signed ack: the RECEIPT, a `fetch` build's only (a one-way Holger sends
/// nothing back, so it signs none).
#[cfg(feature = "fetch")]
pub ack: Signed,
/// Where the ack was kept (the outbox).
#[cfg(feature = "fetch")]
pub ack_path: PathBuf,
/// Pull mode: the ack reached the upstream.
#[cfg(feature = "fetch")]
pub ack_sent: bool,
}
/// **This Holger's part in the sync protocol.** Built with the intake chain when
/// the config has a `sync:` block, and reached through [`IntakeChain::sync`].
pub struct SyncHub {
cfg: SyncConfig,
/// The key as it was last read, with the bytes it was read from ([`SyncHub::key`]).
key: Mutex<(Vec<u8>, Arc<SyncKey>)>,
key_path: PathBuf,
dir: PathBuf,
audit: Arc<dyn AuditLog>,
chain: Weak<IntakeChain>,
serial: Mutex<()>,
/// This Holger's own sync door (`http(s)://host:port`), the `door` of the direct invites it
/// gives (HOL-PRODUCTS-7): its first exposed endpoint's HTTP address ([`SyncHub::with_door`]).
door: Option<String>,
}
/// **A package's coordinate in sync documents**: `ns/name@version`, without the
/// empty parts (a raw or pip file is its path alone).
pub fn coordinate_of(id: &traits::ArtifactId) -> String {
let base = match id.namespace.as_deref().filter(|n| !n.is_empty()) {
Some(ns) => format!("{ns}/{}", id.name),
None => id.name.clone(),
};
if id.version.is_empty() {
base
} else {
format!("{base}@{}", id.version)
}
}
/// Does `coordinate` name `id`, in the sync form or in the intake record's label?
fn names(id: &traits::ArtifactId, coordinate: &str) -> bool {
coordinate_of(id) == coordinate || crate::promote::artifact_label(id) == coordinate
}
fn is_key_hex(k: &str) -> bool {
let k = k.trim();
k.len() == 64 && k.chars().all(|c| c.is_ascii_hexdigit())
}
impl SyncHub {
/// The hub for `cfg` under `data_root`. The key is opened, or minted on the first start.
pub fn new(
cfg: SyncConfig,
data_root: &Path,
audit: Arc<dyn AuditLog>,
chain: Weak<IntakeChain>,
) -> Result<Self, SyncFailure> {
let bad = |why: String| {
SyncFailure::new(
&holger_errcode::SYNC_MALFORMED,
400,
format!("config `sync`: {why}"),
)
};
if cfg.name.trim().is_empty() {
return Err(bad(
"`name` is empty; a Holger signs its sync documents under its name".into(),
));
}
for p in cfg.downstreams.iter().chain(cfg.upstreams.iter()) {
if !is_key_hex(&p.key) {
return Err(bad(format!(
"peer '{}': `key` must be its Ed25519 public key as 64 hex digits",
p.name
)));
}
}
for p in &cfg.downstreams {
if p.url.is_some() || p.token_file.is_some() {
return Err(bad(format!("downstream '{}': `url` and `token_file` belong to an upstream; a downstream reaches its upstream, never the other way", p.name)));
}
}
for p in &cfg.downstreams {
if p.invite.is_some() {
return Err(bad(format!("downstream '{}': `invite` belongs to an upstream entry (the invite that upstream gave this Holger)", p.name)));
}
}
for p in &cfg.upstreams {
if p.via != holger_downstream_rules::via::Via::Front {
return Err(bad(format!("upstream '{}': `via` belongs to a downstream entry; a downstream reaches a WireGuard upstream through the `invite:` that upstream gave it", p.name)));
}
}
holger_downstream_rules::via::validate(&downstreams::decls(&cfg.downstreams))
.map_err(|e| SyncFailure::new(&holger_errcode::SYNC_DOWNSTREAM_REFUSED, 400, e))?;
let dir = data_root.join(STATE_DIR);
let key_path = cfg
.key
.as_ref()
.map(PathBuf::from)
.unwrap_or_else(|| dir.join(KEY_FILE));
crate::keyfile::check_private(&key_path, "the sync key")
.map_err(|e| SyncError::Io(e.to_string()))?;
let key = SyncKey::open_or_mint(&key_path)?;
let der = crate::keyfile::read_private(&key_path, "the sync key")
.map_err(|e| SyncError::Io(format!("read {}: {e}", key_path.display())))?;
Ok(Self {
cfg,
key: Mutex::new((der, Arc::new(key))),
key_path,
dir,
audit,
chain,
serial: Mutex::new(()),
door: None,
})
}
/// ★ **The sync key, as the disk holds it now** (lane OX-WGDS2, twin parity). A main
/// rebuilt from its twin starts as a fresh box and mints a key of its own; the recharge
/// then brings the estate's own key over it (`replication::ESTATE` names `holger-sync`),
/// while this process runs. From that moment this Holger signs under the key its
/// downstreams trust, without a restart. A key file that cannot be read keeps the last one.
fn key(&self) -> Arc<SyncKey> {
let mut held = self.key.lock_or_recover();
let read = crate::keyfile::read_private(&self.key_path, "the sync key").inspect_err(|e| {
if e.kind() == std::io::ErrorKind::PermissionDenied {
log::error!("sync: {e}; signing under the last key");
}
});
if let Ok(der) = read {
if der != held.0 {
match SyncKey::open_or_mint(&self.key_path) {
Ok(k) => {
log::warn!("sync: the key at {} changed on disk (the estate's own, from a recharge); '{}' signs under {} from now on", self.key_path.display(), self.cfg.name, k.public_hex());
*held = (der, Arc::new(k));
}
Err(e) => log::error!("sync: the key at {} changed on disk and does not read ({e}); signing under the last one", self.key_path.display()),
}
}
}
Arc::clone(&held.1)
}
pub fn config(&self) -> &SyncConfig {
&self.cfg
}
pub fn name(&self) -> &str {
&self.cfg.name
}
/// This Holger's sync public key, hex: what its peers put in their `sync:` block.
pub fn public_key(&self) -> String {
self.key().public_hex()
}
/// Does this Holger have downstreams (the RON's or the console's, live)?
pub fn is_upstream(&self) -> bool {
self.downstreams()
.map(|d| !d.is_empty())
.unwrap_or(!self.cfg.downstreams.is_empty())
}
/// Does this Holger have upstreams?
pub fn is_downstream(&self) -> bool {
!self.cfg.upstreams.is_empty()
}
/// The data root this hub's state lives under (`<data>/holger-sync`'s parent). Read by the
/// fetch code only (the pull, the WireGuard tunnel), so a one-way build has none.
#[cfg(feature = "fetch")]
fn data_root(&self) -> PathBuf {
self.dir
.parent()
.map(Path::to_path_buf)
.unwrap_or_else(|| self.dir.clone())
}
/// ★ **This downstream's WireGuard public key** (lane OX-WGDS3), base64 — public data: what
/// its upstream's admin declares `via: Wireguard(key: ..)`. Minted once, on the first ask.
/// `None` on a Holger that reaches no upstream over WireGuard (no `invite:` in its RON).
#[cfg(feature = "downstream-wg")]
pub fn wireguard_key(&self) -> Option<String> {
if !self.cfg.upstreams.iter().any(|u| u.invite.is_some()) {
return None;
}
tunnel::identity(&self.data_root())
.ok()
.map(|id| id.public_b64)
}
/// ★ **The invite of `upstream`, handed in at this downstream's door** (lane OX-WGDS3,
/// `POST /-/sync/upstreams/invite`): verified and kept once ([`tunnel::take`]), the tunnel
/// brought up when it can be, and the taking audited. An upstream this RON does not reach
/// over WireGuard is a 404 by name.
#[cfg(feature = "downstream-wg")]
pub fn take_invite(
&self,
upstream: &str,
given: &[u8],
actor: &str,
source: &str,
) -> Result<serde_json::Value, SyncFailure> {
let Some(u) = self.cfg.upstreams.iter().find(|u| u.name == upstream) else {
return Err(SyncFailure::new(
&holger_errcode::SYNC_DOWNSTREAM_REFUSED,
404,
format!("'{upstream}' is no upstream of this Holger"),
));
};
if u.invite.is_none() {
return Err(SyncFailure::new(&holger_errcode::SYNC_DOWNSTREAM_REFUSED, 404, format!("upstream '{upstream}' is not reached over WireGuard here (its entry names no `invite:`)")));
}
let _one = self.serial.lock_or_recover();
let answer = tunnel::take(&self.data_root(), &self.cfg.name, u, given)?;
self.record(self.event(actor, upstream, source, format!("sync: the invite of upstream '{upstream}' kept (address {}, the front at {}); tunnel {}", answer["address"].as_str().unwrap_or(""), answer["endpoint"].as_str().unwrap_or(""), answer["tunnel"].as_str().unwrap_or(""))))?;
Ok(answer)
}
/// This Holger's own sync door, named in the direct invites it gives.
pub fn with_door(mut self, door: impl Into<String>) -> Self {
self.door = Some(door.into());
self
}
/// The door the direct invites name when the admin names none.
pub fn door(&self) -> Option<&str> {
self.door.as_deref()
}
fn chain(&self) -> Result<Arc<IntakeChain>, SyncFailure> {
self.chain.upgrade().ok_or_else(|| {
SyncFailure::not_configured("the intake chain is gone (the server is stopping)")
})
}
fn enrolled(&self) -> Result<Vec<TrustedPeer>, SyncFailure> {
Ok(self.downstreams()?.iter().map(SyncPeer::trusted).collect())
}
fn enrolled_names(&self) -> Result<Vec<String>, SyncFailure> {
Ok(self.downstreams()?.iter().map(|p| p.name.clone()).collect())
}
fn ledger(&self) -> Result<Ledger, SyncFailure> {
let mut l: Ledger = holger_sync::read_state(&self.dir.join(LEDGER_FILE))?;
if l.upstream.is_empty() {
l.upstream = self.cfg.name.clone();
}
Ok(l)
}
fn save_ledger(&self, l: &Ledger) -> Result<(), SyncFailure> {
Ok(holger_sync::write_state(&self.dir.join(LEDGER_FILE), l)?)
}
fn record(&self, ev: AuditEvent) -> Result<(), SyncFailure> {
self.audit
.record(ev)
.and_then(|()| self.audit.flush())
.map_err(|e| SyncFailure::audit(format!("{e:?}")))
}
fn event(&self, actor: &str, target: &str, source: &str, detail: String) -> AuditEvent {
AuditEvent::new(actor, AuditAction::Other, STATE_DIR, target, source, 200, 0)
.with_detail(detail)
}
/// The AVAILABLE set of this chain, as package refs.
pub fn available(&self) -> Result<Vec<PackageRef>, SyncFailure> {
let chain = self.chain()?;
Ok(chain
.available()
.into_iter()
.map(|(id, s)| PackageRef {
coordinate: coordinate_of(&id),
sha256: s.sha256,
})
.collect())
}
// ── the upstream ────────────────────────────────────────────────────
/// **Invalidate a package** and hold it on this Holger too. `sha256` `None`: the
/// bytes this chain holds at `coordinate`, when it holds any; else every package
/// at the coordinate.
pub fn invalidate(
&self,
actor: &str,
source: &str,
coordinate: &str,
sha256: Option<&str>,
reason: Reason,
detail: &str,
) -> Result<Invalidated, SyncFailure> {
let chain = self.chain()?;
let sha = match sha256.map(str::trim).filter(|s| !s.is_empty()) {
Some(s) => Some(s.to_ascii_lowercase()),
None => chain
.all()
.into_iter()
.find(|(id, _)| names(id, coordinate))
.map(|(_, s)| s.sha256),
};
let (inv, new) = {
let _one = self.serial.lock_or_recover();
let mut l = self.ledger()?;
let (inv, new) = l.invalidate(
coordinate,
sha,
reason,
detail,
actor,
holger_sync::now_secs(),
);
if new {
self.record(self.event(
actor,
coordinate,
source,
format!(
"sync invalidation {} issued in generation {}: {} ({}){}; sha256 {}",
inv.id,
inv.generation,
inv.coordinate,
inv.reason.words(),
if inv.detail.is_empty() {
String::new()
} else {
format!(": {}", inv.detail)
},
inv.sha256.as_deref().unwrap_or("any")
),
))?;
self.save_ledger(&l)?;
}
(inv, new)
};
let (held_here, _, _) = hold(
&chain,
std::slice::from_ref(&inv),
&self.cfg.name,
actor,
source,
)?;
Ok(Invalidated {
invalidation: inv,
new,
held_here,
})
}
fn signed_manifest(&self, since: u64, carrier: &str) -> Result<Signed, SyncFailure> {
let available = self.available()?;
let _one = self.serial.lock_or_recover();
let mut l = self.ledger()?;
if l.absorb_available(&available, holger_sync::now_secs())
.is_some()
{
self.save_ledger(&l)?;
}
Ok(self.key().sign_manifest(&l.manifest(since, carrier))?)
}
/// **The manifest a downstream pulls**: the AVAILABLE set not yet in a
/// generation becomes one, then every package after `since` and every
/// invalidation, signed.
pub fn pull_manifest(&self, downstream: &str, since: u64) -> Result<Signed, SyncFailure> {
if !self.downstreams()?.iter().any(|p| p.name == downstream) {
return Err(SyncError::UnknownDownstream(downstream.to_string()).into());
}
self.signed_manifest(since, "pull")
}
/// **The manifest for a stick**, without a bundle (an invalidation that must
/// travel now, or the metadata beside a bundle).
pub fn usb_manifest(&self, since: u64) -> Result<Signed, SyncFailure> {
self.signed_manifest(since, "usb")
}
/// **The manifest an airgap bundle carries**: a new generation with exactly
/// the packages the bundle carries.
pub fn bundle_manifest(
&self,
added: Vec<PackageRef>,
carrier: &str,
) -> Result<Signed, SyncFailure> {
let _one = self.serial.lock_or_recover();
let mut l = self.ledger()?;
l.add_generation(carrier, added.clone(), holger_sync::now_secs());
self.save_ledger(&l)?;
// The decision profile (HOL-NCSC, `crate::export_profile`): an invalidation crosses
// without the upstream principal who issued it.
let mut m = l.manifest_with(added, carrier);
for inv in &mut m.invalidations {
inv.issued_by.clear();
}
Ok(self.key().sign_manifest(&m)?)
}
/// Write `m` into `dir` as `holger-sync-<upstream>-<generation>.json`.
pub fn write_manifest(&self, m: &Signed, dir: &Path) -> Result<PathBuf, SyncFailure> {
let doc: holger_sync::SyncManifest = m.open(
holger_sync::Kind::Manifest,
&[TrustedPeer {
name: self.cfg.name.clone(),
key: self.public_key(),
}],
)?;
write_file(dir, &holger_sync::manifest_file_name(&doc), &m.to_bytes())
}
/// **Take a downstream's ack** (a file from the stick, or the pull's POST).
pub fn import_ack(
&self,
bytes: &[u8],
carrier: &str,
source: &str,
) -> Result<AckOutcome, SyncFailure> {
if !self.is_upstream() {
return Err(SyncFailure::not_configured(
"this Holger enrolls no downstream (`sync.downstreams`), so it takes no ack",
));
}
let ack = holger_sync::open_ack(bytes, &self.enrolled()?)?;
let _one = self.serial.lock_or_recover();
let mut l = self.ledger()?;
let mut probe = l.clone();
let out = probe.record_ack(&ack, holger_sync::now_secs())?;
let actor = format!("holger-sync:{}", ack.downstream);
self.record(self.event(
&actor,
&ack.downstream,
source,
format!(
"sync ack ({carrier}) from '{}': generation {} applied, {} package(s) added, {} invalidation(s) in force, {} reached now [{}]{}",
ack.downstream,
ack.generation,
ack.packages_added,
ack.invalidations_applied.len(),
out.reached_now.len(),
out.reached_now.join(", "),
if out.cleared.is_empty() { String::new() } else { format!("; alarm cleared for [{}]", out.cleared.join(", ")) }
),
))?;
l = probe;
if let Some(r) = l.downstreams.get_mut(&ack.downstream) {
r.carrier = carrier.to_string();
}
self.save_ledger(&l)?;
for id in &out.cleared {
log::warn!(target: ALARM_TARGET, "CLEARED sync: downstream '{}' acknowledged invalidation {id}", ack.downstream);
}
Ok(out)
}
/// **Raise the alarms due at `now`**: each is audited, logged on
/// [`ALARM_TARGET`] and flagged on the board, once.
pub fn check_alarms(&self, now: i64) -> Result<Vec<Alarm>, SyncFailure> {
if !self.is_upstream() {
return Ok(Vec::new());
}
let _one = self.serial.lock_or_recover();
let mut l = self.ledger()?;
let alarms = l.check_alarms(&self.enrolled_names()?, now, self.cfg.ack_deadline_secs);
if alarms.is_empty() {
return Ok(alarms);
}
for a in &alarms {
let mut ev = self.event("holger-sync", &a.invalidation.coordinate, "sync", a.line());
ev.status = 503;
self.record(ev)?;
}
self.save_ledger(&l)?;
for a in &alarms {
log::error!(target: ALARM_TARGET, "{}", a.line());
}
Ok(alarms)
}
/// **The board's sync data** at `now` (the alarms due are raised first).
pub fn board(&self, now: i64) -> Result<SyncBoard, SyncFailure> {
self.check_alarms(now)?;
let l = self.ledger()?;
let live = self.downstreams()?;
let mut b = l.board(&self.enrolled_names()?, now, self.cfg.ack_deadline_secs);
// Each downstream with how it reaches this Holger (lane HOL-WGDS).
// And with its mirror's last report (lane OX-WGDS2): when it last synced, how far behind.
let heard = self.reports()?;
for r in &mut b.downstreams {
r.via = live
.iter()
.find(|p| p.name == r.name)
.map(|p| p.via.word())
.unwrap_or("front")
.to_string();
if let Some(h) = heard.iter().find(|h| h.name == r.name) {
reports::with_report(r, h);
}
}
Ok(b)
}
// ── the downstream ──────────────────────────────────────────────────
/// **Apply a signed manifest**: verify it under `sync.upstreams`, refuse a
/// rollback, audit first, hold every package an invalidation names, record the
/// generation, and sign the ack (kept in the outbox).
pub fn apply(
&self,
bytes: &[u8],
carrier: &str,
actor: &str,
source: &str,
) -> Result<AppliedSync, SyncFailure> {
if !self.is_downstream() {
return Err(SyncFailure::not_configured(
"this Holger trusts no upstream (`sync.upstreams`), so it applies no manifest",
));
}
let trusted: Vec<TrustedPeer> = self.cfg.upstreams.iter().map(SyncPeer::trusted).collect();
let m = holger_sync::open_manifest(bytes, &trusted)?;
let chain = self.chain()?;
let _one = self.serial.lock_or_recover();
let path = self.dir.join(APPLIED_FILE);
let mut applied: Applied = holger_sync::read_state(&path)?;
let admission = applied.admit(&m)?;
let word = if admission == Admission::Replay {
"replay"
} else {
"new"
};
self.record(self.event(
actor,
&m.upstream,
source,
format!(
"sync apply ({carrier}, {word}) of '{}' generation {}: {} package(s) added, {} invalidation(s)",
m.upstream,
m.generation,
m.added.len(),
m.invalidations.len()
),
))?;
let (quarantined, already_held, unmatched) = hold(
&chain,
&m.invalidations,
&m.upstream,
&format!("holger-sync:{}", m.upstream),
source,
)?;
let now = holger_sync::now_secs();
applied.record(&m, now);
holger_sync::write_state(&path, &applied)?;
#[cfg(not(feature = "fetch"))]
return Ok(AppliedSync {
upstream: m.upstream,
generation: m.generation,
admission: word.into(),
added: m.added.len(),
quarantined,
already_held,
unmatched,
});
#[cfg(feature = "fetch")]
let ack = SyncAck {
schema: ACK_SCHEMA.into(),
downstream: self.cfg.name.clone(),
upstream: m.upstream.clone(),
generation: m.generation,
applied_at: now,
packages_added: m.added.len() as u64,
invalidations_applied: applied.ids(&m.upstream),
quarantined: quarantined.clone(),
carrier: carrier.to_string(),
};
#[cfg(feature = "fetch")]
let signed = self.key().sign_ack(&ack)?;
#[cfg(feature = "fetch")]
let ack_path = write_file(
&self.dir.join(OUTBOX_DIR),
&holger_sync::ack_file_name(&ack),
&signed.to_bytes(),
)?;
#[cfg(feature = "fetch")]
Ok(AppliedSync {
upstream: m.upstream,
generation: m.generation,
admission: word.into(),
added: m.added.len(),
quarantined,
already_held,
unmatched,
ack: signed,
ack_path,
ack_sent: false,
})
}
/// **Verify a manifest without applying it** (an import checks the one its
/// bundle carries before anything lands): the signature under `sync.upstreams`
/// and the anti-rollback floor.
pub fn check_manifest(&self, bytes: &[u8]) -> Result<holger_sync::SyncManifest, SyncFailure> {
let trusted: Vec<TrustedPeer> = self.cfg.upstreams.iter().map(SyncPeer::trusted).collect();
let m = holger_sync::open_manifest(bytes, &trusted)?;
let applied: Applied = holger_sync::read_state(&self.dir.join(APPLIED_FILE))?;
applied.admit(&m)?;
Ok(m)
}
/// Write the signed `ack` into `dir` (the stick) as `holger-sync-ack-<downstream>-<generation>.json`.
#[cfg(feature = "fetch")]
pub fn write_ack(&self, ack: &Signed, dir: &Path) -> Result<PathBuf, SyncFailure> {
let doc: SyncAck = ack.open(
holger_sync::Kind::Ack,
&[TrustedPeer {
name: self.cfg.name.clone(),
key: self.public_key(),
}],
)?;
write_file(dir, &holger_sync::ack_file_name(&doc), &ack.to_bytes())
}
/// **Pull mode, once**: fetch the manifest from `upstream`'s door, apply it, post the ack.
#[cfg(feature = "fetch")]
pub fn pull_once(
&self,
link: &dyn SyncLink,
upstream: &SyncPeer,
) -> Result<AppliedSync, SyncFailure> {
// Its `url:`, else the door of the direct invite it kept (HOL-PRODUCTS-7).
let Some(base) = invited::door_of(
&self.data_root(),
&self.cfg.name,
&self.public_key(),
upstream,
)
.map(|u| u.trim_end_matches('/').to_string()) else {
return Err(SyncFailure::not_configured(format!(
"upstream '{}' has no `url` and no direct invite kept; it syncs by stick only",
upstream.name
)));
};
let token = upstream.token()?;
let since = {
let applied: Applied = holger_sync::read_state(&self.dir.join(APPLIED_FILE))?;
applied.last_generation(&upstream.name)
};
let url = format!(
"{base}/-/sync/manifest?downstream={}&since={since}",
self.cfg.name
);
let body = link
.get(&url, token.as_deref())
.map_err(SyncFailure::link)?;
let mut applied = self.apply(
&body,
"pull",
&format!("holger-sync:pull:{}", upstream.name),
&base,
)?;
match link.post(&format!("{base}/-/sync/ack"), &applied.ack.to_bytes(), token.as_deref()) {
Ok(_) => applied.ack_sent = true,
Err(e) => log::warn!(
"sync: the ack of generation {} could not reach '{}' ({e}); it stays in {} and goes with the next pull",
applied.generation,
upstream.name,
applied.ack_path.display()
),
}
Ok(applied)
}
/// **One tick of the server's sync job**: the alarms due on an upstream; on a
/// downstream, one pull from each upstream with a `url` or a kept direct invite (with a link).
pub fn tick(&self, link: Option<&dyn SyncLink>) -> String {
let mut words = Vec::new();
match self.check_alarms(holger_sync::now_secs()) {
Ok(a) if !a.is_empty() => words.push(format!("{} alarm(s) raised", a.len())),
Ok(_) => {}
Err(e) => words.push(format!("the alarm check failed: {}", e.tagged())),
}
#[cfg(not(feature = "fetch"))]
let _ = link;
#[cfg(feature = "fetch")]
if let Some(link) = link {
let (dir, me, key) = (self.data_root(), self.cfg.name.clone(), self.public_key());
for u in self
.cfg
.upstreams
.iter()
.filter(|u| invited::door_of(&dir, &me, &key, u).is_some())
{
match self.pull_once(link, u) {
Ok(a) => words.push(format!(
"'{}' generation {} ({}), {} held",
u.name,
a.generation,
a.admission,
a.quarantined.len()
)),
Err(e) => {
log::warn!("sync: the pull from '{}' failed: {}", u.name, e.tagged());
words.push(format!("'{}': {}", u.name, e.tagged()));
}
}
}
}
if words.is_empty() {
"sync: nothing due".into()
} else {
format!("sync: {}", words.join("; "))
}
}
}
/// Hold on `chain` every package `invs` name. `(held now, held already, ids that matched nothing)`.
/// `(held now, held already, ids that matched nothing)`.
type Held = (Vec<String>, Vec<String>, Vec<String>);
fn hold(
chain: &IntakeChain,
invs: &[Invalidation],
upstream: &str,
actor: &str,
source: &str,
) -> Result<Held, SyncFailure> {
let records = chain.all();
let (mut held, mut already, mut unmatched) = (Vec::new(), Vec::new(), Vec::new());
for inv in invs {
let mut any = false;
for (id, status) in &records {
let label = coordinate_of(id);
let hit = match &inv.sha256 {
Some(_) => inv.matches(&label, &status.sha256),
None => names(id, &inv.coordinate),
};
if !hit {
continue;
}
any = true;
let detail = if inv.detail.is_empty() {
inv.reason.words().to_string()
} else {
inv.detail.clone()
};
let reason = format!(
"{INVALIDATED_PREFIX} '{upstream}' (generation {}): {}: {detail} [{}]",
inv.generation, inv.reason, inv.id
);
match chain.invalidate(id, actor, source, &reason) {
Ok(InvalidateOutcome::Held(_)) => held.push(label),
Ok(InvalidateOutcome::AlreadyHeld(_)) => already.push(label),
Err(e) => {
return Err(SyncFailure::new(
e.code(),
e.status(),
format!("holding '{label}' for invalidation {}: {e}", inv.id),
))
}
}
}
if !any {
unmatched.push(inv.id.clone());
}
}
Ok((held, already, unmatched))
}
fn write_file(dir: &Path, name: &str, bytes: &[u8]) -> Result<PathBuf, SyncFailure> {
std::fs::create_dir_all(dir)
.map_err(|e| SyncError::Io(format!("create {}: {e}", dir.display())))?;
let path = dir.join(name);
std::fs::write(&path, bytes)
.map_err(|e| SyncError::Io(format!("write {}: {e}", path.display())))?;
Ok(path)
}
/// **The pull's road to its upstream.** The server's is [`HttpLink`]; a test may
/// hand its own.
pub trait SyncLink: Send + Sync {
fn get(&self, url: &str, token: Option<&str>) -> Result<Vec<u8>, String>;
fn post(&self, url: &str, body: &[u8], token: Option<&str>) -> Result<Vec<u8>, String>;
}
/// The HTTP road (`online`): every dial asks [`crate::egress::check`] first.
#[cfg(all(feature = "online", feature = "fetch"))]
pub struct HttpLink {
agent: ureq::Agent,
}
#[cfg(all(feature = "online", feature = "fetch"))]
impl HttpLink {
pub fn new() -> Self {
Self {
agent: crate::upstream_rust::build_agent(
std::time::Duration::from_secs(10),
std::time::Duration::from_secs(60),
),
}
}
fn finish(
url: &str,
resp: Result<ureq::http::Response<ureq::Body>, ureq::Error>,
) -> Result<Vec<u8>, String> {
use std::io::Read as _;
let resp = resp.map_err(|e| format!("{url}: {e}"))?;
let status = resp.status().as_u16();
let mut body = Vec::new();
resp.into_body()
.into_reader()
.take(MAX_DOCUMENT as u64 + 1)
.read_to_end(&mut body)
.map_err(|e| format!("{url}: reading the body: {e}"))?;
if body.len() > MAX_DOCUMENT {
return Err(format!(
"{url}: the answer is larger than {MAX_DOCUMENT} bytes"
));
}
if status != 200 {
return Err(format!(
"{url}: HTTP {status}: {}",
String::from_utf8_lossy(&body).trim()
));
}
Ok(body)
}
}
#[cfg(all(feature = "online", feature = "fetch"))]
impl Default for HttpLink {
fn default() -> Self {
Self::new()
}
}
#[cfg(all(feature = "online", feature = "fetch"))]
impl SyncLink for HttpLink {
fn get(&self, url: &str, token: Option<&str>) -> Result<Vec<u8>, String> {
crate::egress::check("sync pull", url).map_err(|e| e.to_string())?;
let mut req = self.agent.get(url);
if let Some(t) = token {
req = req.header("Authorization", &format!("Bearer {t}"));
}
Self::finish(url, req.call())
}
fn post(&self, url: &str, body: &[u8], token: Option<&str>) -> Result<Vec<u8>, String> {
crate::egress::check("sync ack", url).map_err(|e| e.to_string())?;
let mut req = self
.agent
.post(url)
.header("Content-Type", "application/json");
if let Some(t) = token {
req = req.header("Authorization", &format!("Bearer {t}"));
}
Self::finish(url, req.send(body))
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn every_sync_error_has_its_code() {
let all = [
SyncError::Malformed(String::new()),
SyncError::UntrustedSigner {
signer: String::new(),
why: String::new(),
},
SyncError::BadSignature(String::new()),
SyncError::Rollback {
upstream: String::new(),
applied: 2,
offered: 1,
},
SyncError::UnknownDownstream(String::new()),
SyncError::FutureGeneration {
offered: 2,
issued: 1,
},
SyncError::Io(String::new()),
];
for e in all {
let n = e.number();
let f = SyncFailure::from(e);
assert_eq!(f.code.code, format!("holger-sync-{n}"), "{f}");
}
assert_eq!(SyncFailure::not_configured("x").code.code, "holger-sync-8");
assert_eq!(SyncFailure::audit("x").code.code, "holger-sync-9");
assert_eq!(SyncFailure::link("x").code.code, "holger-sync-10");
}
#[test]
fn a_config_with_a_bad_key_or_a_downstream_url_is_refused_by_name() {
let d = tempfile::tempdir().unwrap();
let audit: Arc<dyn AuditLog> = Arc::new(crate::audit::NoopAuditLog);
let base = SyncConfig {
name: "dmz".into(),
key: None,
downstreams: vec![],
upstreams: vec![],
ack_deadline_secs: 60,
tick_secs: 30,
};
let peer = |key: &str, url: Option<&str>| SyncPeer {
name: "p".into(),
key: key.into(),
url: url.map(str::to_string),
token_file: None,
via: Default::default(),
invite: None,
};
let mut c = base.clone();
c.upstreams = vec![peer("abc", None)];
assert!(SyncHub::new(c, d.path(), audit.clone(), Weak::new())
.err()
.unwrap()
.tagged()
.contains("64 hex digits"));
let mut c = base.clone();
c.downstreams = vec![peer(&"a".repeat(64), Some("http://x"))];
assert!(SyncHub::new(c, d.path(), audit.clone(), Weak::new())
.err()
.unwrap()
.tagged()
.contains("belong to an upstream"));
let hub = SyncHub::new(base, d.path(), audit, Weak::new()).unwrap();
assert!(
d.path().join(STATE_DIR).join(KEY_FILE).exists(),
"the key is minted on the first start"
);
assert_eq!(hub.public_key().len(), 64);
}
}