server/lib/src/estate_wire.rs
//! **The estate over the wire** (owner plan 2b, 2026-09-27) — the pure half.
//!
//! A twin — or a paired downstream acting as one — must hold what its primary
//! holds: the artifacts (holger's mirror already pulls those over
//! `RepositoryService`) AND the rest of the data root, [`crate::replication::ESTATE`]:
//! the config and every side store. Those were copied only disk to disk
//! (`replicate_estate`), so a twin on another machine could not hold them
//! (stinky 02 S2-1). `holger.estate.v1.EstateService` is the wire; this module
//! is what both ends do, with no socket in it:
//!
//! * [`manifest`] — every estate file under a data root, with size and blake3;
//! * [`plan`] — against what the follower holds, which files to fetch;
//! * [`write_fetched`] — put one fetched file in place, atomically, and only if
//! its bytes are the ones the manifest named;
//! * [`safe_path`] — the one rule that keeps a path the other end sent from
//! leaving the estate.
//!
//! The same calls the other way round RECHARGE a new main from its twin: the
//! new main pulls the twin's estate before it serves.
use std::path::{Path, PathBuf};
use crate::replication::{walk_files, ESTATE};
/// The largest single fetch the server answers (a follower loops by offset).
pub const MAX_FETCH_BYTES: u64 = 4 * 1024 * 1024;
/// One estate file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EstateFile {
/// Relative to the data root, `/`-separated.
pub path: String,
pub size: u64,
/// blake3 hex of its bytes.
pub blake3: String,
}
/// What a data root holds of the estate.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Manifest {
pub files: Vec<EstateFile>,
/// Estate items this root does not have at all (a normal state).
pub absent: Vec<String>,
}
fn hash_file(p: &Path) -> Result<String, String> {
let bytes = std::fs::read(p).map_err(|e| format!("{}: {e}", p.display()))?;
Ok(blake3::hash(&bytes).to_hex().to_string())
}
/// **Every estate file under `root`**, in a stable order.
pub fn manifest(root: &Path) -> Result<Manifest, String> {
let mut m = Manifest::default();
for item in ESTATE {
let at = root.join(item.path);
if item.is_file {
match std::fs::metadata(&at) {
Ok(meta) if meta.is_file() => m.files.push(EstateFile {
path: item.path.to_string(),
size: meta.len(),
blake3: hash_file(&at)?,
}),
_ => m.absent.push(item.path.to_string()),
}
continue;
}
if !at.is_dir() {
m.absent.push(item.path.to_string());
continue;
}
for (abs, rel, size) in walk_files(&at).map_err(|e| format!("{}: {e}", at.display()))? {
let rel = rel
.components()
.map(|c| c.as_os_str().to_string_lossy().to_string())
.collect::<Vec<_>>()
.join("/");
m.files.push(EstateFile {
path: format!("{}/{rel}", item.path),
size,
blake3: hash_file(&abs)?,
});
}
}
Ok(m)
}
/// **A path the other end sent, kept inside the estate** — relative, no `..`,
/// no root, and under one of [`ESTATE`]'s items (the file item exactly, or
/// below a store item). `Err` names why.
pub fn safe_path(path: &str) -> Result<PathBuf, String> {
let p = traits::safe_path::sanitize_under(Path::new(""), path).map_err(|e| {
format!(
"estate path `{path}` is not a plain relative path: {}",
e.why
)
})?;
let under = ESTATE.iter().any(|i| {
if i.is_file {
path == i.path
} else {
path.starts_with(&format!("{}/", i.path))
}
});
if !under {
return Err(format!(
"`{path}` is not part of the estate ({})",
ESTATE.iter().map(|i| i.path).collect::<Vec<_>>().join(", ")
));
}
Ok(p.to_path_buf())
}
/// **Which files to fetch**: every manifest file the follower does not hold
/// with the same blake3. A path that is not safe is an error, never skipped
/// silently. Files the follower holds that the primary does not are LEFT —
/// a follow never deletes (an append-only audit log must not be shortened by
/// a primary that lost a segment).
pub fn plan(m: &Manifest, local_root: &Path) -> Result<Vec<EstateFile>, String> {
let mut want = Vec::new();
for f in &m.files {
let rel = safe_path(&f.path)?;
let at = local_root.join(rel);
let same = std::fs::metadata(&at)
.map(|meta| meta.len() == f.size)
.unwrap_or(false)
&& hash_file(&at).map(|h| h == f.blake3).unwrap_or(false);
if !same {
want.push(f.clone());
}
}
Ok(want)
}
/// **Put one fetched file in place** — only when its bytes hash to what the
/// manifest named (a torn or tampered fetch is refused, the old file kept);
/// temporary file, fsync, rename.
pub fn write_fetched(local_root: &Path, f: &EstateFile, bytes: &[u8]) -> Result<(), String> {
let rel = safe_path(&f.path)?;
let got = blake3::hash(bytes).to_hex().to_string();
if got != f.blake3 || bytes.len() as u64 != f.size {
return Err(format!(
"`{}`: fetched {} bytes hashing to {got}, the manifest named {} bytes, {}",
f.path,
bytes.len(),
f.size,
f.blake3
));
}
let at = local_root.join(rel);
if let Some(dir) = at.parent() {
std::fs::create_dir_all(dir).map_err(|e| format!("{}: {e}", dir.display()))?;
}
let tmp = at.with_extension("estate-tmp");
{
use std::io::Write as _;
// HOL-NCSC (NCSC P3): the estate holds keys (the sync key, the passkey store, the
// WireGuard identity), so every fetched file is written 0600.
let mut o = std::fs::OpenOptions::new();
o.write(true).create(true).truncate(true);
#[cfg(unix)]
std::os::unix::fs::OpenOptionsExt::mode(&mut o, 0o600);
let mut w = o
.open(&tmp)
.map_err(|e| format!("{}: {e}", tmp.display()))?;
w.write_all(bytes)
.and_then(|_| w.sync_all())
.map_err(|e| format!("{}: {e}", tmp.display()))?;
}
std::fs::rename(&tmp, &at).map_err(|e| format!("{}: {e}", at.display()))
}
/// **Serve one byte range of one estate file** (the server's half of Fetch).
pub fn read_range(
root: &Path,
path: &str,
offset: u64,
max: u64,
) -> Result<(Vec<u8>, u64), String> {
let rel = safe_path(path)?;
let at = root.join(rel);
let bytes = std::fs::read(&at).map_err(|e| format!("{path}: {e}"))?;
let size = bytes.len() as u64;
let cap = if max == 0 {
MAX_FETCH_BYTES
} else {
max.min(MAX_FETCH_BYTES)
};
let start = offset.min(size) as usize;
let end = (offset.saturating_add(cap)).min(size) as usize;
Ok((bytes[start..end].to_vec(), size))
}
/// What one pull did.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Pulled {
pub fetched: usize,
pub fetched_bytes: u64,
pub already: usize,
pub failed: Vec<String>,
/// [`manifest_root`] of the SOURCE's manifest — set only when the pull was
/// whole, when every file it names is held here blake3-identical (an
/// `already` file matched by [`plan`], a fetched one checked by
/// [`write_fetched`]). Empty otherwise.
pub root: String,
/// How many files that manifest named.
pub files: usize,
}
/// The estate item whose files only ever GROW (the audit segments).
pub const APPEND_ONLY: &str = "holger-audit";
/// Is `path` (relative, `/`-separated) inside [`APPEND_ONLY`]?
pub fn in_append_only(path: &str) -> bool {
path == APPEND_ONLY
|| path
.strip_prefix(APPEND_ONLY)
.is_some_and(|r| r.starts_with('/'))
}
/// **One word for a whole manifest**: sha256 (hex) over its files sorted by
/// path, one line each, `path \t size \t blake3 \n`. A twin says it after a
/// whole pull (`holger.grow.v1.TwinReportRequest.estate_root`); anyone holding
/// the source's manifest (`EstateService.Manifest`) computes the same word and
/// compares — the functest's estate parity, with no door onto the twin.
///
/// **The append-only audit (`holger-audit/`) is left out of the word**: it grows
/// with every read the primary serves — the twin's own mirror downloads are
/// audited there — so a root over it could never agree between a pull and the
/// next reading. The twin still pulls it; its parity is its own hash chain.
pub fn manifest_root(files: &[EstateFile]) -> String {
use sha2::Digest as _;
let mut sorted: Vec<&EstateFile> = files.iter().filter(|f| !in_append_only(&f.path)).collect();
sorted.sort_by(|a, b| a.path.cmp(&b.path));
let mut h = sha2::Sha256::new();
for f in sorted {
h.update(format!("{}\t{}\t{}\n", f.path, f.size, f.blake3).as_bytes());
}
hex::encode(h.finalize())
}
/// The root and the count, when `r` is whole.
fn seal(mut r: Pulled, m: &Manifest) -> Pulled {
if r.failed.is_empty() {
r.root = manifest_root(&m.files);
r.files = m.files.len();
}
r
}
/// **How many bytes a pull assembles of `f`** while the source now holds `size`:
/// exactly what the manifest named — the prefix of a file that GREW since (the
/// primary's open audit segment grows with every request it serves) is the file
/// the manifest hashed; a file that shrank stops at what is there, and a file
/// that changed either way is refused by [`write_fetched`]'s hash.
fn named(f: &EstateFile, size: u64) -> u64 {
size.min(f.size)
}
/// **Pull an estate** through `fetch(path, offset) -> (bytes, size)` — the
/// wire in production, a local read in tests. Every file is assembled by
/// offset, checked against the manifest and written atomically.
/// `fetch(path, offset) -> (bytes, size)`, the wire or a local read.
pub type Fetch<'a> = dyn FnMut(&str, u64) -> Result<(Vec<u8>, u64), String> + 'a;
#[allow(clippy::needless_lifetimes)]
pub fn pull(m: &Manifest, local_root: &Path, fetch: &mut Fetch<'_>) -> Result<Pulled, String> {
let want = plan(m, local_root)?;
let mut r = Pulled {
already: m.files.len() - want.len(),
..Pulled::default()
};
for f in want {
let mut bytes = Vec::with_capacity(f.size as usize);
let outcome = loop {
match fetch(&f.path, bytes.len() as u64) {
Ok((chunk, size)) => {
bytes.extend_from_slice(&chunk);
if chunk.is_empty() || bytes.len() as u64 >= named(&f, size) {
bytes.truncate(f.size as usize);
break write_fetched(local_root, &f, &bytes);
}
}
Err(e) => break Err(e),
}
};
match outcome {
Ok(()) => {
r.fetched += 1;
r.fetched_bytes += f.size;
}
Err(e) => r.failed.push(e),
}
}
Ok(seal(r, m))
}
/// One request, with the admin-read bearer when there is one.
fn authed<T>(bearer: Option<&str>, body: T) -> Result<tonic::Request<T>, String> {
let mut r = tonic::Request::new(body);
if let Some(t) = bearer {
r.metadata_mut().insert(
"authorization",
format!("Bearer {t}")
.parse()
.map_err(|_| "the bearer is not a header value".to_string())?,
);
}
Ok(r)
}
/// **Pull a remote box's estate** over `holger.estate.v1` — a twin following
/// its primary, or a new main recharging from its twin. `endpoint` is
/// `http://host:port` (or https); `bearer` the admin-read credential the
/// remote accepts, when it gates one.
pub async fn pull_remote(
endpoint: &str,
bearer: Option<&str>,
local_root: &Path,
) -> Result<Pulled, String> {
crate::egress::check("estate pull", endpoint).map_err(|e| e.to_string())?;
let channel = tonic::transport::Channel::from_shared(endpoint.to_string())
.map_err(|e| format!("`{endpoint}` is not an endpoint: {e}"))?
.connect()
.await
.map_err(|e| format!("{endpoint}: {e}"))?;
pull_over(channel, endpoint, bearer, local_root).await
}
/// **Pull an estate over an already-dialled channel** — the replication door's
/// mutual TLS 1.3 channel (`crate::twin_tls::channel`) or a plain one.
/// `endpoint` labels the errors.
pub async fn pull_over(
channel: tonic::transport::Channel,
endpoint: &str,
bearer: Option<&str>,
local_root: &Path,
) -> Result<Pulled, String> {
use crate::grpc::holger_estate_proto::estate_service_client::EstateServiceClient;
use crate::grpc::holger_estate_proto::{FetchRequest, ManifestRequest};
let mut client =
EstateServiceClient::new(channel).max_decoding_message_size((MAX_FETCH_BYTES as usize) * 2);
let m = client
.manifest(authed(bearer, ManifestRequest {})?)
.await
.map_err(|s| format!("{endpoint} Manifest: {s}"))?
.into_inner();
let m = Manifest {
files: m
.files
.into_iter()
.map(|f| EstateFile {
path: f.path,
size: f.size,
blake3: f.blake3,
})
.collect(),
absent: m.absent,
};
let want = plan(&m, local_root)?;
let mut r = Pulled {
already: m.files.len() - want.len(),
..Pulled::default()
};
for f in want {
let mut bytes = Vec::with_capacity(f.size as usize);
let outcome = loop {
let req = authed(
bearer,
FetchRequest {
path: f.path.clone(),
offset: bytes.len() as u64,
max_bytes: 0,
},
)?;
match client.fetch(req).await {
Ok(resp) => {
let resp = resp.into_inner();
let done = resp.data.is_empty()
|| bytes.len() as u64 + resp.data.len() as u64 >= named(&f, resp.size);
bytes.extend_from_slice(&resp.data);
if done {
bytes.truncate(f.size as usize);
break write_fetched(local_root, &f, &bytes);
}
}
Err(s) => break Err(format!("{endpoint} Fetch {}: {s}", f.path)),
}
};
match outcome {
Ok(()) => {
r.fetched += 1;
r.fetched_bytes += f.size;
}
Err(e) => r.failed.push(e),
}
}
Ok(seal(r, &m))
}
#[cfg(test)]
mod tests {
use super::*;
fn root(name: &str) -> PathBuf {
let p = std::env::temp_dir().join(format!("estate-wire-{name}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&p);
std::fs::create_dir_all(&p).unwrap();
p
}
fn put(root: &Path, rel: &str, bytes: &[u8]) {
let p = root.join(rel);
std::fs::create_dir_all(p.parent().unwrap()).unwrap();
std::fs::write(p, bytes).unwrap();
}
#[test]
fn a_follower_pulls_what_differs_and_only_that_and_never_deletes() {
let (primary, twin) = (root("p"), root("t"));
put(&primary, "holger.ron", b"(repositories: [])");
put(
&primary,
"holger-audit/seg-000001.arrow",
&vec![7u8; 9_000_000],
);
put(&primary, "holger-passkeys/credentials.json", b"{}");
put(&primary, "not-the-estate/x", b"nope");
put(&twin, "holger-passkeys/credentials.json", b"{}");
put(
&twin,
"holger-audit/seg-000000.arrow",
b"the twin's own older segment",
);
let m = manifest(&primary).unwrap();
assert!(m
.files
.iter()
.all(|f| !f.path.starts_with("not-the-estate")));
assert!(m.absent.contains(&"holger-sboms".to_string()));
let mut calls = 0;
let r = pull(&m, &twin, &mut |p, off| {
calls += 1;
read_range(&primary, p, off, 0)
})
.unwrap();
assert_eq!((r.fetched, r.already, r.failed.len()), (2, 1, 0), "{r:?}");
assert!(
calls >= 4,
"the 9 MB segment comes in {} MB chunks",
MAX_FETCH_BYTES >> 20
);
assert_eq!(
manifest(&twin).unwrap().files.len(),
4,
"the twin keeps its own older segment"
);
assert_eq!(
std::fs::read(twin.join("holger.ron")).unwrap(),
b"(repositories: [])"
);
// Again: nothing moves.
let again = pull(&m, &twin, &mut |p, off| read_range(&primary, p, off, 0)).unwrap();
assert_eq!((again.fetched, again.already), (0, 3));
// Whole: the root of the SOURCE's manifest, which anyone holding that
// manifest computes the same — the twin's own older segment is not in it.
assert_eq!(
(r.root.as_str(), r.files),
(manifest_root(&m.files).as_str(), 3)
);
assert_eq!(again.root, r.root);
assert_eq!(r.root.len(), 64);
// Not whole: no root.
let broken = pull(&m, &root("t2"), &mut |_, _| {
Err("the source went away".into())
})
.unwrap();
assert!(broken.root.is_empty() && broken.files == 0, "{broken:?}");
}
#[test]
fn the_manifest_root_is_order_free_and_moves_with_any_byte() {
let f = |p: &str, b: &str| EstateFile {
path: p.into(),
size: 1,
blake3: b.into(),
};
let a = vec![f("holger.ron", "r"), f("holger-audit/1", "x")];
let b = vec![f("holger-audit/1", "x"), f("holger.ron", "r")];
assert_eq!(manifest_root(&a), manifest_root(&b));
assert_ne!(
manifest_root(&a),
manifest_root(&[f("holger.ron", "s"), f("holger-audit/1", "x")]),
"the config moved"
);
// ★ The AUDIT is append-only and grows with every read — the twin's own
// mirror downloads are audited on the primary — so a root over it can never
// agree at an instant (MEASURED on the mock 2026-09-29, dwarf: H5 "the twin
// holds the estate of root 96fe…; the primary's is 7cae…" for 608 s, 4 files
// each). The twin still pulls it; the root is the parity of the rest.
assert_eq!(
manifest_root(&a),
manifest_root(&[f("holger.ron", "r"), f("holger-audit/1", "y")]),
"an audit segment that grew is not a divergence"
);
assert_eq!(
manifest_root(&a),
manifest_root(&a[..1]),
"nor is one more segment"
);
// The spelling the ops functest recomputes: sha256 over `path\tsize\tblake3\n`, sorted, the audit left out.
use sha2::Digest as _;
let want = hex::encode(sha2::Sha256::digest(b"holger.ron\t1\tr\n"));
assert_eq!(manifest_root(&a), want);
}
/// ★ **An audit segment that GREW between the manifest and the fetch is not a
/// failure**: the primary appends to its open segment with every request it
/// serves, so the fetch read MORE bytes than the manifest named and the pull
/// refused the file — MEASURED on the dwarf KVM mock 2026-10-05 (ops rung 3,
/// H9): "0 fetched, 1 FAILED: `holger-audit/audit-…-1-0.arrows`: fetched 156864
/// bytes hashing to …, the manifest named 147520 bytes", the twin's last cycle
/// before the main died was not whole, and the recharge of the new main was
/// refused for ever ("Say accept_lost"). A pull assembles exactly the size the
/// manifest named: the prefix of a file that only grows is the file the
/// manifest hashed. A file that CHANGED is still refused by its hash.
#[test]
fn a_segment_that_grew_after_the_manifest_is_pulled_as_the_manifest_named_it() {
let (primary, twin) = (root("grew-p"), root("grew-t"));
put(&primary, "holger.ron", b"(repositories: [])");
let seg = "holger-audit/audit-1-1-0.arrows";
let before: Vec<u8> = (0..147_520u32).map(|i| (i % 251) as u8).collect();
put(&primary, seg, &before);
let m = manifest(&primary).unwrap();
// The primary serves more requests: its open segment grows.
let mut grown = before.clone();
grown.extend((0..9_344u32).map(|i| (i % 13) as u8));
put(&primary, seg, &grown);
let r = pull(&m, &twin, &mut |p, off| read_range(&primary, p, off, 0)).unwrap();
assert!(r.failed.is_empty(), "{r:?}");
assert_eq!((r.fetched, r.root.len()), (2, 64), "whole: {r:?}");
assert_eq!(
std::fs::read(twin.join(seg)).unwrap(),
before,
"the twin holds what the manifest named"
);
// In 4 MiB chunks too (a segment bigger than one fetch).
let (p2, t2) = (root("grew-p2"), root("grew-t2"));
let big = vec![3u8; (MAX_FETCH_BYTES as usize) + 1000];
put(&p2, seg, &big);
let m2 = manifest(&p2).unwrap();
put(&p2, seg, &[big.clone(), vec![4u8; 5000]].concat());
let r2 = pull(&m2, &t2, &mut |p, off| read_range(&p2, p, off, 0)).unwrap();
assert!(
r2.failed.is_empty() && std::fs::read(t2.join(seg)).unwrap() == big,
"{r2:?}"
);
// A file REWRITTEN since the manifest still fails by its hash.
let (p3, t3) = (root("grew-p3"), root("grew-t3"));
put(&p3, "holger.ron", b"(repositories: [])");
let m3 = manifest(&p3).unwrap();
put(&p3, "holger.ron", b"(repositories: [x])");
let r3 = pull(&m3, &t3, &mut |p, off| read_range(&p3, p, off, 0)).unwrap();
assert_eq!(r3.failed.len(), 1, "{r3:?}");
assert!(r3.root.is_empty());
}
#[test]
fn a_path_outside_the_estate_or_a_tampered_fetch_is_refused() {
for bad in [
"../etc/passwd",
"/etc/passwd",
"holger-audit/../../x",
"",
"not-the-estate/x",
"holger.ron/x",
] {
assert!(safe_path(bad).is_err(), "{bad}");
}
assert!(safe_path("holger-audit/seg-1").is_ok() && safe_path("holger.ron").is_ok());
let t = root("tamper");
let f = EstateFile {
path: "holger.ron".into(),
size: 3,
blake3: blake3::hash(b"abc").to_hex().to_string(),
};
assert!(write_fetched(&t, &f, b"abd")
.unwrap_err()
.contains("hashing to"));
assert!(!t.join("holger.ron").exists(), "nothing written");
write_fetched(&t, &f, b"abc").unwrap();
assert!(read_range(&t, "../x", 0, 0).is_err());
}
}