use std::io::{Cursor, Read, Write};
use std::path::{Path, PathBuf};
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
use ring::signature::{UnparsedPublicKey, ED25519};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use crate::store::Store;
pub const HUB_URL_ENV: &str = "DBMD_HUB_URL";
pub const HUB_KEY_ENV: &str = "DBMD_HUB_KEY";
pub const HUB_CREDENTIAL_ORIGIN_ENV: &str = "DBMD_HUB_CREDENTIAL_ORIGIN";
pub const STATE_DIR_ENV: &str = "DBMD_STATE_DIR";
pub const ALLOW_PRIVATE_REGISTRY_HOME_ENV: &str = "DBMD_ALLOW_PRIVATE_REGISTRY_HOME";
pub const ALLOW_PRIVATE_OBJECT_URL_ENV: &str = "DBMD_ALLOW_PRIVATE_OBJECT_URL";
pub const BRAIN_KEY_FILE_ENV: &str = "DBMD_BRAIN_KEY_FILE";
pub const AGENT_KEY_FILE_ENV: &str = "DBMD_AGENT_KEY_FILE";
pub const CONFIG_REL_PATH: &str = ".dbmd/config";
const MAX_RESPONSE_BYTES: u64 = 8 * 1024 * 1024;
const MAX_FEED_RESPONSE_BYTES: u64 = 16 * 1024 * 1024;
const MAX_REGISTRY_CARD_BYTES: u64 = 1024 * 1024;
const MAX_PUSH_BYTES: usize = 4 * 1024 * 1024;
const MAX_PUSH_FILES: usize = u16::MAX as usize;
const MAX_STORE_PATH_BYTES: usize = 1_024;
const MAX_STORE_BYTES: u64 = 512 * 1024 * 1024;
const MAX_PACK_BYTES: u64 =
MAX_STORE_BYTES + MAX_PUSH_FILES as u64 * (76 + 2 * MAX_STORE_PATH_BYTES) as u64 + 22;
const MAX_IDENTITY_ROTATIONS: usize = 1_024;
const MAX_FEED_REPLAY_ENTRIES: u64 = 100_000;
const MAX_FEED_REPLAY_BYTES: u64 = 64 * 1024 * 1024;
const FEED_PAGE_LIMIT: usize = 100;
pub const MAX_PROPOSE_BYTES: u64 = 16 * 1024;
const CONNECT_TIMEOUT_SECS: u64 = 10;
const READ_TIMEOUT_SECS: u64 = 120;
const OVERALL_REQUEST_TIMEOUT_SECS: u64 = 120;
const CONNECT_ATTEMPTS: usize = 3;
const CONNECT_RETRY_BACKOFF_MS: [u64; CONNECT_ATTEMPTS - 1] = [100, 300];
#[derive(Debug, thiserror::Error)]
pub enum LinkError {
#[error(
"no hub configured — pass --hub <URL>, set {HUB_URL_ENV}, or add `hub = <URL>` to {CONFIG_REL_PATH}"
)]
NoHub,
#[error("no hub credential — set {HUB_KEY_ENV} (credentials never live in {CONFIG_REL_PATH})")]
NoCredential,
#[error(
"the hub credential in {HUB_KEY_ENV} contains whitespace or non-ASCII characters — re-copy it (the key is not shown here on purpose)"
)]
BadKey,
#[error(
"refusing to send an ambient credential to the hub selected by {CONFIG_REL_PATH} — set {HUB_CREDENTIAL_ORIGIN_ENV} to that exact origin, or choose the hub explicitly with --hub/{HUB_URL_ENV}"
)]
UnboundCredential,
#[error("invalid agent signing key ({message}) — mint one with `dbmd key generate`")]
BadAgentKey {
message: String,
},
#[error("refusing non-HTTPS hub {hub} — the credential would travel in cleartext (localhost is exempt)")]
UnsafeHub {
hub: String,
},
#[error("hub unreachable at {hub}: {message}")]
Transport {
hub: String,
message: String,
},
#[error("{what} failed (HTTP {status}): {message}")]
Http {
what: &'static str,
status: u16,
message: String,
code: Option<String>,
},
#[error("{what}: the hub answered HTTP {status} with a non-JSON body — check the hub URL")]
NotJson {
what: &'static str,
status: u16,
},
#[error("hub response exceeded the {limit_bytes}-byte endpoint cap — refusing to buffer it")]
ResponseTooLarge {
limit_bytes: u64,
},
#[error("invalid address `{given}`: {reason}")]
BadAddress {
given: String,
reason: String,
},
#[error(
"invalid grant id `{given}` — grant ids come from `grant list` (lowercase letters, digits, hyphens)"
)]
BadGrantId {
given: String,
},
#[error("refusing unsafe path from the hub: `{path}`")]
UnsafePath {
path: String,
},
#[error(
"push too large ({detail}) — one snapshot caps at {} MB uncompressed, {} MB as a pack, and {MAX_PUSH_FILES} files",
MAX_STORE_BYTES / (1024 * 1024),
MAX_PACK_BYTES / (1024 * 1024)
)]
PushTooLarge {
detail: String,
},
#[error(
"propose body too large ({bytes} bytes) — the hub's inbox caps one submission at {} KB",
MAX_PROPOSE_BYTES / 1024
)]
ProposeTooLarge {
bytes: u64,
},
#[error("store file `{path}` is not valid UTF-8 — the JSON push path carries text only")]
NotUtf8 {
path: String,
},
#[error("invalid store pack: {message}")]
InvalidPack {
message: String,
},
#[error("invalid signed feed: {message}")]
InvalidFeed {
message: String,
},
#[error(
"{operation} is unavailable on this platform — use the official macOS/Linux build or WSL"
)]
UnsupportedPlatform {
operation: &'static str,
},
#[error(transparent)]
Io(#[from] std::io::Error),
#[error(transparent)]
Store(#[from] crate::StoreError),
}
pub type LinkResult<T> = std::result::Result<T, LinkError>;
fn require_hardened_filesystem(operation: &'static str) -> LinkResult<()> {
#[cfg(any(target_os = "linux", target_os = "macos"))]
{
let _ = operation;
Ok(())
}
#[cfg(not(any(target_os = "linux", target_os = "macos")))]
{
Err(LinkError::UnsupportedPlatform { operation })
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AddressTarget {
Id(String),
Path(String),
}
const BAD_BRAIN_REASON: &str =
"the brain reference must be a brain id (lowercase ULID) or a slug (lowercase letters, digits, hyphens)";
const BAD_TARGET_REASON: &str =
"the part after `/` must be a record id (lowercase ULID) or a store-relative `.md` path";
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Address {
pub brain: String,
pub target: Option<AddressTarget>,
}
impl Address {
pub fn parse(raw: &str) -> LinkResult<Address> {
let bad = |reason: &str| LinkError::BadAddress {
given: raw.to_string(),
reason: reason.to_string(),
};
let trimmed = raw.trim();
let body = trimmed.strip_prefix('@').unwrap_or(trimmed);
if body.is_empty() {
return Err(bad("empty address"));
}
let (brain, rest) = match body.split_once('/') {
Some((b, r)) => (b, Some(r)),
None => (body, None),
};
if brain.is_empty() {
return Err(bad("missing brain reference before `/`"));
}
if !is_safe_ref(brain) {
return Err(bad(BAD_BRAIN_REASON));
}
let target = match rest {
None => None,
Some("") => return Err(bad("trailing `/` with no record id or path")),
Some(r) if crate::ulid::is_ulid(r) => Some(AddressTarget::Id(r.to_string())),
Some(r) => {
if !safe_store_rel_path(r) || !r.ends_with(".md") {
return Err(bad(BAD_TARGET_REASON));
}
Some(AddressTarget::Path(r.to_string()))
}
};
Ok(Address {
brain: brain.to_string(),
target,
})
}
}
fn is_safe_ref(s: &str) -> bool {
!s.is_empty()
&& s.len() <= 64
&& s.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
}
pub fn is_valid_handle(s: &str) -> bool {
is_safe_ref(s)
}
pub fn safe_store_rel_path(p: &str) -> bool {
if p.is_empty() || p.len() > MAX_STORE_PATH_BYTES || p.starts_with('/') {
return false;
}
if !p
.bytes()
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-' | b'/'))
{
return false;
}
p.split('/')
.all(|seg| !seg.is_empty() && seg != "." && seg != ".." && !seg.starts_with('.'))
}
fn require_safe_ref(brain: &str) -> LinkResult<()> {
if is_safe_ref(brain) {
Ok(())
} else {
Err(LinkError::BadAddress {
given: brain.to_string(),
reason: BAD_BRAIN_REASON.to_string(),
})
}
}
fn require_valid_handle(handle: &str) -> LinkResult<()> {
if is_valid_handle(handle) {
Ok(())
} else {
Err(LinkError::BadAddress {
given: handle.to_string(),
reason: "the site handle must be lowercase letters, digits, hyphens".to_string(),
})
}
}
fn require_safe_grant_id(id: &str) -> LinkResult<()> {
if is_safe_ref(id) {
Ok(())
} else {
Err(LinkError::BadGrantId {
given: id.to_string(),
})
}
}
#[derive(Debug, Clone)]
pub struct HubConfig {
pub hub: String,
pub key: Option<String>,
pub agent_key: Option<AgentSigningKey>,
pub brain_key: Option<AgentSigningKey>,
pub state_dir: PathBuf,
store_selected: bool,
}
#[derive(Clone)]
pub struct AgentSigningKey {
pkcs8: Vec<u8>,
pub multikey: String,
pub public_key_spki: String,
}
impl std::fmt::Debug for AgentSigningKey {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AgentSigningKey")
.field("multikey", &self.multikey)
.field("pkcs8", &"<redacted>")
.finish()
}
}
impl HubConfig {
pub fn require_key(&self) -> LinkResult<&str> {
self.key.as_deref().ok_or(LinkError::NoCredential)
}
}
pub fn hub_config(flag_hub: Option<&str>, dir: &Path) -> LinkResult<HubConfig> {
let explicit_hub = flag_hub
.map(str::to_string)
.or_else(|| env_nonempty(HUB_URL_ENV));
let selected_by_store = explicit_hub.is_none();
let hub = explicit_hub
.or_else(|| config_file_hub(&dir.join(CONFIG_REL_PATH)))
.ok_or(LinkError::NoHub)?;
let hub = hub.trim().trim_end_matches('/').to_string();
assert_safe_hub(&hub)?;
if selected_by_store {
let parsed =
url::Url::parse(&hub).map_err(|_| LinkError::UnsafeHub { hub: hub.clone() })?;
if !parsed.scheme().eq_ignore_ascii_case("https")
|| (parsed.path() != "/" && !parsed.path().is_empty())
{
return Err(LinkError::UnsafeHub { hub });
}
}
let key = match env_nonempty(HUB_KEY_ENV) {
Some(raw) => Some(clean_key(&raw)?),
None => None,
};
let agent_key = match env_nonempty(AGENT_KEY_FILE_ENV) {
Some(path) => Some(load_agent_key(Path::new(&path))?),
None => None,
};
let brain_key = match env_nonempty(BRAIN_KEY_FILE_ENV) {
Some(path) => Some(load_agent_key(Path::new(&path))?),
None => None,
};
if selected_by_store && (key.is_some() || agent_key.is_some() || brain_key.is_some()) {
let bound = env_nonempty(HUB_CREDENTIAL_ORIGIN_ENV)
.and_then(|value| normalized_origin(&value).ok());
let selected_origin = normalized_origin(&hub)?;
if bound.as_deref() != Some(selected_origin.as_str()) {
return Err(LinkError::UnboundCredential);
}
}
Ok(HubConfig {
hub,
key,
agent_key,
brain_key,
state_dir: toolkit_state_dir()?,
store_selected: selected_by_store,
})
}
fn toolkit_state_dir() -> LinkResult<PathBuf> {
if let Some(path) = env_nonempty(STATE_DIR_ENV) {
let path = PathBuf::from(path);
if !path.is_absolute() {
return Err(LinkError::UnsafePath {
path: path.display().to_string(),
});
}
return Ok(path);
}
#[cfg(windows)]
if let Some(base) = env_nonempty("LOCALAPPDATA") {
let base = PathBuf::from(base);
if base.is_absolute() {
return Ok(base.join("dbmd").join("state"));
}
}
#[cfg(not(windows))]
if let Some(base) = env_nonempty("XDG_STATE_HOME") {
let base = PathBuf::from(base);
if base.is_absolute() {
return Ok(base.join("dbmd"));
}
}
let home = PathBuf::from(env_nonempty("HOME").ok_or_else(|| {
LinkError::Io(std::io::Error::new(
std::io::ErrorKind::NotFound,
format!("cannot locate user state; set {STATE_DIR_ENV}"),
))
})?);
if !home.is_absolute() {
return Err(LinkError::UnsafePath {
path: home.display().to_string(),
});
}
#[cfg(target_os = "macos")]
{
Ok(home
.join("Library")
.join("Application Support")
.join("dbmd")
.join("state"))
}
#[cfg(all(not(target_os = "macos"), not(windows)))]
{
Ok(home.join(".local").join("state").join("dbmd"))
}
}
fn normalized_origin(value: &str) -> LinkResult<String> {
let parsed = url::Url::parse(value).map_err(|_| LinkError::UnsafeHub {
hub: value.to_string(),
})?;
if !(parsed.scheme().eq_ignore_ascii_case("https")
|| parsed.scheme().eq_ignore_ascii_case("http"))
|| !parsed.username().is_empty()
|| parsed.password().is_some()
|| (parsed.path() != "/" && !parsed.path().is_empty())
|| parsed.query().is_some()
|| parsed.fragment().is_some()
{
return Err(LinkError::UnsafeHub {
hub: value.to_string(),
});
}
let host = parsed.host_str().ok_or_else(|| LinkError::UnsafeHub {
hub: value.to_string(),
})?;
let host = if host.contains(':') {
format!("[{host}]")
} else {
host.to_ascii_lowercase()
};
let port = parsed
.port_or_known_default()
.ok_or_else(|| LinkError::UnsafeHub {
hub: value.to_string(),
})?;
let default = (parsed.scheme().eq_ignore_ascii_case("https") && port == 443)
|| (parsed.scheme().eq_ignore_ascii_case("http") && port == 80);
Ok(format!(
"{}://{}{}",
parsed.scheme().to_ascii_lowercase(),
host,
if default {
String::new()
} else {
format!(":{port}")
}
))
}
const ED25519_SPKI_PREFIX: [u8; 12] = [
0x30, 0x2a, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x03, 0x21, 0x00,
];
fn bad_agent_key(message: &str) -> LinkError {
LinkError::BadAgentKey {
message: message.to_string(),
}
}
fn agent_keypair(pkcs8: &[u8]) -> LinkResult<ring::signature::Ed25519KeyPair> {
ring::signature::Ed25519KeyPair::from_pkcs8(pkcs8)
.or_else(|_| ring::signature::Ed25519KeyPair::from_pkcs8_maybe_unchecked(pkcs8))
.map_err(|_| bad_agent_key("not an Ed25519 PKCS#8 key"))
}
fn public_identity_for(pair: &ring::signature::Ed25519KeyPair) -> (String, String) {
use ring::signature::KeyPair as _;
let mut spki = Vec::with_capacity(44);
spki.extend_from_slice(&ED25519_SPKI_PREFIX);
spki.extend_from_slice(pair.public_key().as_ref());
(
URL_SAFE_NO_PAD.encode(&spki),
format!("ed25519:{}", URL_SAFE_NO_PAD.encode(Sha256::digest(&spki))),
)
}
pub fn load_signing_key(path: &Path) -> LinkResult<AgentSigningKey> {
load_agent_key(path)
}
fn load_agent_key(path: &Path) -> LinkResult<AgentSigningKey> {
#[cfg(unix)]
let file = {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
use std::os::unix::ffi::OsStrExt as _;
let parent = open_existing_dir_nofollow(path.parent().unwrap_or_else(|| Path::new(".")))
.map_err(|e| bad_agent_key(&format!("cannot open the key parent: {e}")))?;
let leaf = path
.file_name()
.ok_or_else(|| bad_agent_key("the key path has no file name"))?;
let leaf = c_name(leaf.as_bytes(), &path.display().to_string())?;
let fd = unsafe {
libc::openat(
parent.as_raw_fd(),
leaf.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if fd < 0 {
return Err(bad_agent_key(
"the key path must be an existing regular file without symlink ancestors",
));
}
unsafe { std::fs::File::from_raw_fd(fd) }
};
#[cfg(not(unix))]
let file = std::fs::File::open(path)
.map_err(|e| bad_agent_key(&format!("cannot read the key file: {e}")))?;
let metadata = file
.metadata()
.map_err(|e| bad_agent_key(&format!("cannot inspect the key file: {e}")))?;
if !metadata.is_file() {
return Err(bad_agent_key("the key path must be a regular file"));
}
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
if metadata.permissions().mode() & 0o077 != 0 {
return Err(bad_agent_key(
"the key file is accessible to group/other; set mode 0600",
));
}
}
let mut text = String::new();
file.take(1024 * 1024 + 1)
.read_to_string(&mut text)
.map_err(|e| bad_agent_key(&format!("cannot read the key file: {e}")))?;
if text.len() > 1024 * 1024 {
return Err(bad_agent_key("the key file exceeds the size limit"));
}
let pkcs8 = URL_SAFE_NO_PAD
.decode(text.trim())
.map_err(|_| bad_agent_key("the key file is not one base64url line"))?;
let (public_key_spki, multikey) = public_identity_for(&agent_keypair(&pkcs8)?);
Ok(AgentSigningKey {
pkcs8,
multikey,
public_key_spki,
})
}
fn write_secret_new(path: &Path, bytes: &[u8]) -> LinkResult<()> {
#[cfg(unix)]
let (mut file, parent, leaf) = {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
use std::os::unix::ffi::OsStrExt as _;
let parent = open_or_create_dir_nofollow(path.parent().unwrap_or_else(|| Path::new(".")))?;
let leaf_name = path
.file_name()
.ok_or_else(|| bad_agent_key("the output key path has no file name"))?;
let leaf = c_name(leaf_name.as_bytes(), &path.display().to_string())?;
let fd = unsafe {
libc::openat(
parent.as_raw_fd(),
leaf.as_ptr(),
libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC | libc::O_NOFOLLOW,
0o600,
)
};
if fd < 0 {
let error = std::io::Error::last_os_error();
if error.kind() == std::io::ErrorKind::AlreadyExists {
return Err(bad_agent_key(
"the output file already exists — refusing to overwrite a key",
));
}
return Err(error.into());
}
(unsafe { std::fs::File::from_raw_fd(fd) }, parent, leaf)
};
#[cfg(not(unix))]
let mut file = std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
.map_err(|error| {
if error.kind() == std::io::ErrorKind::AlreadyExists {
bad_agent_key("the output file already exists — refusing to overwrite a key")
} else {
LinkError::Io(error)
}
})?;
if let Err(error) = file.write_all(bytes).and_then(|_| file.sync_all()) {
drop(file);
#[cfg(unix)]
let _ =
unsafe { libc::unlinkat(std::os::fd::AsRawFd::as_raw_fd(&parent), leaf.as_ptr(), 0) };
#[cfg(not(unix))]
let _ = std::fs::remove_file(path);
return Err(LinkError::Io(error));
}
drop(file);
#[cfg(unix)]
parent.sync_all()?;
Ok(())
}
#[derive(Debug, Serialize)]
pub struct GeneratedAgentKey {
pub multikey: String,
#[serde(rename = "publicKeySpki")]
pub public_key_spki: String,
#[serde(rename = "keyFile")]
pub key_file: String,
}
pub fn generate_agent_key(out: &Path) -> LinkResult<GeneratedAgentKey> {
require_hardened_filesystem("key generation")?;
let rng = ring::rand::SystemRandom::new();
let pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng)
.map_err(|_| bad_agent_key("key generation failed"))?;
let pair = agent_keypair(pkcs8.as_ref())?;
let (spki_b64u, multikey) = public_identity_for(&pair);
write_secret_new(
out,
format!("{}\n", URL_SAFE_NO_PAD.encode(pkcs8.as_ref())).as_bytes(),
)?;
Ok(GeneratedAgentKey {
multikey,
public_key_spki: spki_b64u,
key_file: out.display().to_string(),
})
}
fn linkmd_sig_header(
key: &AgentSigningKey,
origin: &str,
method: &str,
path: &str,
body: Option<&str>,
) -> LinkResult<String> {
let ts = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_err(|_| bad_agent_key("system clock is before the epoch"))?
.as_secs();
let body_hash = match body {
Some(b) => format!("{:x}", Sha256::digest(b.as_bytes())),
None => "-".to_string(),
};
let canonical = format!(
"v2\n{}\n{}\n{}\n{}\n{}",
origin,
method.to_uppercase(),
path,
ts,
body_hash
);
let pair = agent_keypair(&key.pkcs8)?;
let sig = URL_SAFE_NO_PAD.encode(pair.sign(canonical.as_bytes()).as_ref());
let fingerprint = key.multikey.trim_start_matches("ed25519:");
Ok(format!(
"LinkMD-Sig v2,key=ed25519:{fingerprint},ts={ts},sig={sig}"
))
}
#[derive(Serialize)]
struct WireFeedFile {
path: String,
sha256: String,
bytes: u64,
}
#[derive(Serialize)]
struct UnsignedWireEntry<'a> {
v: u8,
seq: u64,
ts: String,
brain: &'a str,
public_key: &'a str,
kind: &'a str,
op: &'a str,
pack_sha256: &'a str,
files: &'a [WireFeedFile],
removed: &'a [String],
prev_entry_hash: Option<&'a str>,
}
fn self_custody_entry(
key: &AgentSigningKey,
seq: u64,
ts: String,
pack_sha256: &str,
files: &[WireFeedFile],
prev_entry_hash: Option<&str>,
) -> LinkResult<String> {
let removed: [String; 0] = [];
let unsigned = serde_json::to_string(&UnsignedWireEntry {
v: 1,
seq,
ts,
brain: &key.multikey,
public_key: &key.public_key_spki,
kind: "push",
op: "snapshot",
pack_sha256,
files,
removed: &removed,
prev_entry_hash,
})
.expect("serialize feed entry");
let pair = agent_keypair(&key.pkcs8)?;
let sig = URL_SAFE_NO_PAD.encode(pair.sign(unsigned.as_bytes()).as_ref());
Ok(format!(
"{},\"sig\":\"{}\"}}",
&unsigned[..unsigned.len() - 1],
sig
))
}
fn env_nonempty(name: &str) -> Option<String> {
std::env::var(name).ok().filter(|v| !v.trim().is_empty())
}
fn config_file_hub(path: &Path) -> Option<String> {
const MAX_CONFIG_BYTES: u64 = 64 * 1024;
#[cfg(unix)]
let file = {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
use std::os::unix::ffi::OsStrExt as _;
let parent =
open_existing_dir_nofollow(path.parent().unwrap_or_else(|| Path::new("."))).ok()?;
let leaf = c_name(path.file_name()?.as_bytes(), &path.display().to_string()).ok()?;
let fd = unsafe {
libc::openat(
parent.as_raw_fd(),
leaf.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if fd < 0 {
return None;
}
unsafe { std::fs::File::from_raw_fd(fd) }
};
#[cfg(not(unix))]
let file = std::fs::File::open(path).ok()?;
let metadata = file.metadata().ok()?;
if !metadata.is_file() || metadata.len() > MAX_CONFIG_BYTES {
return None;
}
let mut bytes = Vec::with_capacity(metadata.len() as usize);
file.take(MAX_CONFIG_BYTES + 1)
.read_to_end(&mut bytes)
.ok()?;
if bytes.len() as u64 > MAX_CONFIG_BYTES {
return None;
}
let text = String::from_utf8(bytes).ok()?;
for line in text.lines() {
let line = line.trim();
if line.is_empty() || line.starts_with('#') {
continue;
}
if let Some((k, v)) = line.split_once('=') {
if k.trim() == "hub" {
let v = v.trim();
if !v.is_empty() {
return Some(v.to_string());
}
}
}
}
None
}
fn assert_safe_hub(hub: &str) -> LinkResult<()> {
let parsed = url::Url::parse(hub).map_err(|_| LinkError::UnsafeHub {
hub: hub.to_string(),
})?;
if !(parsed.scheme().eq_ignore_ascii_case("https")
|| parsed.scheme().eq_ignore_ascii_case("http"))
|| !parsed.username().is_empty()
|| parsed.password().is_some()
|| (parsed.path() != "/" && !parsed.path().is_empty())
|| parsed.query().is_some()
|| parsed.fragment().is_some()
{
return Err(LinkError::UnsafeHub {
hub: hub.to_string(),
});
}
let loopback = match parsed.host() {
Some(url::Host::Domain(host)) => host.eq_ignore_ascii_case("localhost"),
Some(url::Host::Ipv4(ip)) => ip.is_loopback(),
Some(url::Host::Ipv6(ip)) => ip.is_loopback(),
None => false,
};
if parsed.scheme().eq_ignore_ascii_case("https") || loopback {
Ok(())
} else {
Err(LinkError::UnsafeHub {
hub: hub.to_string(),
})
}
}
fn clean_key(raw: &str) -> LinkResult<String> {
let k = raw.trim();
if k.is_empty() || k.bytes().any(|b| !(0x21..=0x7e).contains(&b)) {
return Err(LinkError::BadKey);
}
Ok(k.to_string())
}
#[derive(Debug)]
pub struct HubResponse {
pub status: u16,
pub body: Option<Value>,
}
struct RawHubResponse {
status: u16,
body: Vec<u8>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Auth {
Required,
None,
Optional,
}
fn agent_builder_with_timeout(overall: std::time::Duration) -> ureq::AgentBuilder {
ureq::AgentBuilder::new()
.user_agent(concat!("dbmd/", env!("CARGO_PKG_VERSION")))
.redirects(0)
.timeout_connect(std::time::Duration::from_secs(CONNECT_TIMEOUT_SECS))
.timeout_read(std::time::Duration::from_secs(READ_TIMEOUT_SECS))
.timeout_write(overall)
.timeout(overall)
}
fn agent_builder() -> ureq::AgentBuilder {
agent_builder_with_timeout(std::time::Duration::from_secs(OVERALL_REQUEST_TIMEOUT_SECS))
}
fn agent() -> ureq::Agent {
agent_builder().build()
}
fn hub_agent(cfg: &HubConfig) -> LinkResult<ureq::Agent> {
if !cfg.store_selected {
return Ok(agent());
}
let parsed = url::Url::parse(&cfg.hub).map_err(|_| LinkError::UnsafeHub {
hub: cfg.hub.clone(),
})?;
pinned_public_agent(&parsed, false, "store-selected hub")
}
fn request_raw(
cfg: &HubConfig,
method: &str,
path: &str,
body: Option<&Value>,
auth: Auth,
max_response_bytes: u64,
) -> LinkResult<RawHubResponse> {
let url = format!("{}{}", cfg.hub, path);
let encoded_body = body.map(Value::to_string);
let origin = normalized_origin(&cfg.hub)?;
let credential = match auth {
Auth::Required => Some(match &cfg.agent_key {
Some(key) => linkmd_sig_header(key, &origin, method, path, encoded_body.as_deref())?,
None => format!("Bearer {}", cfg.require_key()?),
}),
Auth::Optional => match &cfg.agent_key {
Some(key) => Some(linkmd_sig_header(
key,
&origin,
method,
path,
encoded_body.as_deref(),
)?),
None => cfg.key.as_deref().map(|k| format!("Bearer {k}")),
},
Auth::None => None,
};
let http = hub_agent(cfg)?;
let result = with_connect_retries(|| {
let mut req = http.request(method, &url);
if let Some(value) = &credential {
req = req.set("authorization", value);
}
match &encoded_body {
Some(value) => req
.set("content-type", "application/json")
.send_string(value)
.map_err(Box::new),
None => req.call().map_err(Box::new),
}
});
let resp = match result {
Ok(resp) => resp,
Err(error) => match *error {
ureq::Error::Status(_, resp) => resp,
ureq::Error::Transport(error) => {
return Err(LinkError::Transport {
hub: cfg.hub.clone(),
message: error.to_string(),
});
}
},
};
let status = resp.status();
let mut buf = Vec::new();
resp.into_reader()
.take(max_response_bytes + 1)
.read_to_end(&mut buf)?;
if buf.len() as u64 > max_response_bytes {
return Err(LinkError::ResponseTooLarge {
limit_bytes: max_response_bytes,
});
}
Ok(RawHubResponse { status, body: buf })
}
fn request_capped(
cfg: &HubConfig,
method: &str,
path: &str,
body: Option<&Value>,
auth: Auth,
max_response_bytes: u64,
) -> LinkResult<HubResponse> {
let raw = request_raw(cfg, method, path, body, auth, max_response_bytes)?;
let parsed: Option<Value> = serde_json::from_slice(&raw.body).ok();
Ok(HubResponse {
status: raw.status,
body: parsed,
})
}
fn request(
cfg: &HubConfig,
method: &str,
path: &str,
body: Option<&Value>,
auth: Auth,
) -> LinkResult<HubResponse> {
request_capped(cfg, method, path, body, auth, MAX_RESPONSE_BYTES)
}
fn ensure_raw_ok(r: RawHubResponse, what: &'static str) -> LinkResult<Vec<u8>> {
if (200..300).contains(&r.status) {
return Ok(r.body);
}
ensure_ok(
HubResponse {
status: r.status,
body: serde_json::from_slice(&r.body).ok(),
},
what,
)
.and_then(|_| Err(invalid_feed("a non-2xx response was accepted unexpectedly")))
}
fn is_pre_request_transport(kind: ureq::ErrorKind) -> bool {
matches!(
kind,
ureq::ErrorKind::Dns | ureq::ErrorKind::ConnectionFailed | ureq::ErrorKind::ProxyConnect
)
}
fn with_connect_retries(
mut send: impl FnMut() -> Result<ureq::Response, Box<ureq::Error>>,
) -> Result<ureq::Response, Box<ureq::Error>> {
let mut attempt = 0;
loop {
match send() {
Err(error)
if matches!(
error.as_ref(),
ureq::Error::Transport(transport)
if is_pre_request_transport(transport.kind())
) && attempt + 1 < CONNECT_ATTEMPTS =>
{
std::thread::sleep(std::time::Duration::from_millis(
CONNECT_RETRY_BACKOFF_MS[attempt],
));
attempt += 1;
}
result => return result,
}
}
}
fn hub_is_loopback(hub: &str) -> bool {
url::Url::parse(hub).ok().is_some_and(|parsed| {
parsed.host().is_some_and(|host| match host {
url::Host::Domain(host) => host.eq_ignore_ascii_case("localhost"),
url::Host::Ipv4(ip) => ip.is_loopback(),
url::Host::Ipv6(ip) => ip.is_loopback(),
})
})
}
fn presigned_agent(cfg: &HubConfig, raw: &str) -> LinkResult<ureq::Agent> {
let parsed = url::Url::parse(raw).map_err(|_| LinkError::InvalidPack {
message: "the hub returned an invalid object-store URL".to_string(),
})?;
let allow_private = hub_is_loopback(&cfg.hub)
|| env_nonempty(ALLOW_PRIVATE_OBJECT_URL_ENV).as_deref() == Some("1");
if (!parsed.scheme().eq_ignore_ascii_case("https") && !allow_private)
|| !parsed.username().is_empty()
|| parsed.password().is_some()
|| parsed.fragment().is_some()
{
return Err(LinkError::InvalidPack {
message: "the hub returned an unsafe object-store URL".to_string(),
});
}
pinned_public_agent(&parsed, allow_private, "object-store URL").map_err(|_| {
LinkError::InvalidPack {
message: "the hub returned an object-store URL with an unsafe network target"
.to_string(),
}
})
}
fn put_presigned(cfg: &HubConfig, raw: &str, headers: &Value, bytes: &[u8]) -> LinkResult<()> {
let http = presigned_agent(cfg, raw)?;
let result = with_connect_retries(|| {
let mut req = http.put(raw);
if let Some(map) = headers.as_object() {
for (name, value) in map {
if let Some(value) = value.as_str() {
req = req.set(name, value);
}
}
}
req.send_bytes(bytes).map_err(Box::new)
});
match result {
Ok(resp) if (200..300).contains(&resp.status()) => Ok(()),
Ok(resp) => Err(LinkError::Http {
what: "pack upload",
status: resp.status(),
message: "object store rejected the upload".to_string(),
code: None,
}),
Err(error) => match *error {
ureq::Error::Status(_, resp) => Err(LinkError::Http {
what: "pack upload",
status: resp.status(),
message: "object store rejected the upload".to_string(),
code: None,
}),
ureq::Error::Transport(err) => Err(LinkError::Transport {
hub: "the object store".to_string(),
message: err.to_string(),
}),
},
}
}
fn one_past_bounded_limit(max_bytes: u64) -> Option<u64> {
max_bytes.checked_add(1)
}
fn presigned_download_read_limit() -> u64 {
one_past_bounded_limit(MAX_PACK_BYTES)
.expect("the fixed presigned-download ceiling must leave room for the refusal byte")
}
fn get_presigned(cfg: &HubConfig, raw: &str) -> LinkResult<Vec<u8>> {
let http = presigned_agent(cfg, raw)?;
let resp = match with_connect_retries(|| http.get(raw).call().map_err(Box::new)) {
Ok(resp) => resp,
Err(error) => match *error {
ureq::Error::Status(_, resp) => {
return Err(LinkError::Http {
what: "pack download",
status: resp.status(),
message: "object store rejected the download".to_string(),
code: None,
});
}
ureq::Error::Transport(err) => {
return Err(LinkError::Transport {
hub: "the object store".to_string(),
message: err.to_string(),
});
}
},
};
if !(200..300).contains(&resp.status()) {
return Err(LinkError::Http {
what: "pack download",
status: resp.status(),
message: "object store rejected the download".to_string(),
code: None,
});
}
let mut bytes = Vec::new();
resp.into_reader()
.take(presigned_download_read_limit())
.read_to_end(&mut bytes)?;
if bytes.len() as u64 > MAX_PACK_BYTES {
return Err(LinkError::InvalidPack {
message: "download exceeds the compressed-size limit".to_string(),
});
}
Ok(bytes)
}
fn ensure_ok(r: HubResponse, what: &'static str) -> LinkResult<Value> {
if !(200..300).contains(&r.status) {
let message = r
.body
.as_ref()
.and_then(|b| b.get("error"))
.and_then(Value::as_str)
.unwrap_or("unknown error")
.to_string();
let code = r
.body
.as_ref()
.and_then(|b| b.get("code"))
.and_then(Value::as_str)
.map(str::to_string);
return Err(LinkError::Http {
what,
status: r.status,
message,
code,
});
}
r.body.ok_or(LinkError::NotJson {
what,
status: r.status,
})
}
fn is_public_registry_ip(ip: std::net::IpAddr) -> bool {
match ip {
std::net::IpAddr::V4(ip) => {
let [a, b, c, _] = ip.octets();
!(a == 0
|| a == 10
|| a == 127
|| (a == 100 && (64..=127).contains(&b))
|| (a == 169 && b == 254)
|| (a == 172 && (16..=31).contains(&b))
|| (a == 192 && b == 0 && c == 0)
|| (a == 192 && b == 0 && c == 2)
|| (a == 192 && b == 88 && c == 99)
|| (a == 192 && b == 168)
|| (a == 198 && (b == 18 || b == 19))
|| (a == 198 && b == 51 && c == 100)
|| (a == 203 && b == 0 && c == 113)
|| a >= 224)
}
std::net::IpAddr::V6(ip) => {
let segments = ip.segments();
(segments[0] & 0xe000) == 0x2000
&& !(segments[0] == 0x2001 && (segments[1] & 0xfe00) == 0)
&& !(segments[0] == 0x2001 && segments[1] == 0x0db8)
&& segments[0] != 0x2002
&& !(segments[0] == 0x3fff && (segments[1] & 0xf000) == 0)
}
}
}
#[derive(Clone)]
struct PinnedRegistryResolver {
netloc: String,
addresses: Vec<std::net::SocketAddr>,
}
impl ureq::Resolver for PinnedRegistryResolver {
fn resolve(&self, requested: &str) -> std::io::Result<Vec<std::net::SocketAddr>> {
if requested == self.netloc {
Ok(self.addresses.clone())
} else {
Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"registry request attempted to resolve an unvalidated authority",
))
}
}
}
fn pinned_public_agent(
url: &url::Url,
allow_private: bool,
label: &str,
) -> LinkResult<ureq::Agent> {
let host = url
.host_str()
.ok_or_else(|| invalid_feed(format!("{label} has no host")))?;
let port = url
.port_or_known_default()
.ok_or_else(|| invalid_feed(format!("{label} has no port")))?;
let addresses = resolve_addresses_with_deadline(
host,
port,
std::time::Duration::from_secs(CONNECT_TIMEOUT_SECS),
)
.map_err(|error| invalid_feed(format!("{label} DNS resolution failed: {error}")))?;
if addresses.is_empty() {
return Err(invalid_feed(format!("{label} DNS returned no addresses")));
}
if !allow_private
&& addresses
.iter()
.any(|address| !is_public_registry_ip(address.ip()))
{
return Err(invalid_feed(format!(
"{label} resolves to a non-public address"
)));
}
let netloc = if host.contains(':') {
format!("[{host}]:{port}")
} else {
format!("{host}:{port}")
};
Ok(agent_builder()
.resolver(PinnedRegistryResolver { netloc, addresses })
.build())
}
fn resolve_addresses_with_deadline(
host: &str,
port: u16,
timeout: std::time::Duration,
) -> std::io::Result<Vec<std::net::SocketAddr>> {
use std::net::ToSocketAddrs as _;
let host = host.to_string();
let (send, receive) = std::sync::mpsc::sync_channel(1);
std::thread::Builder::new()
.name("dbmd-dns".to_string())
.spawn(move || {
let result = (host.as_str(), port)
.to_socket_addrs()
.map(|addresses| addresses.collect());
let _ = send.send(result);
})
.map_err(|error| std::io::Error::other(format!("cannot start resolver: {error}")))?;
match receive.recv_timeout(timeout) {
Ok(result) => result,
Err(std::sync::mpsc::RecvTimeoutError::Timeout) => Err(std::io::Error::new(
std::io::ErrorKind::TimedOut,
"resolution exceeded its deadline",
)),
Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => Err(std::io::Error::other(
"resolver stopped without returning a result",
)),
}
}
fn registry_agent(url: &url::Url) -> LinkResult<ureq::Agent> {
let allow_private = env_nonempty(ALLOW_PRIVATE_REGISTRY_HOME_ENV).as_deref() == Some("1");
pinned_public_agent(url, allow_private, "registry home")
}
fn get_json_absolute(url: &str) -> LinkResult<Value> {
let parsed = url::Url::parse(url).map_err(|_| invalid_feed("invalid registry home URL"))?;
let allow_private = env_nonempty(ALLOW_PRIVATE_REGISTRY_HOME_ENV).as_deref() == Some("1");
if (!parsed.scheme().eq_ignore_ascii_case("https") && !allow_private)
|| !parsed.username().is_empty()
|| parsed.password().is_some()
|| parsed.query().is_some()
|| parsed.fragment().is_some()
{
return Err(invalid_feed("unsafe registry home URL"));
}
let http = registry_agent(&parsed)?;
let resp = match with_connect_retries(|| http.get(url).call().map_err(Box::new)) {
Ok(resp) => resp,
Err(error) => match *error {
ureq::Error::Status(status, resp) => {
let _ = resp;
return Err(LinkError::Http {
what: "registry home fetch",
status,
message: "the home node rejected the card request".to_string(),
code: None,
});
}
ureq::Error::Transport(err) => {
return Err(LinkError::Transport {
hub: url.to_string(),
message: err.to_string(),
});
}
},
};
if !(200..300).contains(&resp.status()) {
return Err(LinkError::Http {
what: "registry home fetch",
status: resp.status(),
message: "the home node returned a redirect or error".to_string(),
code: None,
});
}
let mut buf = Vec::new();
resp.into_reader()
.take(MAX_REGISTRY_CARD_BYTES + 1)
.read_to_end(&mut buf)?;
if buf.len() as u64 > MAX_REGISTRY_CARD_BYTES {
return Err(LinkError::ResponseTooLarge {
limit_bytes: MAX_REGISTRY_CARD_BYTES,
});
}
serde_json::from_slice(&buf).map_err(|_| LinkError::InvalidFeed {
message: "the home node returned invalid JSON".to_string(),
})
}
pub fn resolve_registry(cfg: &HubConfig, handle: &str) -> LinkResult<Option<Value>> {
require_safe_ref(handle)?;
let trust_directory = open_trust_dir(cfg)?;
let reg = request_capped(
cfg,
"GET",
&format!("/api/hub/registry/{handle}"),
None,
Auth::None,
MAX_REGISTRY_CARD_BYTES,
)?;
if reg.status == 404 {
return Ok(None);
}
let body = ensure_ok(reg, "registry resolve")?;
let home = body
.get("home")
.and_then(Value::as_str)
.ok_or_else(|| invalid_feed("registry entry has no home"))?;
let brain = body
.get("brain")
.and_then(Value::as_str)
.ok_or_else(|| invalid_feed("registry entry has no brain"))?;
if !crate::ulid::is_ulid(brain) {
return Err(invalid_feed(
"registry entry brain is not a canonical lowercase ULID",
));
}
let want_fp = body
.get("identity")
.and_then(|i| i.get("fingerprint"))
.and_then(Value::as_str)
.ok_or_else(|| invalid_feed("registry entry has no identity fingerprint"))?;
let home = home.trim_end_matches('/');
let origin = normalized_origin(home)?;
if origin != home {
return Err(invalid_feed(
"registry home must be an origin without a path, query, or fragment",
));
}
let _trust_locks = lock_trust_many(cfg, &trust_directory, &[handle, brain])?;
let (pinned, alias_binding) = load_canonical_pin(cfg, &trust_directory, handle, brain)?;
if let Some(binding) = &alias_binding {
if binding
.home
.as_deref()
.is_some_and(|pinned_home| pinned_home != home)
{
return Err(invalid_feed(
"registry relocated a pinned handle to a different home",
));
}
}
let card = get_json_absolute(&format!("{home}/api/hub/brains/{brain}"))?;
if card.get("id").and_then(Value::as_str) != Some(brain) {
return Err(invalid_feed(
"the home node served a card for a different brain",
));
}
let identity: FeedIdentity = serde_json::from_value(
card.get("identity")
.cloned()
.ok_or_else(|| invalid_feed("the home node served no identity"))?,
)
.map_err(|_| invalid_feed("the home node served an invalid identity"))?;
let anchor = verify_identity_chain(&identity, pinned.as_ref())?;
let got_fp = card
.get("identity")
.and_then(|i| i.get("fingerprint"))
.and_then(Value::as_str)
.unwrap_or_default();
if got_fp != want_fp {
return Err(invalid_feed(
"the home node served an identity that does not match the registry — refusing",
));
}
let current = format!("ed25519:{}", identity.fingerprint);
let advertised_seq = card
.get("headSeq")
.and_then(Value::as_u64)
.ok_or_else(|| invalid_feed("the home node served an invalid head sequence"))?;
let advertised_hash = card.get("feedHash").and_then(Value::as_str);
if (advertised_seq == 0 && !card.get("feedHash").is_some_and(Value::is_null))
|| (advertised_seq > 0 && advertised_hash.is_none_or(|hash| !is_sha256(hash)))
{
return Err(invalid_feed(
"the home node served an invalid feed head boundary",
));
}
verify_rotation_feed_boundaries(&identity, pinned.as_ref(), &[], advertised_seq)?;
let registry_alias = AliasBinding {
v: 1,
origin: normalized_origin(&cfg.hub)?,
requested: handle.to_string(),
brain: brain.to_string(),
home: Some(home.to_string()),
};
save_canonical_pin_and_alias(
cfg,
&trust_directory,
handle,
brain,
TrustState {
v: 2,
origin: normalized_origin(&cfg.hub)?,
requested: brain.to_string(),
brain: brain.to_string(),
home: None,
anchor,
current,
head_seq: pinned.as_ref().map_or(0, |checkpoint| checkpoint.head_seq),
feed_hash: pinned
.as_ref()
.and_then(|checkpoint| checkpoint.feed_hash.clone()),
rotations: identity.rotations.clone(),
},
Some(®istry_alias),
)?;
let mut out = card;
if let Value::Object(map) = &mut out {
map.insert("home".to_string(), Value::String(home.to_string()));
map.insert(
"resolvedVia".to_string(),
Value::String("registry".to_string()),
);
}
Ok(Some(out))
}
pub fn resolve(cfg: &HubConfig, addr: &Address) -> LinkResult<Value> {
require_safe_ref(&addr.brain)?;
if let Some(target) = &addr.target {
let (given, ok) = match target {
AddressTarget::Id(id) => (id, crate::ulid::is_ulid(id)),
AddressTarget::Path(p) => (p, safe_store_rel_path(p) && p.ends_with(".md")),
};
if !ok {
return Err(LinkError::BadAddress {
given: given.clone(),
reason: BAD_TARGET_REASON.to_string(),
});
}
}
if let Some(target) = &addr.target {
let remote = verified_remote_head(cfg, &addr.brain, false)?;
if !remote.head.verified {
return Err(invalid_feed(
"a path-scoped feed cannot prove a record against the full signed snapshot",
));
}
if remote.head.seq == 0 {
return Err(LinkError::Http {
what: "resolve",
status: 404,
message: "record not found".to_string(),
code: Some("NOT_FOUND".to_string()),
});
}
let brain = remote.head.brain.clone();
let pack = download_verified_snapshot_pack(cfg, &brain, &remote)?;
return resolve_from_verified_pack(&brain, target, pack);
}
let path = format!("/api/hub/brains/{}", addr.brain);
let direct = request(cfg, "GET", &path, None, Auth::Required)?;
if direct.status == 404 && addr.target.is_none() && !crate::ulid::is_ulid(&addr.brain) {
if let Some(card) = resolve_registry(cfg, &addr.brain)? {
return Ok(card);
}
}
let resolved = ensure_ok(direct, "resolve")?;
let remote = verified_remote_head(cfg, &addr.brain, false)?;
if resolved.get("id").and_then(Value::as_str) != Some(remote.head.brain.as_str())
|| resolved.get("headSeq").and_then(Value::as_u64) != Some(remote.head.seq)
|| resolved.get("feedHash").and_then(Value::as_str) != remote.head.feed_hash.as_deref()
{
return Err(invalid_feed(
"resolve card is not bound to the exact verified feed checkpoint",
));
}
let card_identity: FeedIdentity = serde_json::from_value(
resolved
.get("identity")
.cloned()
.ok_or_else(|| invalid_feed("resolve card has no signed identity"))?,
)
.map_err(|_| invalid_feed("resolve card has an invalid identity"))?;
if remote.identity.as_ref() != Some(&card_identity) {
return Err(invalid_feed(
"resolve card identity differs from the verified feed identity",
));
}
Ok(resolved)
}
fn resolve_from_verified_pack(
brain: &str,
target: &AddressTarget,
pack: Vec<u8>,
) -> LinkResult<Value> {
let entries = parse_store_pack(pack)?;
let mut matched: Option<(String, Vec<u8>)> = None;
for (path, bytes) in entries {
let is_candidate = match target {
AddressTarget::Path(want) => &path == want,
AddressTarget::Id(_) => {
path.ends_with(".md")
&& (path.starts_with("records/") || path.starts_with("sources/"))
}
};
if !is_candidate {
continue;
}
let text = std::str::from_utf8(&bytes)
.map_err(|_| invalid_feed(format!("signed snapshot record `{path}` is not UTF-8")))?;
let parsed = crate::parser::split_frontmatter(text, Path::new(&path))
.map_err(|_| invalid_feed(format!("signed snapshot record `{path}` is malformed")))?;
if let AddressTarget::Id(want) = target {
let frontmatter =
crate::parser::Frontmatter::parse(&parsed.frontmatter_yaml, Path::new(&path))
.map_err(|_| {
invalid_feed(format!("signed snapshot record `{path}` is malformed"))
})?;
if frontmatter.id.as_deref() != Some(want) {
continue;
}
}
if matched.is_some() {
return Err(invalid_feed(
"signed snapshot contains more than one record for the requested target",
));
}
matched = Some((path, bytes));
}
let (path, bytes) = matched.ok_or_else(|| LinkError::Http {
what: "resolve",
status: 404,
message: "record not found".to_string(),
code: Some("NOT_FOUND".to_string()),
})?;
let text = std::str::from_utf8(&bytes)
.map_err(|_| invalid_feed(format!("signed snapshot record `{path}` is not UTF-8")))?;
let parsed = crate::parser::split_frontmatter(text, Path::new(&path))
.map_err(|_| invalid_feed(format!("signed snapshot record `{path}` is malformed")))?;
let frontmatter: Value = serde_norway::from_str(&parsed.frontmatter_yaml)
.map_err(|_| invalid_feed(format!("signed snapshot record `{path}` is malformed")))?;
let Value::Object(fields) = frontmatter else {
return Err(invalid_feed(format!(
"signed snapshot record `{path}` frontmatter is not a mapping"
)));
};
let mut document = serde_json::Map::new();
document.insert("path".to_string(), Value::String(path));
for (key, value) in fields {
document.insert(key, value);
}
document.insert("body".to_string(), Value::String(parsed.body));
document.insert(
"contentSha".to_string(),
Value::String(content_sha256(&bytes)),
);
Ok(json!({
"brain": brain,
"document": Value::Object(document),
}))
}
#[derive(Debug, serde::Serialize)]
pub struct PullReport {
pub brain: String,
pub slug: String,
#[serde(rename = "headSeq")]
pub head_seq: u64,
pub files: usize,
pub dest: String,
#[serde(rename = "extraLocal")]
pub extra_local: Vec<String>,
}
fn download_verified_snapshot_pack(
cfg: &HubConfig,
brain: &str,
remote: &VerifiedRemote,
) -> LinkResult<Vec<u8>> {
let feed_hash = remote
.head
.feed_hash
.as_deref()
.ok_or_else(|| invalid_feed("non-empty snapshot has no verified feed hash"))?;
let signed_head = remote
.head_entry
.as_ref()
.ok_or_else(|| invalid_feed("verified snapshot has no signed head entry"))?;
let expected = &signed_head.entry.pack_sha256;
if !is_sha256(expected) {
return Err(invalid_feed(
"signed head carries an invalid snapshot pack digest",
));
}
let path = format!(
"/api/hub/brains/{brain}/export?format=pack&atSeq={}&feedHash={feed_hash}",
remote.head.seq
);
let body = ensure_ok(
request(cfg, "GET", &path, None, Auth::Required)?,
"sync pull",
)?;
if body.get("headSeq").and_then(Value::as_u64) != Some(remote.head.seq)
|| body.get("feedHash").and_then(Value::as_str) != Some(feed_hash)
|| body.get("brain").and_then(Value::as_str) != Some(remote.head.brain.as_str())
|| body.get("sha256").and_then(Value::as_str) != Some(expected.as_str())
{
return Err(invalid_feed(
"export response is not bound to the exact verified snapshot",
));
}
let url = body
.get("url")
.and_then(Value::as_str)
.ok_or_else(|| invalid_feed("verified snapshot export carried no exact pack URL"))?;
let bytes = get_presigned(cfg, url)?;
if content_sha256(&bytes) != *expected {
return Err(LinkError::InvalidPack {
message: "downloaded pack does not match the signed snapshot digest".to_string(),
});
}
let entries = parse_store_pack(bytes.clone())?;
if signed_head.entry.kind == "push" {
verify_snapshot_manifest(&entries, &signed_head.entry.files)?;
}
Ok(bytes)
}
pub fn sync_pull(cfg: &HubConfig, brain: &str, out: Option<&Path>) -> LinkResult<PullReport> {
require_hardened_filesystem("sync pull")?;
require_safe_ref(brain)?;
let remote = verified_remote_head(cfg, brain, false)?;
if !remote.head.verified {
return Err(invalid_feed(
"the grant exposes head movement but no signed snapshot; refusing an unverifiable pull",
));
}
let snapshot_hash = remote.head.feed_hash.as_deref().unwrap_or("none");
let path = format!(
"/api/hub/brains/{brain}/export?format=pack&atSeq={}&feedHash={snapshot_hash}",
remote.head.seq
);
let body = ensure_ok(
request(cfg, "GET", &path, None, Auth::Required)?,
"sync pull",
)?;
if body.get("headSeq").and_then(Value::as_u64) != Some(remote.head.seq)
|| body.get("feedHash").and_then(Value::as_str) != remote.head.feed_hash.as_deref()
{
return Err(invalid_feed(
"export response is not bound to the verified snapshot token",
));
}
let remote_slug = body
.get("slug")
.and_then(Value::as_str)
.filter(|slug| is_safe_slug(slug));
let slug = remote_slug
.or_else(|| is_safe_slug(brain).then_some(brain))
.unwrap_or("brain")
.to_string();
let brain_id = body
.get("brain")
.and_then(Value::as_str)
.unwrap_or(&remote.head.brain)
.to_string();
if brain_id != remote.head.brain {
return Err(invalid_feed(
"export response names a different brain than the verified head",
));
}
let head_seq = remote.head.seq;
let dest: PathBuf = match out {
Some(p) => p.to_path_buf(),
None => PathBuf::from(&slug),
};
let entries = if head_seq == 0 {
let files = body
.get("files")
.and_then(Value::as_array)
.ok_or_else(|| invalid_feed("empty snapshot export did not carry an empty manifest"))?;
if !files.is_empty() || body.get("url").is_some() {
return Err(invalid_feed(
"empty signed feed cannot authorize non-empty exported content",
));
}
Vec::new()
} else {
let signed_head = remote
.head_entry
.as_ref()
.ok_or_else(|| invalid_feed("verified snapshot has no signed head entry"))?;
let expected = &signed_head.entry.pack_sha256;
if !is_sha256(expected) {
return Err(invalid_feed(
"signed head carries an invalid snapshot pack digest",
));
}
if let Some(url) = body.get("url").and_then(Value::as_str) {
if body.get("sha256").and_then(Value::as_str) != Some(expected.as_str()) {
return Err(invalid_feed(
"export pack digest does not match the signed head entry",
));
}
let bytes = get_presigned(cfg, url)?;
let actual = format!("{:x}", Sha256::digest(&bytes));
if actual != *expected {
return Err(LinkError::InvalidPack {
message: "downloaded pack does not match the signed snapshot digest"
.to_string(),
});
}
let entries = parse_store_pack(bytes)?;
if signed_head.entry.kind == "push" {
verify_snapshot_manifest(&entries, &signed_head.entry.files)?;
}
entries
} else {
if signed_head.entry.kind != "push" {
return Err(invalid_feed(
"delta snapshots must export the exact signed pack",
));
}
let files = body.get("files").and_then(Value::as_array).ok_or_else(|| {
invalid_feed("verified snapshot export carried neither a pack nor files")
})?;
let mut entries = Vec::with_capacity(files.len());
for file in files {
let path = file
.get("path")
.and_then(Value::as_str)
.ok_or_else(|| invalid_feed("exported file path is not a string"))?;
let content = file
.get("content")
.and_then(Value::as_str)
.ok_or_else(|| invalid_feed("exported inline content is not UTF-8 text"))?;
entries.push((path.to_string(), content.as_bytes().to_vec()));
}
verify_snapshot_manifest(&entries, &signed_head.entry.files)?;
entries
}
};
let mut seen = std::collections::HashSet::new();
for (path, _) in &entries {
if !safe_store_rel_path(path) {
return Err(LinkError::UnsafePath { path: path.clone() });
}
if !seen.insert(path) {
return Err(LinkError::InvalidPack {
message: format!("duplicate path `{path}`"),
});
}
}
let pulled: std::collections::BTreeSet<&str> =
entries.iter().map(|(p, _)| p.as_str()).collect();
let mut extra_local = Vec::new();
if let Ok(store) = Store::open(&dest) {
if let Ok(walked) = store.walk() {
for rel in walked {
let rel_str = rel.to_string_lossy().replace('\\', "/");
if !pulled.contains(rel_str.as_str()) {
extra_local.push(rel_str);
}
}
}
}
#[cfg(unix)]
install_pulled_snapshot(&dest, &entries)?;
Ok(PullReport {
brain: brain_id,
slug,
head_seq,
files: entries.len(),
dest: dest.to_string_lossy().into_owned(),
extra_local,
})
}
#[cfg(unix)]
fn c_name(bytes: &[u8], display: &str) -> LinkResult<std::ffi::CString> {
std::ffi::CString::new(bytes).map_err(|_| LinkError::UnsafePath {
path: display.to_string(),
})
}
#[cfg(unix)]
fn open_dir_at(
parent: std::os::fd::RawFd,
name: &std::ffi::CStr,
display: &str,
) -> LinkResult<std::fs::File> {
use std::os::fd::FromRawFd as _;
let fd = unsafe {
libc::openat(
parent,
name.as_ptr(),
libc::O_RDONLY | libc::O_DIRECTORY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if fd < 0 {
return Err(LinkError::UnsafePath {
path: display.to_string(),
});
}
Ok(unsafe { std::fs::File::from_raw_fd(fd) })
}
#[cfg(unix)]
fn open_dir_path_nofollow(path: &Path, create: bool) -> LinkResult<std::fs::File> {
use std::os::fd::AsRawFd as _;
#[cfg(target_os = "macos")]
let normalized = [("/var", "/private/var"), ("/tmp", "/private/tmp")]
.into_iter()
.find_map(|(alias, real)| {
path.strip_prefix(alias)
.ok()
.map(|rest| Path::new(real).join(rest))
})
.unwrap_or_else(|| path.to_path_buf());
#[cfg(not(target_os = "macos"))]
let normalized = path.to_path_buf();
let start = if normalized.is_absolute() {
std::fs::File::open("/")?
} else {
std::fs::File::open(".")?
};
let mut directory = start;
for component in normalized.components() {
use std::path::Component;
let name = match component {
Component::RootDir | Component::CurDir => continue,
Component::Normal(name) => name,
Component::ParentDir | Component::Prefix(_) => {
return Err(LinkError::UnsafePath {
path: path.display().to_string(),
});
}
};
use std::os::unix::ffi::OsStrExt as _;
let name = c_name(name.as_bytes(), &path.display().to_string())?;
if create {
let made = unsafe { libc::mkdirat(directory.as_raw_fd(), name.as_ptr(), 0o700) };
if made != 0 {
let error = std::io::Error::last_os_error();
if error.raw_os_error() != Some(libc::EEXIST) {
return Err(error.into());
}
}
}
directory = open_dir_at(directory.as_raw_fd(), &name, &path.display().to_string())?;
}
Ok(directory)
}
#[cfg(unix)]
fn open_or_create_dir_nofollow(path: &Path) -> LinkResult<std::fs::File> {
open_dir_path_nofollow(path, true)
}
#[cfg(unix)]
fn open_existing_dir_nofollow(path: &Path) -> LinkResult<std::fs::File> {
open_dir_path_nofollow(path, false)
}
#[cfg(unix)]
fn entry_is_dir_at(parent: std::os::fd::RawFd, name: &std::ffi::CStr) -> LinkResult<Option<bool>> {
let mut stat: libc::stat = unsafe { std::mem::zeroed() };
let result =
unsafe { libc::fstatat(parent, name.as_ptr(), &mut stat, libc::AT_SYMLINK_NOFOLLOW) };
if result == 0 {
return Ok(Some((stat.st_mode & libc::S_IFMT) == libc::S_IFDIR));
}
let error = std::io::Error::last_os_error();
if error.kind() == std::io::ErrorKind::NotFound {
Ok(None)
} else {
Err(error.into())
}
}
#[cfg(unix)]
fn create_dir_exclusive_at(
parent: std::os::fd::RawFd,
name: &std::ffi::CStr,
display: &str,
) -> LinkResult<std::fs::File> {
let made = unsafe { libc::mkdirat(parent, name.as_ptr(), 0o700) };
if made != 0 {
return Err(LinkError::UnsafePath {
path: display.to_string(),
});
}
open_dir_at(parent, name, display)
}
#[cfg(unix)]
fn directory_entry_names(directory: &std::fs::File) -> LinkResult<Vec<std::ffi::CString>> {
use std::os::fd::AsRawFd as _;
let duplicate = unsafe { libc::dup(directory.as_raw_fd()) };
if duplicate < 0 {
return Err(std::io::Error::last_os_error().into());
}
let stream = unsafe { libc::fdopendir(duplicate) };
if stream.is_null() {
let error = std::io::Error::last_os_error();
unsafe {
libc::close(duplicate);
}
return Err(error.into());
}
let mut names = Vec::new();
loop {
let entry = unsafe { libc::readdir(stream) };
if entry.is_null() {
break;
}
let raw = unsafe { std::ffi::CStr::from_ptr((*entry).d_name.as_ptr()) };
if raw.to_bytes() != b"." && raw.to_bytes() != b".." {
names.push(raw.to_owned());
}
}
if unsafe { libc::closedir(stream) } != 0 {
return Err(std::io::Error::last_os_error().into());
}
Ok(names)
}
#[cfg(unix)]
fn remove_tree_at(
parent: std::os::fd::RawFd,
name: &std::ffi::CStr,
display: &str,
) -> LinkResult<()> {
use std::os::fd::AsRawFd as _;
match entry_is_dir_at(parent, name)? {
None => return Ok(()),
Some(false) => {
if unsafe { libc::unlinkat(parent, name.as_ptr(), 0) } != 0 {
return Err(std::io::Error::last_os_error().into());
}
}
Some(true) => {
let directory = open_dir_at(parent, name, display)?;
for child in directory_entry_names(&directory)? {
let child_display =
format!("{display}/{}", String::from_utf8_lossy(child.to_bytes()));
remove_tree_at(directory.as_raw_fd(), &child, &child_display)?;
}
drop(directory);
if unsafe { libc::unlinkat(parent, name.as_ptr(), libc::AT_REMOVEDIR) } != 0 {
return Err(std::io::Error::last_os_error().into());
}
}
}
Ok(())
}
#[cfg(unix)]
fn clone_tree_contents(
source: &std::fs::File,
destination: &std::fs::File,
display: &str,
) -> LinkResult<()> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
for name in directory_entry_names(source)? {
let child_display = format!("{display}/{}", String::from_utf8_lossy(name.to_bytes()));
let mut stat: libc::stat = unsafe { std::mem::zeroed() };
if unsafe {
libc::fstatat(
source.as_raw_fd(),
name.as_ptr(),
&mut stat,
libc::AT_SYMLINK_NOFOLLOW,
)
} != 0
{
return Err(std::io::Error::last_os_error().into());
}
match stat.st_mode & libc::S_IFMT {
libc::S_IFDIR => {
if unsafe {
libc::mkdirat(destination.as_raw_fd(), name.as_ptr(), stat.st_mode & 0o777)
} != 0
{
return Err(std::io::Error::last_os_error().into());
}
let source_child = open_dir_at(source.as_raw_fd(), &name, &child_display)?;
let destination_child =
open_dir_at(destination.as_raw_fd(), &name, &child_display)?;
clone_tree_contents(&source_child, &destination_child, &child_display)?;
destination_child.sync_all()?;
}
libc::S_IFREG => {
let source_fd = unsafe {
libc::openat(
source.as_raw_fd(),
name.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if source_fd < 0 {
return Err(std::io::Error::last_os_error().into());
}
let destination_fd = unsafe {
libc::openat(
destination.as_raw_fd(),
name.as_ptr(),
libc::O_WRONLY
| libc::O_CREAT
| libc::O_EXCL
| libc::O_CLOEXEC
| libc::O_NOFOLLOW,
(stat.st_mode & 0o777) as libc::c_uint,
)
};
if destination_fd < 0 {
unsafe {
libc::close(source_fd);
}
return Err(std::io::Error::last_os_error().into());
}
let mut input = unsafe { std::fs::File::from_raw_fd(source_fd) };
let mut output = unsafe { std::fs::File::from_raw_fd(destination_fd) };
std::io::copy(&mut input, &mut output)?;
output.sync_all()?;
}
libc::S_IFLNK => {
let mut target = vec![0_u8; 4097];
let length = unsafe {
libc::readlinkat(
source.as_raw_fd(),
name.as_ptr(),
target.as_mut_ptr().cast(),
target.len(),
)
};
if length < 0 || length as usize >= target.len() {
return Err(LinkError::UnsafePath {
path: child_display,
});
}
target.truncate(length as usize);
let target = c_name(&target, &child_display)?;
if unsafe {
libc::symlinkat(target.as_ptr(), destination.as_raw_fd(), name.as_ptr())
} != 0
{
return Err(std::io::Error::last_os_error().into());
}
}
_ => {
return Err(LinkError::UnsafePath {
path: child_display,
});
}
}
}
destination.sync_all()?;
Ok(())
}
#[cfg(target_os = "linux")]
fn install_stage_at(
parent: std::os::fd::RawFd,
stage: &std::ffi::CStr,
dest: &std::ffi::CStr,
dest_exists: bool,
) -> LinkResult<()> {
let flags = if dest_exists {
libc::RENAME_EXCHANGE
} else {
libc::RENAME_NOREPLACE
};
let result = unsafe {
libc::syscall(
libc::SYS_renameat2,
parent,
stage.as_ptr(),
parent,
dest.as_ptr(),
flags,
)
};
if result == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error().into())
}
}
#[cfg(target_os = "macos")]
fn install_stage_at(
parent: std::os::fd::RawFd,
stage: &std::ffi::CStr,
dest: &std::ffi::CStr,
dest_exists: bool,
) -> LinkResult<()> {
let flags = if dest_exists {
libc::RENAME_SWAP
} else {
libc::RENAME_EXCL
};
let result =
unsafe { libc::renameatx_np(parent, stage.as_ptr(), parent, dest.as_ptr(), flags) };
if result == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error().into())
}
}
#[cfg(unix)]
fn write_pull_entries_beneath_dir(
root: &std::fs::File,
entries: &[(String, Vec<u8>)],
) -> LinkResult<()> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
for (path, content) in entries {
let components: Vec<&str> = path.split('/').collect();
let (leaf, parents) = components
.split_last()
.ok_or_else(|| LinkError::UnsafePath { path: path.clone() })?;
let mut directory = root.try_clone()?;
for component in parents {
let name = c_name(component.as_bytes(), path)?;
let made = unsafe { libc::mkdirat(directory.as_raw_fd(), name.as_ptr(), 0o700) };
if made != 0 {
let error = std::io::Error::last_os_error();
if error.raw_os_error() != Some(libc::EEXIST) {
return Err(error.into());
}
}
directory = open_dir_at(directory.as_raw_fd(), &name, path)?;
}
let leaf_name = c_name(leaf.as_bytes(), path)?;
let mut existing: libc::stat = unsafe { std::mem::zeroed() };
let inspected = unsafe {
libc::fstatat(
directory.as_raw_fd(),
leaf_name.as_ptr(),
&mut existing,
libc::AT_SYMLINK_NOFOLLOW,
)
};
if inspected == 0 && (existing.st_mode & libc::S_IFMT) == libc::S_IFLNK {
return Err(LinkError::UnsafePath { path: path.clone() });
}
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let temp_name = format!(
".dbmd-pull-{}-{nonce}-{}",
std::process::id(),
content_sha256(format!("{path}\0{}", content.len()).as_bytes())
);
let temp = c_name(temp_name.as_bytes(), path)?;
let fd = unsafe {
libc::openat(
directory.as_raw_fd(),
temp.as_ptr(),
libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC | libc::O_NOFOLLOW,
0o600,
)
};
if fd < 0 {
return Err(std::io::Error::last_os_error().into());
}
let mut file = unsafe { std::fs::File::from_raw_fd(fd) };
if let Err(error) = file.write_all(content).and_then(|_| file.sync_all()) {
let _ = unsafe { libc::unlinkat(directory.as_raw_fd(), temp.as_ptr(), 0) };
return Err(error.into());
}
drop(file);
let renamed = unsafe {
libc::renameat(
directory.as_raw_fd(),
temp.as_ptr(),
directory.as_raw_fd(),
leaf_name.as_ptr(),
)
};
if renamed != 0 {
let error = std::io::Error::last_os_error();
let _ = unsafe { libc::unlinkat(directory.as_raw_fd(), temp.as_ptr(), 0) };
return Err(error.into());
}
directory.sync_all()?;
}
root.sync_all()?;
Ok(())
}
#[cfg(unix)]
fn install_pulled_snapshot(dest: &Path, entries: &[(String, Vec<u8>)]) -> LinkResult<()> {
use ring::rand::SecureRandom as _;
use std::os::fd::AsRawFd as _;
use std::os::unix::ffi::OsStrExt as _;
let parent = dest.parent().unwrap_or_else(|| Path::new("."));
let name = dest
.file_name()
.filter(|name| !name.is_empty() && *name != "." && *name != "..")
.ok_or_else(|| LinkError::UnsafePath {
path: dest.display().to_string(),
})?;
let parent_dir = open_or_create_dir_nofollow(parent)?;
let dest_name = c_name(name.as_bytes(), &dest.display().to_string())?;
let dest_exists = match entry_is_dir_at(parent_dir.as_raw_fd(), &dest_name)? {
None => false,
Some(true) => true,
Some(false) => {
return Err(LinkError::UnsafePath {
path: dest.display().to_string(),
});
}
};
let mut nonce = [0_u8; 16];
ring::rand::SystemRandom::new()
.fill(&mut nonce)
.map_err(|_| invalid_feed("could not mint a pull staging name"))?;
let stage_label = format!(
".{}.dbmd-pull-stage-{}",
name.to_string_lossy(),
URL_SAFE_NO_PAD.encode(nonce)
);
let stage_name = c_name(stage_label.as_bytes(), &dest.display().to_string())?;
let stage_dir = create_dir_exclusive_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
)?;
let prepared = (|| -> LinkResult<()> {
if dest_exists {
let live = open_dir_at(
parent_dir.as_raw_fd(),
&dest_name,
&dest.display().to_string(),
)?;
clone_tree_contents(&live, &stage_dir, &dest.display().to_string())?;
}
write_pull_entries_beneath_dir(&stage_dir, entries)?;
stage_dir.sync_all()?;
Ok(())
})();
if let Err(error) = prepared {
let _ = remove_tree_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
);
return Err(error);
}
if let Err(error) =
install_stage_at(parent_dir.as_raw_fd(), &stage_name, &dest_name, dest_exists)
{
let _ = remove_tree_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
);
return Err(error);
}
parent_dir.sync_all()?;
if dest_exists {
let _ = remove_tree_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
);
let _ = parent_dir.sync_all();
}
Ok(())
}
fn is_safe_slug(slug: &str) -> bool {
!slug.is_empty()
&& slug.len() <= 63
&& !slug.starts_with('-')
&& !slug.ends_with('-')
&& slug
.bytes()
.all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
}
fn le_u16(bytes: &[u8], at: usize) -> Option<u16> {
Some(u16::from_le_bytes(bytes.get(at..at + 2)?.try_into().ok()?))
}
fn le_u32(bytes: &[u8], at: usize) -> Option<u32> {
Some(u32::from_le_bytes(bytes.get(at..at + 4)?.try_into().ok()?))
}
fn le_u64(bytes: &[u8], at: usize) -> Option<u64> {
Some(u64::from_le_bytes(bytes.get(at..at + 8)?.try_into().ok()?))
}
fn preflight_zip_central_directory(
bytes: &[u8],
offset: usize,
size: usize,
count: u64,
) -> LinkResult<()> {
const CENTRAL_ENTRY_SIG: &[u8; 4] = b"PK\x01\x02";
let end = offset
.checked_add(size)
.filter(|end| *end <= bytes.len())
.ok_or_else(|| LinkError::InvalidPack {
message: "ZIP central directory is out of bounds".to_string(),
})?;
let mut cursor = offset;
for _ in 0..count {
if bytes.get(cursor..cursor + 4) != Some(CENTRAL_ENTRY_SIG.as_slice()) {
return Err(LinkError::InvalidPack {
message: "ZIP central directory entry count is inconsistent".to_string(),
});
}
if le_u16(bytes, cursor + 34) != Some(0) {
return Err(LinkError::InvalidPack {
message: "multi-disk ZIP archives are not supported".to_string(),
});
}
let variable = [28, 30, 32].into_iter().try_fold(0usize, |total, at| {
total.checked_add(le_u16(bytes, cursor + at)? as usize)
});
cursor = cursor
.checked_add(46)
.and_then(|fixed| fixed.checked_add(variable?))
.filter(|cursor| *cursor <= end)
.ok_or_else(|| LinkError::InvalidPack {
message: "ZIP central directory entry is truncated".to_string(),
})?;
}
if cursor != end {
return Err(LinkError::InvalidPack {
message: "ZIP central directory size is inconsistent".to_string(),
});
}
Ok(())
}
fn preflight_zip_entry_count(bytes: &[u8], max_entries: usize) -> LinkResult<()> {
const EOCD_SIG: &[u8; 4] = b"PK\x05\x06";
const ZIP64_LOCATOR_SIG: &[u8; 4] = b"PK\x06\x07";
const ZIP64_EOCD_SIG: &[u8; 4] = b"PK\x06\x06";
let search_start = bytes.len().saturating_sub(22 + u16::MAX as usize);
let eocd = bytes[search_start..]
.windows(4)
.rposition(|window| window == EOCD_SIG)
.map(|offset| search_start + offset)
.ok_or_else(|| LinkError::InvalidPack {
message: "ZIP has no end-of-central-directory record".to_string(),
})?;
let invalid_end = || LinkError::InvalidPack {
message: "ZIP has an invalid end-of-central-directory structure".to_string(),
};
let comment_len = le_u16(bytes, eocd + 20).ok_or_else(invalid_end)? as usize;
if eocd
.checked_add(22)
.and_then(|end| end.checked_add(comment_len))
!= Some(bytes.len())
{
return Err(invalid_end());
}
let disk = le_u16(bytes, eocd + 4);
let central_disk = le_u16(bytes, eocd + 6);
if disk != Some(0) || central_disk != Some(0) {
return Err(LinkError::InvalidPack {
message: "multi-disk ZIP archives are not supported".to_string(),
});
}
let entries_on_disk = le_u16(bytes, eocd + 8).ok_or_else(invalid_end)?;
let ordinary = le_u16(bytes, eocd + 10).ok_or_else(invalid_end)?;
if entries_on_disk != ordinary {
return Err(LinkError::InvalidPack {
message: "multi-disk ZIP archives are not supported".to_string(),
});
}
let zip64_locator = eocd
.checked_sub(20)
.filter(|at| bytes.get(*at..*at + 4) == Some(ZIP64_LOCATOR_SIG.as_slice()));
let (count, central_offset, central_size) = if ordinary != u16::MAX || zip64_locator.is_none() {
let central_size = le_u32(bytes, eocd + 12).ok_or_else(invalid_end)? as usize;
let central_offset = le_u32(bytes, eocd + 16).ok_or_else(invalid_end)? as usize;
if central_offset
.checked_add(central_size)
.filter(|end| *end == eocd)
.is_none()
{
return Err(invalid_end());
}
(ordinary as u64, central_offset, central_size)
} else {
let Some(locator) = zip64_locator else {
return Err(invalid_end());
};
if le_u32(bytes, locator + 4) != Some(0) || le_u32(bytes, locator + 16) != Some(1) {
return Err(LinkError::InvalidPack {
message: "multi-disk ZIP64 archives are not supported".to_string(),
});
}
let record = le_u64(bytes, locator + 8)
.and_then(|offset| usize::try_from(offset).ok())
.filter(|at| bytes.get(*at..*at + 4) == Some(ZIP64_EOCD_SIG.as_slice()))
.ok_or_else(|| LinkError::InvalidPack {
message: "ZIP64 archive has an invalid end record".to_string(),
})?;
let record_size = le_u64(bytes, record + 4)
.and_then(|size| usize::try_from(size).ok())
.filter(|size| *size >= 44)
.ok_or_else(invalid_end)?;
if record
.checked_add(12)
.and_then(|end| end.checked_add(record_size))
!= Some(locator)
|| le_u32(bytes, record + 16) != Some(0)
|| le_u32(bytes, record + 20) != Some(0)
{
return Err(invalid_end());
}
let zip64_on_disk = le_u64(bytes, record + 24).ok_or_else(invalid_end)?;
let zip64_total = le_u64(bytes, record + 32).ok_or_else(invalid_end)?;
let central_size = le_u64(bytes, record + 40)
.and_then(|size| usize::try_from(size).ok())
.ok_or_else(invalid_end)?;
let central_offset = le_u64(bytes, record + 48)
.and_then(|offset| usize::try_from(offset).ok())
.ok_or_else(invalid_end)?;
if zip64_on_disk != zip64_total
|| central_offset
.checked_add(central_size)
.filter(|end| *end == record)
.is_none()
{
return Err(invalid_end());
}
let legacy_size = le_u32(bytes, eocd + 12).ok_or_else(invalid_end)?;
let legacy_offset = le_u32(bytes, eocd + 16).ok_or_else(invalid_end)?;
if (legacy_size != u32::MAX && legacy_size as usize != central_size)
|| (legacy_offset != u32::MAX && legacy_offset as usize != central_offset)
{
return Err(invalid_end());
}
(zip64_total, central_offset, central_size)
};
if count == 0 || count > max_entries as u64 {
return Err(LinkError::InvalidPack {
message: format!("invalid file count {count}"),
});
}
preflight_zip_central_directory(bytes, central_offset, central_size, count)?;
Ok(())
}
fn parse_store_pack(bytes: Vec<u8>) -> LinkResult<Vec<(String, Vec<u8>)>> {
preflight_zip_entry_count(&bytes, MAX_PUSH_FILES)?;
let mut archive =
zip::ZipArchive::new(Cursor::new(bytes)).map_err(|err| LinkError::InvalidPack {
message: format!("ZIP parse failed: {err}"),
})?;
if archive.is_empty() || archive.len() > MAX_PUSH_FILES {
return Err(LinkError::InvalidPack {
message: format!("invalid file count {}", archive.len()),
});
}
let mut total = 0u64;
let mut seen = std::collections::HashSet::new();
let mut entries = Vec::with_capacity(archive.len());
for index in 0..archive.len() {
let mut file = archive
.by_index(index)
.map_err(|err| LinkError::InvalidPack {
message: format!("ZIP entry failed: {err}"),
})?;
if file.is_dir() {
continue;
}
let path = file.name().to_string();
if file.enclosed_name().is_none() || !safe_store_rel_path(&path) {
return Err(LinkError::UnsafePath { path });
}
if file
.unix_mode()
.is_some_and(|mode| !matches!(mode & 0o170000, 0 | 0o100000))
{
return Err(LinkError::InvalidPack {
message: format!("non-file entry `{path}`"),
});
}
if !seen.insert(path.clone()) {
return Err(LinkError::InvalidPack {
message: format!("duplicate path `{path}`"),
});
}
let remaining = MAX_STORE_BYTES.saturating_sub(total);
if file.size() > remaining {
return Err(LinkError::InvalidPack {
message: "expanded content exceeds the 512 MB limit".to_string(),
});
}
let mut content = Vec::new();
(&mut file)
.take(remaining + 1)
.read_to_end(&mut content)
.map_err(|err| LinkError::InvalidPack {
message: format!("could not decompress `{path}`: {err}"),
})?;
if content.len() as u64 > remaining {
return Err(LinkError::InvalidPack {
message: "expanded content exceeds the 512 MB limit".to_string(),
});
}
if content.len() as u64 != file.size() {
return Err(LinkError::InvalidPack {
message: format!("length mismatch for `{path}`"),
});
}
total += content.len() as u64;
entries.push((path, content));
}
if entries.is_empty() {
return Err(LinkError::InvalidPack {
message: "pack contains no files".to_string(),
});
}
Ok(entries)
}
fn verify_snapshot_manifest(entries: &[(String, Vec<u8>)], signed: &[FeedFile]) -> LinkResult<()> {
let mut expected = std::collections::BTreeMap::new();
for file in signed {
if !safe_store_rel_path(&file.path) {
return Err(LinkError::UnsafePath {
path: file.path.clone(),
});
}
if !is_sha256(&file.sha256)
|| expected
.insert(file.path.as_str(), (file.sha256.as_str(), file.bytes))
.is_some()
{
return Err(invalid_feed(
"signed snapshot manifest contains an invalid or duplicate file",
));
}
}
if expected.len() != entries.len() {
return Err(invalid_feed(
"downloaded pack file set differs from the signed snapshot manifest",
));
}
for (path, bytes) in entries {
let Some((sha256, declared_bytes)) = expected.get(path.as_str()) else {
return Err(invalid_feed(format!(
"downloaded pack contains unsigned path `{path}`"
)));
};
if *declared_bytes != bytes.len() as u64
|| *sha256 != format!("{:x}", Sha256::digest(bytes))
{
return Err(invalid_feed(format!(
"downloaded file `{path}` differs from its signed manifest"
)));
}
}
Ok(())
}
pub fn collect_push_files(store: &Store) -> LinkResult<Vec<(String, String)>> {
require_hardened_filesystem("sync push")?;
preflight_push_ownership(store)?;
let mut out: Vec<(String, String)> = Vec::new();
let mut total = 0u64;
let mut read_text = |rel: &str| -> LinkResult<String> {
let bytes = store.read_bounded(Path::new(rel), MAX_STORE_BYTES.saturating_sub(total))?;
total = total
.checked_add(bytes.len() as u64)
.ok_or_else(|| LinkError::PushTooLarge {
detail: "uncompressed byte count overflow".to_string(),
})?;
if total > MAX_STORE_BYTES {
return Err(LinkError::PushTooLarge {
detail: format!("{total} uncompressed bytes"),
});
}
String::from_utf8(bytes).map_err(|_| LinkError::NotUtf8 {
path: rel.to_string(),
})
};
out.push(("DB.md".to_string(), read_text("DB.md")?));
if store
.regular_file_exists(Path::new("assets.jsonl"))
.unwrap_or(false)
{
out.push(("assets.jsonl".to_string(), read_text("assets.jsonl")?));
}
for rel in store.walk()? {
let rel_str = rel.to_string_lossy().replace('\\', "/");
if !safe_store_rel_path(&rel_str) {
return Err(LinkError::UnsafePath { path: rel_str });
}
let content = read_text(&rel_str)?;
out.push((rel_str, content));
}
out.sort_by(|a, b| a.0.cmp(&b.0));
Ok(out)
}
fn preflight_push_ownership(store: &Store) -> LinkResult<()> {
if let Some(nested) = store.nested_store_roots()?.into_iter().next() {
return Err(LinkError::from(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
format!("cannot push: nested db.md store at {}", nested.display()),
)));
}
if let Some(symlink) = store.unowned_symlinks()?.into_iter().next() {
return Err(LinkError::from(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
format!(
"cannot push: {} is a symlink outside the store ownership model",
symlink.display()
),
)));
}
Ok(())
}
pub fn sync_push(cfg: &HubConfig, brain: &str, files: &[(String, String)]) -> LinkResult<Value> {
require_safe_ref(brain)?;
let remote = verified_remote_head(cfg, brain, false)?;
if files.len() > MAX_PUSH_FILES {
return Err(LinkError::PushTooLarge {
detail: format!("{} files", files.len()),
});
}
let raw_total: u64 = files.iter().map(|(_, content)| content.len() as u64).sum();
if raw_total > MAX_STORE_BYTES {
return Err(LinkError::PushTooLarge {
detail: format!("{raw_total} uncompressed bytes"),
});
}
if cfg.brain_key.is_none() {
let body = json!({
"files": files
.iter()
.map(|(p, c)| json!({ "path": p, "content": c }))
.collect::<Vec<_>>(),
});
if body.to_string().len() <= MAX_PUSH_BYTES {
let path = format!("/api/hub/brains/{brain}/push");
let pushed = ensure_ok(
request(cfg, "POST", &path, Some(&body), Auth::Required)?,
"sync push",
)?;
return Ok(pushed);
}
}
let pack = build_store_pack(files)?;
if pack.len() as u64 > MAX_PACK_BYTES {
return Err(LinkError::PushTooLarge {
detail: format!("{} pack bytes", pack.len()),
});
}
let sha256 = format!("{:x}", Sha256::digest(&pack));
let mut meta = json!({ "sha256": sha256, "bytes": pack.len() });
if let Some(key) = &cfg.brain_key {
if !remote.head.verified {
return Err(invalid_feed(
"self-custody push requires a fully verified, unscoped feed head",
));
}
let identity = remote
.identity
.as_ref()
.ok_or_else(|| invalid_feed("verified brain has no current identity"))?;
let current_multikey = format!("ed25519:{}", identity.fingerprint);
if key.multikey != current_multikey || key.public_key_spki != identity.public_key_spki {
return Err(invalid_feed(
"configured brain key is not the verified current brain identity",
));
}
let next_seq = remote
.head
.seq
.checked_add(1)
.ok_or_else(|| invalid_feed("feed sequence overflow"))?;
let mut manifest: Vec<WireFeedFile> = files
.iter()
.map(|(path, content)| WireFeedFile {
path: path.clone(),
sha256: format!("{:x}", Sha256::digest(content.as_bytes())),
bytes: content.len() as u64,
})
.collect();
manifest.sort_by(|a, b| a.path.cmp(&b.path));
let ts = crate::now()
.with_timezone(&chrono::Utc)
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
.to_string();
let entry = self_custody_entry(
key,
next_seq,
ts,
&sha256,
&manifest,
remote.head.feed_hash.as_deref(),
)?;
meta["entry"] = Value::String(entry);
}
let presigned = ensure_ok(
request(
cfg,
"POST",
&format!("/api/hub/brains/{brain}/packs/presign"),
Some(&meta),
Auth::Required,
)?,
"prepare pack upload",
)?;
let url = presigned
.get("url")
.and_then(Value::as_str)
.ok_or_else(|| LinkError::InvalidPack {
message: "the hub returned no upload URL".to_string(),
})?;
put_presigned(
cfg,
url,
presigned.get("headers").unwrap_or(&Value::Null),
&pack,
)?;
let committed = ensure_ok(
request(
cfg,
"POST",
&format!("/api/hub/brains/{brain}/packs/commit"),
Some(&meta),
Auth::Required,
)?,
"commit pack",
)?;
Ok(committed)
}
fn build_store_pack(files: &[(String, String)]) -> LinkResult<Vec<u8>> {
const LOCAL_HEADER: u32 = 0x0403_4b50;
const CENTRAL_HEADER: u32 = 0x0201_4b50;
const END_OF_CENTRAL_DIRECTORY: u32 = 0x0605_4b50;
const VERSION_20: u16 = 20;
const MADE_BY_UNIX_20: u16 = (3 << 8) | VERSION_20;
const UTF8_FLAG: u16 = 1 << 11;
const STORED: u16 = 0;
const DOS_TIME_MIDNIGHT: u16 = 0;
const DOS_DATE_1980_01_01: u16 = (1 << 5) | 1;
const UNIX_REGULAR_0600: u32 = 0o100600 << 16;
struct CentralEntry<'a> {
name: &'a [u8],
crc32: u32,
size: u32,
local_offset: u32,
}
fn push_u16(out: &mut Vec<u8>, value: u16) {
out.extend_from_slice(&value.to_le_bytes());
}
fn push_u32(out: &mut Vec<u8>, value: u32) {
out.extend_from_slice(&value.to_le_bytes());
}
if files.is_empty() {
return Err(LinkError::InvalidPack {
message: "cannot create an empty snapshot pack".to_string(),
});
}
if files.len() > u16::MAX as usize {
return Err(LinkError::PushTooLarge {
detail: format!(
"{} files (canonical ZIP32 packs cap at {})",
files.len(),
u16::MAX
),
});
}
let mut sorted: Vec<_> = files.iter().collect();
sorted.sort_by(|a, b| a.0.as_bytes().cmp(b.0.as_bytes()));
let mut previous: Option<&str> = None;
for (path, content) in &sorted {
if !safe_store_rel_path(path) {
return Err(LinkError::UnsafePath {
path: (*path).clone(),
});
}
if previous == Some(path.as_str()) {
return Err(LinkError::InvalidPack {
message: format!("duplicate path `{path}`"),
});
}
previous = Some(path.as_str());
u32::try_from(content.len()).map_err(|_| LinkError::PushTooLarge {
detail: format!("file `{path}` exceeds the ZIP32 per-file limit"),
})?;
}
let mut out = Vec::new();
let mut central = Vec::with_capacity(sorted.len());
for (path, content) in sorted {
let name = path.as_bytes();
let name_len = u16::try_from(name.len()).map_err(|_| LinkError::InvalidPack {
message: format!("ZIP entry name is too long: `{path}`"),
})?;
let bytes = content.as_bytes();
let size = u32::try_from(bytes.len()).map_err(|_| LinkError::PushTooLarge {
detail: format!("file `{path}` exceeds the ZIP32 per-file limit"),
})?;
let local_offset = u32::try_from(out.len()).map_err(|_| LinkError::PushTooLarge {
detail: "canonical ZIP32 pack exceeds its offset limit".to_string(),
})?;
let crc32 = crc32fast::hash(bytes);
push_u32(&mut out, LOCAL_HEADER);
push_u16(&mut out, VERSION_20);
push_u16(&mut out, UTF8_FLAG);
push_u16(&mut out, STORED);
push_u16(&mut out, DOS_TIME_MIDNIGHT);
push_u16(&mut out, DOS_DATE_1980_01_01);
push_u32(&mut out, crc32);
push_u32(&mut out, size);
push_u32(&mut out, size);
push_u16(&mut out, name_len);
push_u16(&mut out, 0); out.extend_from_slice(name);
out.extend_from_slice(bytes);
central.push(CentralEntry {
name,
crc32,
size,
local_offset,
});
}
let central_offset = u32::try_from(out.len()).map_err(|_| LinkError::PushTooLarge {
detail: "canonical ZIP32 pack exceeds its offset limit".to_string(),
})?;
for entry in ¢ral {
push_u32(&mut out, CENTRAL_HEADER);
push_u16(&mut out, MADE_BY_UNIX_20);
push_u16(&mut out, VERSION_20);
push_u16(&mut out, UTF8_FLAG);
push_u16(&mut out, STORED);
push_u16(&mut out, DOS_TIME_MIDNIGHT);
push_u16(&mut out, DOS_DATE_1980_01_01);
push_u32(&mut out, entry.crc32);
push_u32(&mut out, entry.size);
push_u32(&mut out, entry.size);
push_u16(&mut out, entry.name.len() as u16);
push_u16(&mut out, 0); push_u16(&mut out, 0); push_u16(&mut out, 0); push_u16(&mut out, 0); push_u32(&mut out, UNIX_REGULAR_0600);
push_u32(&mut out, entry.local_offset);
out.extend_from_slice(entry.name);
}
let central_size = u32::try_from(out.len())
.ok()
.and_then(|end| end.checked_sub(central_offset))
.ok_or_else(|| LinkError::PushTooLarge {
detail: "canonical ZIP32 central directory exceeds its limit".to_string(),
})?;
let entry_count = central.len() as u16;
push_u32(&mut out, END_OF_CENTRAL_DIRECTORY);
push_u16(&mut out, 0); push_u16(&mut out, 0); push_u16(&mut out, entry_count);
push_u16(&mut out, entry_count);
push_u32(&mut out, central_size);
push_u32(&mut out, central_offset);
push_u16(&mut out, 0);
if out.len() > u32::MAX as usize {
return Err(LinkError::PushTooLarge {
detail: "canonical ZIP32 pack exceeds 4 GiB".to_string(),
});
}
Ok(out)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Capability {
Read,
Write,
}
impl Capability {
pub fn as_str(self) -> &'static str {
match self {
Capability::Read => "read",
Capability::Write => "write",
}
}
}
pub fn grant_issue(
cfg: &HubConfig,
brain: &str,
grantee: &str,
can: Capability,
scope: Option<&str>,
until: Option<&str>,
) -> LinkResult<Value> {
require_safe_ref(brain)?;
let _ = verified_remote_head(cfg, brain, false)?;
let is_key_grantee = URL_SAFE_NO_PAD
.decode(grantee)
.map(|der| der.len() == 44 && der.starts_with(&ED25519_SPKI_PREFIX))
.unwrap_or(false);
let mut body = if is_key_grantee {
json!({ "keySpki": grantee, "capability": can.as_str() })
} else {
json!({ "email": grantee, "capability": can.as_str() })
};
if let Some(s) = scope {
body["scopePrefix"] = json!(s);
}
if let Some(u) = until {
body["expiresAt"] = json!(u);
}
let path = format!("/api/hub/brains/{brain}/grants");
ensure_ok(
request(cfg, "POST", &path, Some(&body), Auth::Required)?,
"grant issue",
)
}
pub fn grant_list(cfg: &HubConfig, brain: &str) -> LinkResult<Value> {
require_safe_ref(brain)?;
let _ = verified_remote_head(cfg, brain, false)?;
let path = format!("/api/hub/brains/{brain}/grants");
ensure_ok(
request(cfg, "GET", &path, None, Auth::Required)?,
"grant list",
)
}
pub fn grant_revoke(cfg: &HubConfig, brain: &str, grant_id: &str) -> LinkResult<Value> {
require_safe_ref(brain)?;
require_safe_grant_id(grant_id)?;
let _ = verified_remote_head(cfg, brain, false)?;
let path = format!("/api/hub/brains/{brain}/grants/{grant_id}");
ensure_ok(
request(cfg, "DELETE", &path, None, Auth::Required)?,
"grant revoke",
)
}
pub fn propose(cfg: &HubConfig, handle: &str, app: &str, body: &str) -> LinkResult<Value> {
require_valid_handle(handle)?;
if body.len() as u64 > MAX_PROPOSE_BYTES {
return Err(LinkError::ProposeTooLarge {
bytes: body.len() as u64,
});
}
let payload = json!({ "app": app, "body": body });
let (path, auth) = if crate::ulid::is_ulid(handle) {
(format!("/api/hub/brains/{handle}/inbox"), Auth::Optional)
} else {
(format!("/api/hub/sites/{handle}/inbox"), Auth::None)
};
ensure_ok(
request(cfg, "POST", &path, Some(&payload), auth)?,
"propose",
)
}
#[derive(Debug, serde::Serialize)]
pub struct Head {
pub brain: String,
pub seq: u64,
#[serde(rename = "updatedAt", skip_serializing_if = "Option::is_none")]
pub updated_at: Option<String>,
#[serde(rename = "feedHash", skip_serializing_if = "Option::is_none")]
pub feed_hash: Option<String>,
pub verified: bool,
}
struct BoundedVecVisitor<T, const MAX: usize> {
label: &'static str,
marker: std::marker::PhantomData<T>,
}
impl<'de, T, const MAX: usize> serde::de::Visitor<'de> for BoundedVecVisitor<T, MAX>
where
T: Deserialize<'de>,
{
type Value = Vec<T>;
fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(formatter, "at most {MAX} {}", self.label)
}
fn visit_seq<A>(self, mut sequence: A) -> Result<Self::Value, A::Error>
where
A: serde::de::SeqAccess<'de>,
{
if sequence.size_hint().is_some_and(|size| size > MAX) {
return Err(serde::de::Error::custom(format!(
"{} exceeds the {MAX}-item limit",
self.label
)));
}
let mut values = Vec::with_capacity(sequence.size_hint().unwrap_or(0).min(MAX));
while let Some(value) = sequence.next_element()? {
if values.len() == MAX {
return Err(serde::de::Error::custom(format!(
"{} exceeds the {MAX}-item limit",
self.label
)));
}
values.push(value);
}
Ok(values)
}
}
fn deserialize_bounded_vec<'de, D, T, const MAX: usize>(
deserializer: D,
label: &'static str,
) -> Result<Vec<T>, D::Error>
where
D: serde::Deserializer<'de>,
T: Deserialize<'de>,
{
deserializer.deserialize_seq(BoundedVecVisitor::<T, MAX> {
label,
marker: std::marker::PhantomData,
})
}
fn deserialize_feed_files<'de, D>(deserializer: D) -> Result<Vec<FeedFile>, D::Error>
where
D: serde::Deserializer<'de>,
{
deserialize_bounded_vec::<D, FeedFile, MAX_PUSH_FILES>(deserializer, "feed files")
}
fn deserialize_removed_paths<'de, D>(deserializer: D) -> Result<Vec<String>, D::Error>
where
D: serde::Deserializer<'de>,
{
deserialize_bounded_vec::<D, String, MAX_PUSH_FILES>(deserializer, "removed paths")
}
fn deserialize_previous_identities<'de, D>(
deserializer: D,
) -> Result<Vec<PreviousIdentity>, D::Error>
where
D: serde::Deserializer<'de>,
{
deserialize_bounded_vec::<D, PreviousIdentity, MAX_IDENTITY_ROTATIONS>(
deserializer,
"previous identities",
)
}
fn deserialize_rotations<'de, D>(deserializer: D) -> Result<Vec<String>, D::Error>
where
D: serde::Deserializer<'de>,
{
deserialize_bounded_vec::<D, String, MAX_IDENTITY_ROTATIONS>(
deserializer,
"rotation statements",
)
}
fn deserialize_feed_items<'de, D>(deserializer: D) -> Result<Vec<FeedItem>, D::Error>
where
D: serde::Deserializer<'de>,
{
deserialize_bounded_vec::<D, FeedItem, FEED_PAGE_LIMIT>(deserializer, "feed entries")
}
#[derive(Debug, Clone, Deserialize, Serialize)]
struct FeedFile {
path: String,
sha256: String,
bytes: u64,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
struct FeedEntry {
v: u8,
seq: u64,
ts: String,
brain: String,
public_key: String,
kind: String,
op: String,
pack_sha256: String,
#[serde(deserialize_with = "deserialize_feed_files")]
files: Vec<FeedFile>,
#[serde(deserialize_with = "deserialize_removed_paths")]
removed: Vec<String>,
prev_entry_hash: Option<String>,
sig: String,
}
#[derive(Serialize)]
struct UnsignedFeedEntry<'a> {
v: u8,
seq: u64,
ts: &'a str,
brain: &'a str,
public_key: &'a str,
kind: &'a str,
op: &'a str,
pack_sha256: &'a str,
files: &'a [FeedFile],
removed: &'a [String],
prev_entry_hash: &'a Option<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
struct FeedItem {
hash: String,
entry: FeedEntry,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
struct FeedIdentity {
fingerprint: String,
#[serde(rename = "publicKeySpki")]
public_key_spki: String,
#[serde(default, deserialize_with = "deserialize_previous_identities")]
previous: Vec<PreviousIdentity>,
#[serde(default, deserialize_with = "deserialize_rotations")]
rotations: Vec<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
struct PreviousIdentity {
fingerprint: String,
#[serde(rename = "publicKeySpki")]
public_key_spki: String,
}
#[derive(Debug, Deserialize)]
struct FeedResponse {
#[serde(rename = "headSeq")]
head_seq: u64,
#[serde(rename = "feedHash")]
feed_hash: Option<String>,
identity: Option<FeedIdentity>,
#[serde(deserialize_with = "deserialize_feed_items")]
entries: Vec<FeedItem>,
#[serde(rename = "scopeLimited")]
scope_limited: bool,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct RotationStatement {
v: u8,
op: String,
brain: String,
public_key: String,
new_brain: String,
new_public_key: String,
prior_head_seq: u64,
prior_feed_hash: Option<String>,
ts: String,
sig: String,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
struct TrustState {
v: u8,
origin: String,
#[serde(default)]
requested: String,
brain: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
home: Option<String>,
anchor: String,
current: String,
#[serde(rename = "headSeq")]
head_seq: u64,
#[serde(rename = "feedHash")]
feed_hash: Option<String>,
#[serde(default)]
rotations: Vec<String>,
}
#[derive(Debug, Clone, Deserialize, Serialize)]
struct AliasBinding {
v: u8,
origin: String,
requested: String,
brain: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
home: Option<String>,
}
struct VerifiedRemote {
head: Head,
identity: Option<FeedIdentity>,
head_entry: Option<FeedItem>,
entries: Vec<FeedItem>,
anchor: Option<String>,
}
fn invalid_feed(message: impl Into<String>) -> LinkError {
LinkError::InvalidFeed {
message: message.into(),
}
}
fn is_sha256(value: &str) -> bool {
value.len() == 64
&& value
.bytes()
.all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}
fn identity_fingerprint(public_key_spki: &str) -> LinkResult<String> {
let der = URL_SAFE_NO_PAD
.decode(public_key_spki)
.map_err(|_| invalid_feed("identity public key is not base64url"))?;
if der.len() != ED25519_SPKI_PREFIX.len() + 32 || !der.starts_with(&ED25519_SPKI_PREFIX) {
return Err(invalid_feed(
"identity public key is not a valid Ed25519 SPKI",
));
}
Ok(URL_SAFE_NO_PAD.encode(Sha256::digest(&der)))
}
fn verify_identity_chain(
identity: &FeedIdentity,
pinned: Option<&TrustState>,
) -> LinkResult<String> {
if identity.previous.len() > MAX_IDENTITY_ROTATIONS
|| identity.rotations.len() > MAX_IDENTITY_ROTATIONS
{
return Err(invalid_feed(
"identity rotation history exceeds the client cap",
));
}
if identity.fingerprint != identity_fingerprint(&identity.public_key_spki)? {
return Err(invalid_feed(
"current identity fingerprint does not match its public key",
));
}
for previous in &identity.previous {
if previous.fingerprint != identity_fingerprint(&previous.public_key_spki)? {
return Err(invalid_feed(
"previous identity fingerprint does not match its public key",
));
}
}
if identity.rotations.len() != identity.previous.len() {
return Err(invalid_feed(
"identity history is missing an old-key-signed rotation statement",
));
}
let mut chain: Vec<(&str, &str)> = identity
.previous
.iter()
.rev()
.map(|p| (p.fingerprint.as_str(), p.public_key_spki.as_str()))
.collect();
chain.push((&identity.fingerprint, &identity.public_key_spki));
for (index, raw) in identity.rotations.iter().enumerate() {
let statement: RotationStatement = serde_json::from_str(raw)
.map_err(|_| invalid_feed("rotation statement did not parse exactly"))?;
let (old_fingerprint, old_spki) = chain[index];
let (new_fingerprint, new_spki) = chain[index + 1];
if statement.v != 1
|| statement.op != "rotate"
|| statement.brain != format!("ed25519:{old_fingerprint}")
|| statement.public_key != old_spki
|| statement.new_brain != format!("ed25519:{new_fingerprint}")
|| statement.new_public_key != new_spki
|| (statement.prior_head_seq == 0 && statement.prior_feed_hash.is_some())
|| (statement.prior_head_seq > 0
&& statement
.prior_feed_hash
.as_deref()
.is_none_or(|hash| !is_sha256(hash)))
{
return Err(invalid_feed(
"rotation statement does not connect adjacent identities",
));
}
let unsigned = serde_json::to_string(&UnsignedRotation {
v: statement.v,
op: &statement.op,
brain: &statement.brain,
public_key: &statement.public_key,
new_brain: &statement.new_brain,
new_public_key: &statement.new_public_key,
prior_head_seq: statement.prior_head_seq,
prior_feed_hash: statement.prior_feed_hash.as_deref(),
ts: statement.ts.clone(),
})
.map_err(|_| invalid_feed("could not canonicalize rotation statement"))?;
let exact = format!(
"{},\"sig\":\"{}\"}}",
&unsigned[..unsigned.len() - 1],
statement.sig
);
if exact != *raw {
return Err(invalid_feed(
"rotation statement is not in normative serialization",
));
}
let der = URL_SAFE_NO_PAD
.decode(old_spki)
.map_err(|_| invalid_feed("rotation public key is not base64url"))?;
let signature = URL_SAFE_NO_PAD
.decode(&statement.sig)
.map_err(|_| invalid_feed("rotation signature is not base64url"))?;
UnparsedPublicKey::new(&ED25519, &der[ED25519_SPKI_PREFIX.len()..])
.verify(unsigned.as_bytes(), &signature)
.map_err(|_| invalid_feed("rotation signature verification failed"))?;
if index > 0 {
let prior: RotationStatement = serde_json::from_str(&identity.rotations[index - 1])
.map_err(|_| invalid_feed("prior rotation statement did not parse"))?;
if statement.prior_head_seq < prior.prior_head_seq {
return Err(invalid_feed("rotation feed boundaries move backward"));
}
}
}
let anchor = format!("ed25519:{}", chain[0].0);
let current = format!("ed25519:{}", identity.fingerprint);
if let Some(pin) = pinned {
if pin.anchor != anchor {
return Err(invalid_feed(
"served identity chain does not descend from the pinned anchor",
));
}
if !chain
.iter()
.any(|(fingerprint, _)| pin.current == format!("ed25519:{fingerprint}"))
{
return Err(invalid_feed(
"served identity chain forked away from the last pinned identity",
));
}
if pin.current == current && pin.anchor != current && identity.rotations.is_empty() {
return Err(invalid_feed("served identity discarded its rotation chain"));
}
if pin.v >= 2
&& (identity.rotations.len() < pin.rotations.len()
|| identity.rotations[..pin.rotations.len()] != pin.rotations)
{
return Err(invalid_feed(
"served identity rewrote the locally accepted rotation history",
));
}
}
Ok(anchor)
}
fn verify_rotation_feed_boundaries(
identity: &FeedIdentity,
pinned: Option<&TrustState>,
observed: &[FeedItem],
advertised_seq: u64,
) -> LinkResult<()> {
let mut chain: Vec<String> = identity
.previous
.iter()
.rev()
.map(|previous| format!("ed25519:{}", previous.fingerprint))
.collect();
chain.push(format!("ed25519:{}", identity.fingerprint));
let pinned_index = pinned.and_then(|pin| chain.iter().position(|key| key == &pin.current));
for (index, raw) in identity.rotations.iter().enumerate() {
let rotation: RotationStatement = serde_json::from_str(raw)
.map_err(|_| invalid_feed("rotation statement did not parse"))?;
if rotation.prior_head_seq > advertised_seq {
return Err(invalid_feed(
"rotation claims a feed boundary beyond the advertised head",
));
}
if let (Some(pin), Some(pin_index)) = (pinned, pinned_index) {
if index >= pin_index && rotation.prior_head_seq < pin.head_seq {
return Err(invalid_feed(
"newly disclosed rotation predates the local feed checkpoint",
));
}
}
let actual = if rotation.prior_head_seq == 0 {
None
} else if pinned.is_some_and(|pin| pin.head_seq == rotation.prior_head_seq) {
pinned.and_then(|pin| pin.feed_hash.as_deref())
} else {
observed
.iter()
.find(|item| item.entry.seq == rotation.prior_head_seq)
.map(|item| item.hash.as_str())
};
if let Some(actual) = actual {
if rotation.prior_feed_hash.as_deref() != Some(actual) {
return Err(invalid_feed(
"rotation statement does not commit the verified feed boundary",
));
}
} else if rotation.prior_head_seq == 0 {
} else if pinned.is_some_and(|pin| {
pinned_index.is_some_and(|pin_index| index >= pin_index)
|| rotation.prior_head_seq >= pin.head_seq
}) {
return Err(invalid_feed(
"rotation feed boundary was not present in the verified chain",
));
}
}
Ok(())
}
fn reject_retired_signer_after_checkpoint(
identity: &FeedIdentity,
pinned: Option<&TrustState>,
item: &FeedItem,
) -> LinkResult<()> {
let Some(pin) = pinned else {
return Ok(());
};
if item.entry.seq <= pin.head_seq {
return Ok(());
}
let mut chain: Vec<String> = identity
.previous
.iter()
.rev()
.map(|previous| format!("ed25519:{}", previous.fingerprint))
.collect();
chain.push(format!("ed25519:{}", identity.fingerprint));
let pinned_index = chain
.iter()
.position(|key| key == &pin.current)
.ok_or_else(|| invalid_feed("pinned identity is absent from the served chain"))?;
let signer_index = chain
.iter()
.position(|key| key == &item.entry.brain)
.ok_or_else(|| invalid_feed("feed signer is absent from the served identity chain"))?;
if signer_index < pinned_index {
return Err(invalid_feed(
"a retired identity attempted to sign after the local checkpoint",
));
}
Ok(())
}
fn trust_file_name(cfg: &HubConfig, brain: &str) -> LinkResult<String> {
let origin = normalized_origin(&cfg.hub)?;
let key = format!(
"{:x}",
Sha256::digest(format!("{origin}\0{brain}").as_bytes())
);
Ok(format!("{key}.json"))
}
fn alias_file_name(cfg: &HubConfig, alias: &str) -> LinkResult<String> {
let origin = normalized_origin(&cfg.hub)?;
let key = format!(
"{:x}",
Sha256::digest(format!("{origin}\0alias\0{alias}").as_bytes())
);
Ok(format!("alias-{key}.json"))
}
#[cfg(unix)]
struct TrustLock {
_file: std::fs::File,
}
#[cfg(unix)]
fn lock_trust_name(directory: &std::fs::File, state_name: &str) -> LinkResult<TrustLock> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
let lock_string = format!(".{state_name}.lock");
let lock_name = c_name(lock_string.as_bytes(), &lock_string)?;
let fd = unsafe {
libc::openat(
directory.as_raw_fd(),
lock_name.as_ptr(),
libc::O_RDWR | libc::O_CREAT | libc::O_CLOEXEC | libc::O_NOFOLLOW,
0o600,
)
};
if fd < 0 {
return Err(std::io::Error::last_os_error().into());
}
let file = unsafe { std::fs::File::from_raw_fd(fd) };
if !file.metadata()?.is_file() {
return Err(LinkError::UnsafePath { path: lock_string });
}
if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } != 0 {
return Err(std::io::Error::last_os_error().into());
}
Ok(TrustLock { _file: file })
}
#[cfg(unix)]
fn lock_trust_many(
cfg: &HubConfig,
directory: &std::fs::File,
refs: &[&str],
) -> LinkResult<Vec<TrustLock>> {
let mut names = refs
.iter()
.map(|reference| trust_file_name(cfg, reference))
.collect::<LinkResult<Vec<_>>>()?;
names.sort();
names.dedup();
names
.iter()
.map(|name| lock_trust_name(directory, name))
.collect()
}
#[cfg(not(unix))]
fn lock_trust_many(
_cfg: &HubConfig,
_directory: &TrustDirectory,
_refs: &[&str],
) -> LinkResult<Vec<()>> {
Err(LinkError::UnsupportedPlatform {
operation: "verified link.md state",
})
}
#[cfg(unix)]
type TrustDirectory = std::fs::File;
#[cfg(not(unix))]
struct TrustDirectory;
#[cfg(unix)]
fn open_trust_dir(cfg: &HubConfig) -> LinkResult<TrustDirectory> {
use std::os::fd::AsRawFd as _;
let directory = open_or_create_dir_nofollow(&cfg.state_dir.join("trust"))?;
if unsafe { libc::fchmod(directory.as_raw_fd(), 0o700) } != 0 {
return Err(std::io::Error::last_os_error().into());
}
directory.sync_all()?;
Ok(directory)
}
#[cfg(not(unix))]
fn open_trust_dir(_cfg: &HubConfig) -> LinkResult<TrustDirectory> {
Err(LinkError::UnsupportedPlatform {
operation: "verified link.md state",
})
}
#[cfg(unix)]
fn load_trust_in(
cfg: &HubConfig,
directory: &TrustDirectory,
requested: &str,
) -> LinkResult<Option<TrustState>> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
let name_string = trust_file_name(cfg, requested)?;
let name = c_name(name_string.as_bytes(), &name_string)?;
let fd = unsafe {
libc::openat(
directory.as_raw_fd(),
name.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if fd < 0 {
let error = std::io::Error::last_os_error();
if error.kind() == std::io::ErrorKind::NotFound {
return Ok(None);
}
return Err(LinkError::UnsafePath { path: name_string });
}
let file = unsafe { std::fs::File::from_raw_fd(fd) };
if !file.metadata()?.is_file() {
return Err(LinkError::UnsafePath { path: name_string });
}
let mut bytes = Vec::new();
file.take(1024 * 1024 + 1).read_to_end(&mut bytes)?;
if bytes.len() > 1024 * 1024 {
return Err(invalid_feed("local identity/feed checkpoint is oversized"));
}
let mut state: TrustState = serde_json::from_slice(&bytes)
.map_err(|_| invalid_feed("local identity/feed checkpoint is corrupt"))?;
if !matches!(state.v, 1 | 2) || state.origin != normalized_origin(&cfg.hub)? {
return Err(invalid_feed(
"local identity/feed checkpoint does not match this hub and brain",
));
}
if state.v == 1 {
if state.brain != requested {
return Err(invalid_feed(
"legacy checkpoint is not bound to the requested brain id",
));
}
state.requested = requested.to_string();
} else if state.requested != requested {
return Err(invalid_feed(
"local identity/feed checkpoint is bound to a different requested ref",
));
}
Ok(Some(state))
}
#[cfg(not(unix))]
fn load_trust_in(
_cfg: &HubConfig,
_directory: &TrustDirectory,
_brain: &str,
) -> LinkResult<Option<TrustState>> {
Err(LinkError::UnsupportedPlatform {
operation: "verified link.md state",
})
}
#[cfg(all(test, unix))]
fn load_trust(cfg: &HubConfig, requested: &str) -> LinkResult<Option<TrustState>> {
let directory = open_trust_dir(cfg)?;
load_trust_in(cfg, &directory, requested)
}
#[cfg(unix)]
fn save_trust_in(
cfg: &HubConfig,
directory: &TrustDirectory,
state: &TrustState,
) -> LinkResult<()> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
let name_string = trust_file_name(cfg, &state.requested)?;
let name = c_name(name_string.as_bytes(), &name_string)?;
let mut bytes = serde_json::to_vec(state)
.map_err(|_| invalid_feed("could not serialize local trust checkpoint"))?;
bytes.push(b'\n');
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let temp_string = format!(".{name_string}.tmp.{}-{nonce}", std::process::id());
let temp = c_name(temp_string.as_bytes(), &temp_string)?;
let fd = unsafe {
libc::openat(
directory.as_raw_fd(),
temp.as_ptr(),
libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC | libc::O_NOFOLLOW,
0o600,
)
};
if fd < 0 {
return Err(std::io::Error::last_os_error().into());
}
let mut file = unsafe { std::fs::File::from_raw_fd(fd) };
if let Err(error) = file.write_all(&bytes).and_then(|_| file.sync_all()) {
let _ = unsafe { libc::unlinkat(directory.as_raw_fd(), temp.as_ptr(), 0) };
return Err(error.into());
}
drop(file);
if unsafe {
libc::renameat(
directory.as_raw_fd(),
temp.as_ptr(),
directory.as_raw_fd(),
name.as_ptr(),
)
} != 0
{
let error = std::io::Error::last_os_error();
let _ = unsafe { libc::unlinkat(directory.as_raw_fd(), temp.as_ptr(), 0) };
return Err(error.into());
}
directory.sync_all()?;
Ok(())
}
#[cfg(not(unix))]
fn save_trust_in(
_cfg: &HubConfig,
_directory: &TrustDirectory,
_state: &TrustState,
) -> LinkResult<()> {
Err(LinkError::UnsupportedPlatform {
operation: "verified link.md state",
})
}
#[cfg(unix)]
fn load_alias_in(
cfg: &HubConfig,
directory: &TrustDirectory,
requested: &str,
) -> LinkResult<Option<AliasBinding>> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
let name_string = alias_file_name(cfg, requested)?;
let name = c_name(name_string.as_bytes(), &name_string)?;
let fd = unsafe {
libc::openat(
directory.as_raw_fd(),
name.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if fd < 0 {
let error = std::io::Error::last_os_error();
if error.kind() == std::io::ErrorKind::NotFound {
return Ok(None);
}
return Err(LinkError::UnsafePath { path: name_string });
}
let file = unsafe { std::fs::File::from_raw_fd(fd) };
if !file.metadata()?.is_file() {
return Err(LinkError::UnsafePath { path: name_string });
}
let mut bytes = Vec::new();
file.take(64 * 1024 + 1).read_to_end(&mut bytes)?;
if bytes.len() > 64 * 1024 {
return Err(invalid_feed("local alias binding is oversized"));
}
let alias: AliasBinding = serde_json::from_slice(&bytes)
.map_err(|_| invalid_feed("local alias binding is corrupt"))?;
if alias.v != 1 || alias.origin != normalized_origin(&cfg.hub)? || alias.requested != requested
{
return Err(invalid_feed(
"local alias binding does not match this hub and requested ref",
));
}
Ok(Some(alias))
}
#[cfg(not(unix))]
fn load_alias_in(
_cfg: &HubConfig,
_directory: &TrustDirectory,
_requested: &str,
) -> LinkResult<Option<AliasBinding>> {
Err(LinkError::UnsupportedPlatform {
operation: "verified link.md state",
})
}
#[cfg(unix)]
fn save_alias_in(
cfg: &HubConfig,
directory: &TrustDirectory,
alias: &AliasBinding,
) -> LinkResult<()> {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
let name_string = alias_file_name(cfg, &alias.requested)?;
let name = c_name(name_string.as_bytes(), &name_string)?;
let mut bytes = serde_json::to_vec(alias)
.map_err(|_| invalid_feed("could not serialize local alias binding"))?;
bytes.push(b'\n');
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let temp_string = format!(".{name_string}.tmp.{}-{nonce}", std::process::id());
let temp = c_name(temp_string.as_bytes(), &temp_string)?;
let fd = unsafe {
libc::openat(
directory.as_raw_fd(),
temp.as_ptr(),
libc::O_WRONLY | libc::O_CREAT | libc::O_EXCL | libc::O_CLOEXEC | libc::O_NOFOLLOW,
0o600,
)
};
if fd < 0 {
return Err(std::io::Error::last_os_error().into());
}
let mut file = unsafe { std::fs::File::from_raw_fd(fd) };
if let Err(error) = file.write_all(&bytes).and_then(|_| file.sync_all()) {
let _ = unsafe { libc::unlinkat(directory.as_raw_fd(), temp.as_ptr(), 0) };
return Err(error.into());
}
drop(file);
if unsafe {
libc::renameat(
directory.as_raw_fd(),
temp.as_ptr(),
directory.as_raw_fd(),
name.as_ptr(),
)
} != 0
{
let error = std::io::Error::last_os_error();
let _ = unsafe { libc::unlinkat(directory.as_raw_fd(), temp.as_ptr(), 0) };
return Err(error.into());
}
directory.sync_all()?;
Ok(())
}
#[cfg(not(unix))]
fn save_alias_in(
_cfg: &HubConfig,
_directory: &TrustDirectory,
_alias: &AliasBinding,
) -> LinkResult<()> {
Err(LinkError::UnsupportedPlatform {
operation: "verified link.md state",
})
}
fn load_canonical_pin(
cfg: &HubConfig,
directory: &TrustDirectory,
requested: &str,
resolved_brain: &str,
) -> LinkResult<(Option<TrustState>, Option<AliasBinding>)> {
let mut canonical = load_trust_in(cfg, directory, resolved_brain)?;
if requested == resolved_brain {
return Ok((canonical, None));
}
let mut alias = load_alias_in(cfg, directory, requested)?;
if let Some(binding) = &alias {
if binding.brain != resolved_brain {
return Err(invalid_feed(
"requested brain alias now resolves to a different canonical brain",
));
}
return Ok((canonical, alias));
}
if let Some(legacy) = load_trust_in(cfg, directory, requested)? {
if legacy.brain != resolved_brain {
return Err(invalid_feed(
"legacy alias checkpoint names a different canonical brain",
));
}
if let Some(existing) = &canonical {
if existing.brain != legacy.brain
|| existing.anchor != legacy.anchor
|| existing.current != legacy.current
|| existing.head_seq != legacy.head_seq
|| existing.feed_hash != legacy.feed_hash
|| existing.rotations != legacy.rotations
{
return Err(invalid_feed(
"legacy alias checkpoint conflicts with the canonical checkpoint",
));
}
} else {
let mut promoted = legacy.clone();
promoted.requested = resolved_brain.to_string();
promoted.home = None;
save_trust_in(cfg, directory, &promoted)?;
canonical = Some(promoted);
}
alias = Some(AliasBinding {
v: 1,
origin: normalized_origin(&cfg.hub)?,
requested: requested.to_string(),
brain: resolved_brain.to_string(),
home: legacy.home,
});
save_alias_in(cfg, directory, alias.as_ref().expect("alias just created"))?;
}
Ok((canonical, alias))
}
fn save_canonical_pin_and_alias(
cfg: &HubConfig,
directory: &TrustDirectory,
requested: &str,
resolved_brain: &str,
mut state: TrustState,
existing_alias: Option<&AliasBinding>,
) -> LinkResult<()> {
state.requested = resolved_brain.to_string();
state.brain = resolved_brain.to_string();
state.home = None;
save_trust_in(cfg, directory, &state)?;
if requested != resolved_brain {
save_alias_in(
cfg,
directory,
&AliasBinding {
v: 1,
origin: normalized_origin(&cfg.hub)?,
requested: requested.to_string(),
brain: resolved_brain.to_string(),
home: existing_alias.and_then(|alias| alias.home.clone()),
},
)?;
}
Ok(())
}
fn verify_feed_item(item: &FeedItem, identity: &FeedIdentity) -> LinkResult<()> {
const ED25519_SPKI_PREFIX: &[u8] = &[
0x30, 0x2a, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x03, 0x21, 0x00,
];
let entry = &item.entry;
let public_der = URL_SAFE_NO_PAD
.decode(&entry.public_key)
.map_err(|_| invalid_feed("public key is not base64url"))?;
if public_der.len() != ED25519_SPKI_PREFIX.len() + 32
|| !public_der.starts_with(ED25519_SPKI_PREFIX)
{
return Err(invalid_feed("entry public key is not a valid Ed25519 SPKI"));
}
let fingerprint = URL_SAFE_NO_PAD.encode(Sha256::digest(&public_der));
if entry.brain != format!("ed25519:{fingerprint}") {
return Err(invalid_feed(
"brain fingerprint does not match its public key",
));
}
let _ = verify_identity_chain(identity, None)?;
let mut chain: Vec<(&str, &str)> = identity
.previous
.iter()
.rev()
.map(|previous| {
(
previous.fingerprint.as_str(),
previous.public_key_spki.as_str(),
)
})
.collect();
chain.push((&identity.fingerprint, &identity.public_key_spki));
let signer_index = chain.iter().position(|(known_fingerprint, spki)| {
*known_fingerprint == fingerprint && *spki == entry.public_key
});
let Some(signer_index) = signer_index else {
return Err(invalid_feed(
"entry signer is not this brain's identity (current or rotated-from)",
));
};
let lower_boundary = if signer_index == 0 {
None
} else {
let prior: RotationStatement = serde_json::from_str(&identity.rotations[signer_index - 1])
.map_err(|_| invalid_feed("rotation statement did not parse"))?;
Some(prior.prior_head_seq)
};
let upper_boundary = if signer_index == identity.rotations.len() {
None
} else {
let next: RotationStatement = serde_json::from_str(&identity.rotations[signer_index])
.map_err(|_| invalid_feed("rotation statement did not parse"))?;
Some(next.prior_head_seq)
};
if lower_boundary.is_some_and(|boundary| entry.seq <= boundary)
|| upper_boundary.is_some_and(|boundary| entry.seq > boundary)
{
return Err(invalid_feed(
"entry signer is outside its authenticated rotation epoch",
));
}
let unsigned = UnsignedFeedEntry {
v: entry.v,
seq: entry.seq,
ts: &entry.ts,
brain: &entry.brain,
public_key: &entry.public_key,
kind: &entry.kind,
op: &entry.op,
pack_sha256: &entry.pack_sha256,
files: &entry.files,
removed: &entry.removed,
prev_entry_hash: &entry.prev_entry_hash,
};
let message =
serde_json::to_vec(&unsigned).map_err(|_| invalid_feed("could not canonicalize entry"))?;
let signature = URL_SAFE_NO_PAD
.decode(&entry.sig)
.map_err(|_| invalid_feed("signature is not base64url"))?;
UnparsedPublicKey::new(&ED25519, &public_der[ED25519_SPKI_PREFIX.len()..])
.verify(&message, &signature)
.map_err(|_| invalid_feed("Ed25519 signature verification failed"))?;
let mut exact = serde_json::to_vec(entry).map_err(|_| invalid_feed("could not hash entry"))?;
exact.push(b'\n');
let actual_hash = format!("{:x}", Sha256::digest(&exact));
if actual_hash != item.hash {
return Err(invalid_feed("entry SHA-256 does not match"));
}
Ok(())
}
#[derive(Serialize)]
struct UnsignedRotation<'a> {
v: u8,
op: &'a str,
brain: &'a str,
public_key: &'a str,
new_brain: &'a str,
new_public_key: &'a str,
prior_head_seq: u64,
prior_feed_hash: Option<&'a str>,
ts: String,
}
#[derive(Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct RotationJournal {
v: u8,
origin: String,
brain: String,
old_brain: String,
new_brain: String,
prior_head_seq: u64,
prior_feed_hash: Option<String>,
statement: String,
}
fn rotation_journal_path(key_path: &Path) -> PathBuf {
let mut path = key_path.as_os_str().to_os_string();
path.push(".rotation.json");
PathBuf::from(path)
}
fn read_rotation_journal(path: &Path) -> LinkResult<RotationJournal> {
#[cfg(unix)]
let file = {
use std::os::fd::{AsRawFd as _, FromRawFd as _};
use std::os::unix::ffi::OsStrExt as _;
let parent = open_existing_dir_nofollow(path.parent().unwrap_or_else(|| Path::new(".")))
.map_err(|error| {
bad_agent_key(&format!("cannot open the rotation journal parent: {error}"))
})?;
let leaf_name = path
.file_name()
.ok_or_else(|| bad_agent_key("the rotation journal path has no file name"))?;
let leaf = c_name(leaf_name.as_bytes(), &path.display().to_string())?;
let fd = unsafe {
libc::openat(
parent.as_raw_fd(),
leaf.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if fd < 0 {
return Err(bad_agent_key(
"the rotation journal must be an existing regular file without symlink ancestors",
));
}
unsafe { std::fs::File::from_raw_fd(fd) }
};
#[cfg(not(unix))]
let file = std::fs::File::open(path)
.map_err(|error| bad_agent_key(&format!("cannot read the rotation journal: {error}")))?;
let metadata = file
.metadata()
.map_err(|error| bad_agent_key(&format!("cannot inspect the rotation journal: {error}")))?;
if !metadata.is_file() || metadata.len() > MAX_REGISTRY_CARD_BYTES {
return Err(bad_agent_key(
"the rotation journal must be a bounded regular file",
));
}
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
if metadata.permissions().mode() & 0o077 != 0 {
return Err(bad_agent_key(
"the rotation journal is accessible to group/other; set mode 0600",
));
}
}
serde_json::from_reader(file)
.map_err(|_| bad_agent_key("the rotation journal is not valid exact JSON"))
}
fn remove_rotation_journal(path: &Path) {
#[cfg(unix)]
{
use std::os::fd::AsRawFd as _;
use std::os::unix::ffi::OsStrExt as _;
let Ok(parent) =
open_existing_dir_nofollow(path.parent().unwrap_or_else(|| Path::new(".")))
else {
return;
};
let Some(leaf_name) = path.file_name() else {
return;
};
let Ok(leaf) = c_name(leaf_name.as_bytes(), &path.display().to_string()) else {
return;
};
if unsafe { libc::unlinkat(parent.as_raw_fd(), leaf.as_ptr(), 0) } == 0 {
let _ = parent.sync_all();
}
}
#[cfg(not(unix))]
{
let _ = std::fs::remove_file(path);
}
}
fn validate_rotation_journal(
journal: &RotationJournal,
cfg: &HubConfig,
canonical_brain: &str,
old_key: &AgentSigningKey,
new_key: &AgentSigningKey,
head: &Head,
) -> LinkResult<()> {
if journal.v != 1
|| journal.origin != normalized_origin(&cfg.hub)?
|| journal.brain != canonical_brain
|| journal.old_brain != old_key.multikey
|| journal.new_brain != new_key.multikey
|| journal.prior_head_seq != head.seq
|| journal.prior_feed_hash != head.feed_hash
{
return Err(invalid_feed(
"rotation journal does not match the verified key and feed boundary",
));
}
let statement: RotationStatement = serde_json::from_str(&journal.statement)
.map_err(|_| invalid_feed("rotation journal statement did not parse exactly"))?;
if statement.prior_head_seq != journal.prior_head_seq
|| statement.prior_feed_hash != journal.prior_feed_hash
|| statement.brain != old_key.multikey
|| statement.public_key != old_key.public_key_spki
|| statement.new_brain != new_key.multikey
|| statement.new_public_key != new_key.public_key_spki
{
return Err(invalid_feed(
"rotation journal statement does not match its durable intent",
));
}
let identity = FeedIdentity {
fingerprint: new_key.multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: new_key.public_key_spki.clone(),
previous: vec![PreviousIdentity {
fingerprint: old_key.multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: old_key.public_key_spki.clone(),
}],
rotations: vec![journal.statement.clone()],
};
verify_identity_chain(&identity, None)?;
Ok(())
}
#[derive(Debug, Serialize)]
pub struct RotationReport {
pub brain: String,
pub multikey: String,
#[serde(rename = "keyFile")]
pub key_file: String,
pub previous: Vec<String>,
}
pub fn rotate_brain_key(
cfg: &HubConfig,
brain: &str,
old_key: &AgentSigningKey,
out: &Path,
) -> LinkResult<RotationReport> {
require_hardened_filesystem("key rotation")?;
require_safe_ref(brain)?;
let new_key = if out.exists() {
load_signing_key(out)?
} else {
let rng = ring::rand::SystemRandom::new();
let pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng)
.map_err(|_| bad_agent_key("key generation failed"))?;
let pair = agent_keypair(pkcs8.as_ref())?;
let (public_key_spki, multikey) = public_identity_for(&pair);
write_secret_new(
out,
format!("{}\n", URL_SAFE_NO_PAD.encode(pkcs8.as_ref())).as_bytes(),
)?;
AgentSigningKey {
pkcs8: pkcs8.as_ref().to_vec(),
multikey,
public_key_spki,
}
};
let new_spki = new_key.public_key_spki.clone();
let new_multikey = new_key.multikey.clone();
let journal_path = rotation_journal_path(out);
let before = verified_remote_head(cfg, brain, false)?;
let served_identity = before
.identity
.as_ref()
.ok_or_else(|| invalid_feed("cannot rotate a brain with no signed identity"))?;
let served_multikey = format!("ed25519:{}", served_identity.fingerprint);
if served_multikey == new_multikey {
remove_rotation_journal(&journal_path);
return Ok(RotationReport {
brain: brain.to_string(),
multikey: new_multikey,
key_file: out.display().to_string(),
previous: served_identity
.previous
.iter()
.map(|identity| format!("ed25519:{}", identity.fingerprint))
.collect(),
});
}
if served_multikey != old_key.multikey {
return Err(invalid_feed(
"the supplied old key is not the brain's verified current identity",
));
}
let journal = if journal_path.exists() {
read_rotation_journal(&journal_path)?
} else {
let ts = crate::now()
.with_timezone(&chrono::Utc)
.format("%Y-%m-%dT%H:%M:%S%.3fZ")
.to_string();
let unsigned = serde_json::to_string(&UnsignedRotation {
v: 1,
op: "rotate",
brain: &old_key.multikey,
public_key: &old_key.public_key_spki,
new_brain: &new_multikey,
new_public_key: &new_spki,
prior_head_seq: before.head.seq,
prior_feed_hash: before.head.feed_hash.as_deref(),
ts,
})
.expect("serialize rotation");
let old_pair = agent_keypair(&old_key.pkcs8)?;
let sig = URL_SAFE_NO_PAD.encode(old_pair.sign(unsigned.as_bytes()).as_ref());
let statement = format!("{},\"sig\":\"{}\"}}", &unsigned[..unsigned.len() - 1], sig);
let journal = RotationJournal {
v: 1,
origin: normalized_origin(&cfg.hub)?,
brain: before.head.brain.clone(),
old_brain: old_key.multikey.clone(),
new_brain: new_multikey.clone(),
prior_head_seq: before.head.seq,
prior_feed_hash: before.head.feed_hash.clone(),
statement,
};
let mut exact = serde_json::to_vec(&journal)
.map_err(|_| invalid_feed("could not serialize rotation journal"))?;
exact.push(b'\n');
if write_secret_new(&journal_path, &exact).is_err() {
read_rotation_journal(&journal_path)?
} else {
journal
}
};
validate_rotation_journal(
&journal,
cfg,
&before.head.brain,
old_key,
&new_key,
&before.head,
)?;
let body = json!({ "statement": journal.statement });
let path = format!("/api/hub/brains/{brain}/rotate");
let attempted = request(cfg, "POST", &path, Some(&body), Auth::Required);
let attempted_failure = match attempted {
Ok(response) if (200..300).contains(&response.status) => None,
Ok(response) => Some(ensure_ok(response, "key rotate").unwrap_err()),
Err(error) => Some(error),
};
let after = match verified_remote_head(cfg, brain, false) {
Ok(after) => after,
Err(error) => return Err(attempted_failure.unwrap_or(error)),
};
let identity = after
.identity
.as_ref()
.ok_or_else(|| invalid_feed("rotated brain has no verified identity"))?;
if format!("ed25519:{}", identity.fingerprint) != new_multikey
|| identity.public_key_spki != new_spki
{
return Err(attempted_failure.unwrap_or_else(|| {
invalid_feed("hub acknowledged rotation without committing the verified new identity")
}));
}
let previous = identity
.previous
.iter()
.map(|prior| format!("ed25519:{}", prior.fingerprint))
.collect();
remove_rotation_journal(&journal_path);
Ok(RotationReport {
brain: brain.to_string(),
multikey: new_multikey,
key_file: out.display().to_string(),
previous,
})
}
#[derive(Debug, Serialize)]
pub struct MirrorReport {
pub brain: String,
#[serde(rename = "headSeq")]
pub head_seq: u64,
#[serde(rename = "feedHash")]
pub feed_hash: Option<String>,
pub entries: u64,
pub pinned: String,
pub files: usize,
}
pub const MIRROR_REL_DIR: &str = ".dbmd/mirror";
#[derive(Debug)]
pub struct VerifiedMirrorMaterial {
pub brain: String,
pub head_seq: u64,
pub feed_hash: Option<String>,
pub identity: serde_json::Value,
pub entries: Vec<(u64, String, String)>,
pub pack_sha256: Option<String>,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct StoredMirrorHead {
brain: String,
#[serde(rename = "headSeq")]
head_seq: u64,
#[serde(rename = "feedHash")]
feed_hash: Option<String>,
}
pub fn verify_mirror_material(
head_bytes: &[u8],
identity_bytes: &[u8],
feed_bytes: &[Vec<u8>],
snapshot_pack: Option<&[u8]>,
expected_anchor: &str,
) -> LinkResult<VerifiedMirrorMaterial> {
let snapshot_hash = snapshot_pack
.filter(|pack| !pack.is_empty())
.map(content_sha256);
verify_mirror_material_with_pack_hash(
head_bytes,
identity_bytes,
feed_bytes,
snapshot_hash.as_deref(),
expected_anchor,
)
}
pub fn verify_mirror_material_with_pack_hash(
head_bytes: &[u8],
identity_bytes: &[u8],
feed_bytes: &[Vec<u8>],
snapshot_pack_sha256: Option<&str>,
expected_anchor: &str,
) -> LinkResult<VerifiedMirrorMaterial> {
let head: StoredMirrorHead = serde_json::from_slice(head_bytes)
.map_err(|_| invalid_feed("stored mirror head did not parse exactly"))?;
require_safe_ref(&head.brain)?;
if head.head_seq > MAX_FEED_REPLAY_ENTRIES || feed_bytes.len() as u64 != head.head_seq {
return Err(invalid_feed(
"stored mirror feed count does not match its bounded head sequence",
));
}
let aggregate = feed_bytes
.iter()
.try_fold(0u64, |total, bytes| total.checked_add(bytes.len() as u64))
.ok_or_else(|| invalid_feed("stored mirror feed size overflow"))?;
if aggregate > MAX_FEED_REPLAY_BYTES {
return Err(invalid_feed(
"stored mirror feed metadata exceeds the aggregate limit",
));
}
let identity: FeedIdentity = serde_json::from_slice(identity_bytes)
.map_err(|_| invalid_feed("stored mirror identity did not parse exactly"))?;
let anchor = verify_identity_chain(&identity, None)?;
if anchor != expected_anchor {
return Err(invalid_feed(
"stored mirror identity does not descend from the explicitly trusted anchor",
));
}
let mut entries = Vec::with_capacity(feed_bytes.len());
let mut items = Vec::with_capacity(feed_bytes.len());
let mut previous_hash = None;
let mut pack_sha256 = None;
for (index, bytes) in feed_bytes.iter().enumerate() {
let exact = bytes
.strip_suffix(b"\n")
.ok_or_else(|| invalid_feed("stored feed entry lacks its exact trailing newline"))?;
if exact.ends_with(b"\n") {
return Err(invalid_feed("stored feed entry has extra trailing bytes"));
}
let entry: FeedEntry = serde_json::from_slice(exact)
.map_err(|_| invalid_feed("stored feed entry did not parse exactly"))?;
let expected_seq = index as u64 + 1;
if entry.seq != expected_seq || entry.prev_entry_hash != previous_hash {
return Err(invalid_feed(
"stored mirror feed is not contiguous and hash-chained",
));
}
let canonical = serde_json::to_vec(&entry)
.map_err(|_| invalid_feed("could not canonicalize stored feed entry"))?;
if canonical != exact {
return Err(invalid_feed(
"stored feed entry is not in normative serialization",
));
}
let hash = content_sha256(bytes);
let item = FeedItem {
hash: hash.clone(),
entry,
};
verify_feed_item(&item, &identity)?;
previous_hash = Some(hash.clone());
if expected_seq == head.head_seq {
pack_sha256 = Some(item.entry.pack_sha256.clone());
}
entries.push((
expected_seq,
std::str::from_utf8(exact)
.map_err(|_| invalid_feed("stored feed entry is not UTF-8"))?
.to_string(),
hash,
));
items.push(item);
}
if previous_hash != head.feed_hash {
return Err(invalid_feed(
"stored mirror feed does not converge on its advertised head",
));
}
verify_rotation_feed_boundaries(&identity, None, &items, head.head_seq)?;
match (head.head_seq, snapshot_pack_sha256, pack_sha256.as_deref()) {
(0, None, None) => {}
(_, Some(actual), Some(expected)) if actual == expected => {}
_ => {
return Err(LinkError::InvalidPack {
message: "stored snapshot pack does not match the signed head digest".to_string(),
});
}
}
let identity_value = serde_json::to_value(&identity)
.map_err(|_| invalid_feed("could not serialize verified mirror identity"))?;
Ok(VerifiedMirrorMaterial {
brain: head.brain,
head_seq: head.head_seq,
feed_hash: head.feed_hash,
identity: identity_value,
entries,
pack_sha256,
})
}
pub fn feed_entry_hash(exact_sans_newline: &str) -> String {
format!(
"{:x}",
Sha256::digest(format!("{exact_sans_newline}\n").as_bytes())
)
}
pub fn content_sha256(bytes: &[u8]) -> String {
format!("{:x}", Sha256::digest(bytes))
}
pub fn content_sha256_reader(mut reader: impl Read) -> std::io::Result<String> {
let mut digest = Sha256::new();
let mut buffer = [0u8; 64 * 1024];
loop {
let read = reader.read(&mut buffer)?;
if read == 0 {
break;
}
digest.update(&buffer[..read]);
}
Ok(format!("{:x}", digest.finalize()))
}
pub fn mirror(cfg: &HubConfig, brain: &str, dest: &Path) -> LinkResult<MirrorReport> {
require_hardened_filesystem("mirror")?;
require_safe_ref(brain)?;
let parent = dest.parent().unwrap_or_else(|| Path::new("."));
let name = dest
.file_name()
.and_then(|name| name.to_str())
.filter(|name| !name.is_empty() && *name != "." && *name != "..")
.ok_or_else(|| LinkError::UnsafePath {
path: dest.display().to_string(),
})?;
#[cfg(unix)]
let parent_dir = open_or_create_dir_nofollow(parent)?;
#[cfg(unix)]
use std::os::fd::AsRawFd as _;
#[cfg(unix)]
let dest_name = c_name(name.as_bytes(), &dest.display().to_string())?;
#[cfg(unix)]
let dest_exists = match entry_is_dir_at(parent_dir.as_raw_fd(), &dest_name)? {
None => false,
Some(true) => true,
Some(false) => {
return Err(LinkError::UnsafePath {
path: dest.display().to_string(),
});
}
};
#[cfg(unix)]
let legacy_backup_name = c_name(
format!(".{name}.dbmd-backup").as_bytes(),
&dest.display().to_string(),
)?;
#[cfg(unix)]
if entry_is_dir_at(parent_dir.as_raw_fd(), &legacy_backup_name)?.is_some() {
return Err(LinkError::UnsafePath {
path: parent
.join(format!(".{name}.dbmd-backup"))
.display()
.to_string(),
});
}
let nonce = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let stage_label = format!(".{name}.dbmd-stage-{}-{nonce}", std::process::id());
#[cfg(unix)]
let stage_name = c_name(stage_label.as_bytes(), &dest.display().to_string())?;
#[cfg(unix)]
let stage_dir = create_dir_exclusive_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
)?;
let assembled = (|| -> LinkResult<MirrorReport> {
let remote = verified_remote_head(cfg, brain, true)?;
let brain_id = remote.head.brain.clone();
let identity = remote
.identity
.as_ref()
.ok_or_else(|| invalid_feed("brain has no signed identity"))?;
let anchor = remote
.anchor
.clone()
.ok_or_else(|| invalid_feed("brain has no verified identity anchor"))?;
let pack = download_verified_snapshot_pack(cfg, &brain_id, &remote)?;
let snapshot_entries = parse_store_pack(pack.clone())?;
let snapshot_count = snapshot_entries.len();
let mut staged_entries = snapshot_entries;
staged_entries.push((format!("{MIRROR_REL_DIR}/snapshot.pack"), pack));
for item in &remote.entries {
let mut exact = serde_json::to_vec(&item.entry)
.map_err(|_| invalid_feed("could not serialize signed feed entry"))?;
exact.push(b'\n');
if format!("{:x}", Sha256::digest(&exact)) != item.hash {
return Err(invalid_feed(
"serialized mirror entry differs from its verified hash",
));
}
staged_entries.push((
format!("{MIRROR_REL_DIR}/feed/{}.json", item.entry.seq),
exact,
));
}
let mut identity_bytes = serde_json::to_vec(identity)
.map_err(|_| invalid_feed("could not serialize verified identity chain"))?;
identity_bytes.push(b'\n');
staged_entries.push((format!("{MIRROR_REL_DIR}/identity.json"), identity_bytes));
let mut head_bytes = serde_json::to_vec(&json!({
"brain": brain_id,
"headSeq": remote.head.seq,
"feedHash": remote.head.feed_hash,
}))
.map_err(|_| invalid_feed("could not serialize verified mirror head"))?;
head_bytes.push(b'\n');
staged_entries.push((format!("{MIRROR_REL_DIR}/head.json"), head_bytes));
staged_entries.push((
CONFIG_REL_PATH.to_string(),
format!("hub = {}\npin = {anchor}\n", cfg.hub).into_bytes(),
));
#[cfg(unix)]
write_pull_entries_beneath_dir(&stage_dir, &staged_entries)?;
Ok(MirrorReport {
brain: brain_id,
head_seq: remote.head.seq,
feed_hash: remote.head.feed_hash,
entries: remote.entries.len() as u64,
pinned: anchor,
files: snapshot_count,
})
})();
let report = match assembled {
Ok(report) => report,
Err(error) => {
#[cfg(unix)]
let _ = remove_tree_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
);
return Err(error);
}
};
#[cfg(unix)]
if let Err(error) =
install_stage_at(parent_dir.as_raw_fd(), &stage_name, &dest_name, dest_exists)
{
let _ = remove_tree_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
);
return Err(error);
}
#[cfg(unix)]
if dest_exists {
remove_tree_at(
parent_dir.as_raw_fd(),
&stage_name,
&dest.display().to_string(),
)?;
}
#[cfg(unix)]
parent_dir.sync_all()?;
Ok(report)
}
fn verified_remote_head(
cfg: &HubConfig,
brain: &str,
require_full_chain: bool,
) -> LinkResult<VerifiedRemote> {
require_hardened_filesystem("verified link.md state")?;
require_safe_ref(brain)?;
let trust_directory = open_trust_dir(cfg)?;
let path = format!("/api/hub/brains/{brain}");
let body = ensure_ok(
request(cfg, "GET", &path, None, Auth::Required)?,
"subscribe",
)?;
let resolved_brain = body
.get("id")
.and_then(Value::as_str)
.filter(|id| crate::ulid::is_ulid(id))
.ok_or_else(|| invalid_feed("brain card has no canonical ULID id"))?
.to_string();
if crate::ulid::is_ulid(brain) && resolved_brain != brain {
return Err(invalid_feed(
"brain card id differs from the explicitly requested brain id",
));
}
let _trust_locks = lock_trust_many(cfg, &trust_directory, &[brain, &resolved_brain])?;
let (pinned, alias_binding) =
load_canonical_pin(cfg, &trust_directory, brain, &resolved_brain)?;
let seq = body.get("headSeq").and_then(Value::as_u64).unwrap_or(0);
let advertised_hash = body
.get("feedHash")
.and_then(Value::as_str)
.map(str::to_string);
let updated_at = body
.get("updatedAt")
.and_then(Value::as_str)
.map(str::to_string);
if let Some(pin) = &pinned {
if seq < pin.head_seq {
return Err(invalid_feed(format!(
"feed rollback: hub advertised sequence {seq}, local checkpoint is {}",
pin.head_seq
)));
}
if seq == pin.head_seq && advertised_hash != pin.feed_hash {
return Err(invalid_feed(
"feed equivocation: the checkpoint sequence now has a different hash",
));
}
}
if seq == 0 {
if advertised_hash.is_some() {
return Err(invalid_feed("an empty feed advertised a head hash"));
}
let identity: FeedIdentity = serde_json::from_value(
body.get("identity")
.cloned()
.ok_or_else(|| invalid_feed("empty brain card has no signed identity"))?,
)
.map_err(|_| invalid_feed("empty brain card has an invalid identity"))?;
let anchor = verify_identity_chain(&identity, pinned.as_ref())?;
verify_rotation_feed_boundaries(&identity, pinned.as_ref(), &[], seq)?;
save_canonical_pin_and_alias(
cfg,
&trust_directory,
brain,
&resolved_brain,
TrustState {
v: 2,
origin: normalized_origin(&cfg.hub)?,
requested: resolved_brain.clone(),
brain: resolved_brain.clone(),
home: None,
anchor: anchor.clone(),
current: format!("ed25519:{}", identity.fingerprint),
head_seq: 0,
feed_hash: None,
rotations: identity.rotations.clone(),
},
alias_binding.as_ref(),
)?;
return Ok(VerifiedRemote {
head: Head {
brain: resolved_brain,
seq,
updated_at,
feed_hash: None,
verified: true,
},
identity: Some(identity),
head_entry: None,
entries: Vec::new(),
anchor: Some(anchor),
});
}
if advertised_hash.as_ref().is_none_or(|hash| !is_sha256(hash)) {
return Err(invalid_feed(
"non-empty feed did not advertise a valid SHA-256 head",
));
}
let replay_head_only = !require_full_chain
&& pinned
.as_ref()
.is_none_or(|checkpoint| checkpoint.head_seq == seq);
let mut after = if replay_head_only {
seq - 1
} else if require_full_chain || pinned.is_none() {
0
} else {
pinned.as_ref().map_or(0, |checkpoint| checkpoint.head_seq)
};
let mut expected_seq = after + 1;
let mut previous_hash = if require_full_chain || pinned.is_none() || replay_head_only {
None
} else {
pinned
.as_ref()
.and_then(|checkpoint| checkpoint.feed_hash.clone())
};
let mut identity: Option<FeedIdentity> = None;
let mut anchor: Option<String> = None;
let mut head_entry: Option<FeedItem> = None;
let mut all_entries = Vec::new();
let mut observed_entries = Vec::new();
let replay_count = seq
.checked_sub(after)
.ok_or_else(|| invalid_feed("feed replay range moved backward"))?;
if replay_count > MAX_FEED_REPLAY_ENTRIES {
return Err(invalid_feed(format!(
"feed replay requires {replay_count} entries, over the client cap"
)));
}
let mut replay_bytes = 0u64;
loop {
let feed_bytes = ensure_raw_ok(
request_raw(
cfg,
"GET",
&format!("/api/hub/brains/{brain}/feed?after={after}&limit=100"),
None,
Auth::Required,
MAX_FEED_RESPONSE_BYTES,
)?,
"subscribe feed",
)?;
let feed: FeedResponse = serde_json::from_slice(&feed_bytes)
.map_err(|_| invalid_feed("hub returned an invalid feed shape"))?;
if feed.head_seq != seq || feed.feed_hash != advertised_hash {
return Err(invalid_feed("brain card and feed head disagree"));
}
if feed.entries.len() > FEED_PAGE_LIMIT {
return Err(invalid_feed("feed page exceeds the requested entry limit"));
}
if feed.scope_limited {
if require_full_chain {
return Err(invalid_feed(
"path-scoped grants cannot verify a full snapshot chain",
));
}
return Ok(VerifiedRemote {
head: Head {
brain: resolved_brain,
seq,
updated_at,
feed_hash: advertised_hash,
verified: false,
},
identity: None,
head_entry: None,
entries: Vec::new(),
anchor: None,
});
}
let page_identity = feed
.identity
.ok_or_else(|| invalid_feed("feed has no brain identity"))?;
let page_anchor = verify_identity_chain(&page_identity, pinned.as_ref())?;
if identity
.as_ref()
.is_some_and(|existing| existing != &page_identity)
{
return Err(invalid_feed("identity changed while reading the feed"));
}
if anchor
.as_ref()
.is_some_and(|existing| existing != &page_anchor)
{
return Err(invalid_feed(
"identity anchor changed while reading the feed",
));
}
identity = Some(page_identity.clone());
if anchor.is_none() {
anchor = Some(page_anchor);
}
if feed.entries.is_empty() {
return Err(invalid_feed("feed page was empty before the signed head"));
}
for item in feed.entries {
if item.entry.seq != expected_seq {
return Err(invalid_feed(format!(
"expected entry {expected_seq}, feed served {}",
item.entry.seq
)));
}
if item.entry.seq > seq {
return Err(invalid_feed("feed advanced past the card snapshot"));
}
if !replay_head_only && item.entry.prev_entry_hash != previous_hash {
return Err(invalid_feed(format!(
"entry {} does not chain to the local checkpoint",
item.entry.seq
)));
}
verify_feed_item(&item, &page_identity)?;
reject_retired_signer_after_checkpoint(&page_identity, pinned.as_ref(), &item)?;
replay_bytes = replay_bytes.saturating_add(
serde_json::to_vec(&item)
.map_err(|_| invalid_feed("could not size feed entry"))?
.len() as u64,
);
if replay_bytes > MAX_FEED_REPLAY_BYTES {
return Err(invalid_feed("feed replay metadata exceeds the client cap"));
}
previous_hash = Some(item.hash.clone());
after = item.entry.seq;
expected_seq = expected_seq
.checked_add(1)
.ok_or_else(|| invalid_feed("feed sequence overflow"))?;
if require_full_chain {
all_entries.push(item.clone());
}
observed_entries.push(item.clone());
head_entry = Some(item);
}
if after == seq {
break;
}
}
if head_entry.as_ref().map(|item| &item.hash) != advertised_hash.as_ref() {
return Err(invalid_feed(
"verified chain does not converge on the advertised head",
));
}
let identity = identity.ok_or_else(|| invalid_feed("feed has no brain identity"))?;
let anchor = anchor.ok_or_else(|| invalid_feed("feed has no identity anchor"))?;
verify_rotation_feed_boundaries(&identity, pinned.as_ref(), &observed_entries, seq)?;
save_canonical_pin_and_alias(
cfg,
&trust_directory,
brain,
&resolved_brain,
TrustState {
v: 2,
origin: normalized_origin(&cfg.hub)?,
requested: resolved_brain.clone(),
brain: resolved_brain.clone(),
home: None,
anchor: anchor.clone(),
current: format!("ed25519:{}", identity.fingerprint),
head_seq: seq,
feed_hash: advertised_hash.clone(),
rotations: identity.rotations.clone(),
},
alias_binding.as_ref(),
)?;
Ok(VerifiedRemote {
head: Head {
brain: resolved_brain,
seq,
updated_at,
feed_hash: advertised_hash,
verified: true,
},
identity: Some(identity),
head_entry,
entries: all_entries,
anchor: Some(anchor),
})
}
pub fn head(cfg: &HubConfig, brain: &str) -> LinkResult<Head> {
Ok(verified_remote_head(cfg, brain, false)?.head)
}
#[cfg(test)]
mod tests {
use super::*;
const TEST_BRAIN_ID: &str = "01j5qc3v9k4ym8rwbn2tqe6f7d";
#[cfg(target_os = "linux")]
#[test]
fn linux_snapshot_install_is_atomic_for_create_and_exchange() {
use std::os::fd::AsRawFd as _;
let sandbox = tempfile::TempDir::new().unwrap();
let parent = std::fs::File::open(sandbox.path()).unwrap();
let stage = std::ffi::CString::new("stage").unwrap();
let destination = std::ffi::CString::new("brain").unwrap();
std::fs::create_dir(sandbox.path().join("stage")).unwrap();
std::fs::write(sandbox.path().join("stage/value"), b"created").unwrap();
install_stage_at(
parent.as_raw_fd(),
stage.as_c_str(),
destination.as_c_str(),
false,
)
.unwrap();
assert!(!sandbox.path().join("stage").exists());
assert_eq!(
std::fs::read(sandbox.path().join("brain/value")).unwrap(),
b"created"
);
std::fs::create_dir(sandbox.path().join("stage")).unwrap();
std::fs::write(sandbox.path().join("stage/value"), b"replacement").unwrap();
install_stage_at(
parent.as_raw_fd(),
stage.as_c_str(),
destination.as_c_str(),
true,
)
.unwrap();
assert_eq!(
std::fs::read(sandbox.path().join("brain/value")).unwrap(),
b"replacement"
);
assert_eq!(
std::fs::read(sandbox.path().join("stage/value")).unwrap(),
b"created",
"RENAME_EXCHANGE must leave the predecessor at the unique stage name"
);
}
struct SignedRemoteFixture {
card: String,
feed: String,
key: AgentSigningKey,
identity: FeedIdentity,
}
fn signed_remote_fixture() -> SignedRemoteFixture {
let rng = ring::rand::SystemRandom::new();
let pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng).unwrap();
let pair = ring::signature::Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap();
let (public_key, multikey) = public_identity_for(&pair);
let identity = FeedIdentity {
fingerprint: multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: public_key.clone(),
previous: Vec::new(),
rotations: Vec::new(),
};
let mut entry = FeedEntry {
v: 1,
seq: 1,
ts: "2026-07-30T12:00:00.000Z".to_string(),
brain: multikey.clone(),
public_key: public_key.clone(),
kind: "push".to_string(),
op: "snapshot".to_string(),
pack_sha256: "a".repeat(64),
files: Vec::new(),
removed: Vec::new(),
prev_entry_hash: None,
sig: String::new(),
};
let unsigned = UnsignedFeedEntry {
v: entry.v,
seq: entry.seq,
ts: &entry.ts,
brain: &entry.brain,
public_key: &entry.public_key,
kind: &entry.kind,
op: &entry.op,
pack_sha256: &entry.pack_sha256,
files: &entry.files,
removed: &entry.removed,
prev_entry_hash: &entry.prev_entry_hash,
};
entry.sig =
URL_SAFE_NO_PAD.encode(pair.sign(&serde_json::to_vec(&unsigned).unwrap()).as_ref());
let mut exact = serde_json::to_vec(&entry).unwrap();
exact.push(b'\n');
let hash = content_sha256(&exact);
let card = json!({
"id": TEST_BRAIN_ID,
"headSeq": 1,
"feedHash": hash,
"identity": identity.clone(),
})
.to_string();
let feed = json!({
"headSeq": 1,
"feedHash": hash,
"identity": identity.clone(),
"entries": [{"hash": hash, "entry": entry}],
"scopeLimited": false,
})
.to_string();
SignedRemoteFixture {
card,
feed,
key: AgentSigningKey {
pkcs8: pkcs8.as_ref().to_vec(),
multikey,
public_key_spki: public_key,
},
identity,
}
}
#[test]
fn wire_sequences_are_rejected_at_their_endpoint_count_limits() {
let fixture = signed_remote_fixture();
let feed: Value = serde_json::from_str(&fixture.feed).unwrap();
let item = feed["entries"][0].to_string();
let oversized_page = format!(
"{{\"headSeq\":1,\"feedHash\":null,\"identity\":null,\"entries\":[{}],\"scopeLimited\":false}}",
std::iter::repeat_n(item.as_str(), FEED_PAGE_LIMIT + 1)
.collect::<Vec<_>>()
.join(",")
);
assert!(serde_json::from_str::<FeedResponse>(&oversized_page).is_err());
let oversized_identity = format!(
"{{\"fingerprint\":\"fp\",\"publicKeySpki\":\"spki\",\"previous\":[],\"rotations\":[{}]}}",
std::iter::repeat_n("\"rotation\"", MAX_IDENTITY_ROTATIONS + 1)
.collect::<Vec<_>>()
.join(",")
);
assert!(serde_json::from_str::<FeedIdentity>(&oversized_identity).is_err());
let file = r#"{"path":"a","sha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","bytes":1}"#;
let oversized_entry = format!(
"{{\"v\":1,\"seq\":1,\"ts\":\"t\",\"brain\":\"b\",\"public_key\":\"k\",\"kind\":\"push\",\"op\":\"snapshot\",\"pack_sha256\":\"{}\",\"files\":[{}],\"removed\":[],\"prev_entry_hash\":null,\"sig\":\"s\"}}",
"a".repeat(64),
std::iter::repeat_n(file, MAX_PUSH_FILES + 1)
.collect::<Vec<_>>()
.join(",")
);
assert!(serde_json::from_str::<FeedEntry>(&oversized_entry).is_err());
}
fn scripted_json_hub(responses: Vec<(u16, String)>) -> (String, std::thread::JoinHandle<()>) {
use std::io::{BufRead as _, BufReader, Read as _, Write as _};
use std::net::TcpListener;
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let handle = std::thread::spawn(move || {
for (status, body) in responses {
let (stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream);
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let mut content_length = 0usize;
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" || line == "\n" || line.is_empty() {
break;
}
if let Some((name, value)) = line.split_once(':') {
if name.eq_ignore_ascii_case("content-length") {
content_length = value.trim().parse().unwrap();
}
}
}
let mut request_body = vec![0_u8; content_length];
reader.read_exact(&mut request_body).unwrap();
let response = format!(
"HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len()
);
reader.get_mut().write_all(response.as_bytes()).unwrap();
}
});
(url, handle)
}
fn routed_json_hub(
requests: usize,
mut respond: impl FnMut(&str) -> (u16, String) + Send + 'static,
) -> (String, std::thread::JoinHandle<()>) {
use std::io::{BufRead as _, BufReader, Read as _, Write as _};
use std::net::TcpListener;
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}", listener.local_addr().unwrap());
let handle = std::thread::spawn(move || {
for _ in 0..requests {
let (stream, _) = listener.accept().unwrap();
let mut reader = BufReader::new(stream);
let mut line = String::new();
reader.read_line(&mut line).unwrap();
let path = line.split_whitespace().nth(1).unwrap_or("").to_string();
let mut content_length = 0usize;
loop {
line.clear();
reader.read_line(&mut line).unwrap();
if line == "\r\n" || line == "\n" || line.is_empty() {
break;
}
if let Some((name, value)) = line.split_once(':') {
if name.eq_ignore_ascii_case("content-length") {
content_length = value.trim().parse().unwrap();
}
}
}
let mut request_body = vec![0_u8; content_length];
reader.read_exact(&mut request_body).unwrap();
let (status, body) = respond(&path);
let response = format!(
"HTTP/1.1 {status} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
body.len()
);
reader.get_mut().write_all(response.as_bytes()).unwrap();
}
});
(url, handle)
}
fn test_hub_config(hub: String, state_dir: PathBuf) -> HubConfig {
HubConfig {
hub,
key: Some("test-key".to_string()),
agent_key: None,
brain_key: None,
state_dir,
store_selected: false,
}
}
#[test]
fn linkmd_sig_v2_is_bound_to_the_exact_hub_origin() {
use ring::signature::KeyPair as _;
let rng = ring::rand::SystemRandom::new();
let pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng).unwrap();
let pair = ring::signature::Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap();
let (spki, multikey) = public_identity_for(&pair);
let key = AgentSigningKey {
pkcs8: pkcs8.as_ref().to_vec(),
multikey,
public_key_spki: spki,
};
let header = linkmd_sig_header(
&key,
"https://hub-a.example",
"post",
"/api/hub/brains/brain/push?mode=exact",
Some("{\"ok\":true}"),
)
.unwrap();
assert!(header.starts_with("LinkMD-Sig v2,"));
let ts = header
.split(",ts=")
.nth(1)
.unwrap()
.split(',')
.next()
.unwrap();
let signature = URL_SAFE_NO_PAD
.decode(header.rsplit(",sig=").next().unwrap())
.unwrap();
let body_hash = format!("{:x}", Sha256::digest(b"{\"ok\":true}"));
let accepted = format!(
"v2\nhttps://hub-a.example\nPOST\n/api/hub/brains/brain/push?mode=exact\n{ts}\n{body_hash}"
);
let replayed = format!(
"v2\nhttps://hub-b.example\nPOST\n/api/hub/brains/brain/push?mode=exact\n{ts}\n{body_hash}"
);
let public = pair.public_key().as_ref();
assert!(UnparsedPublicKey::new(&ED25519, public)
.verify(accepted.as_bytes(), &signature)
.is_ok());
assert!(
UnparsedPublicKey::new(&ED25519, public)
.verify(replayed.as_bytes(), &signature)
.is_err(),
"a proof captured at hub A must not authenticate at hub B"
);
}
#[test]
fn explicit_brain_id_rejects_a_substituted_card_before_feed_trust() {
let other = "01j5qc3v9k4ym8rwbn2tqe6f7e";
let card = json!({
"id": other,
"headSeq": 0,
"identity": signed_remote_fixture().identity,
})
.to_string();
let (hub, server) = scripted_json_hub(vec![(200, card)]);
let state = tempfile::tempdir().unwrap();
let cfg = test_hub_config(hub, state.path().to_path_buf());
let error = head(&cfg, TEST_BRAIN_ID).unwrap_err().to_string();
assert!(
error.contains("differs from the explicitly requested"),
"{error}"
);
server.join().unwrap();
}
#[test]
fn empty_brain_identity_is_pinned_and_cannot_be_substituted_later() {
let first = signed_remote_fixture().identity;
let second = signed_remote_fixture().identity;
let card = |identity: FeedIdentity| {
json!({
"id": TEST_BRAIN_ID,
"headSeq": 0,
"identity": identity,
})
.to_string()
};
let (hub, server) = scripted_json_hub(vec![(200, card(first)), (200, card(second))]);
let state = tempfile::tempdir().unwrap();
let cfg = test_hub_config(hub, state.path().to_path_buf());
assert!(head(&cfg, TEST_BRAIN_ID).unwrap().verified);
let error = head(&cfg, TEST_BRAIN_ID).unwrap_err().to_string();
assert!(
error.contains("pinned anchor") || error.contains("forked away"),
"{error}"
);
server.join().unwrap();
}
#[test]
fn empty_head_rejects_a_rotation_that_commits_to_hidden_history() {
let old = signed_remote_fixture();
let new = signed_remote_fixture();
let old_pair = ring::signature::Ed25519KeyPair::from_pkcs8(&old.key.pkcs8).unwrap();
let unsigned = serde_json::to_string(&UnsignedRotation {
v: 1,
op: "rotate",
brain: &old.key.multikey,
public_key: &old.key.public_key_spki,
new_brain: &new.key.multikey,
new_public_key: &new.key.public_key_spki,
prior_head_seq: 1,
prior_feed_hash: Some(&"a".repeat(64)),
ts: "2026-07-30T12:00:00.000Z".to_string(),
})
.unwrap();
let signature = URL_SAFE_NO_PAD.encode(old_pair.sign(unsigned.as_bytes()).as_ref());
let rotation = format!(
"{},\"sig\":\"{}\"}}",
&unsigned[..unsigned.len() - 1],
signature
);
let identity = FeedIdentity {
fingerprint: new.key.multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: new.key.public_key_spki,
previous: vec![PreviousIdentity {
fingerprint: old.key.multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: old.key.public_key_spki,
}],
rotations: vec![rotation],
};
let card = json!({
"id": TEST_BRAIN_ID,
"headSeq": 0,
"feedHash": null,
"identity": identity,
})
.to_string();
let (hub, server) = scripted_json_hub(vec![(200, card)]);
let state = tempfile::tempdir().unwrap();
let cfg = test_hub_config(hub, state.path().to_path_buf());
let error = head(&cfg, TEST_BRAIN_ID).unwrap_err().to_string();
assert!(
error.contains("rotation claims a feed boundary beyond the advertised head"),
"{error}"
);
assert!(
load_trust(&cfg, TEST_BRAIN_ID).unwrap().is_none(),
"an inconsistent empty-head identity must not become the TOFU checkpoint"
);
server.join().unwrap();
}
#[test]
fn trust_checkpoint_rejects_a_later_fork() {
let fixture = signed_remote_fixture();
let mut fork: Value = serde_json::from_str(&fixture.card).unwrap();
fork["feedHash"] = Value::String("b".repeat(64));
let (hub, server) = scripted_json_hub(vec![
(200, fixture.card),
(200, fixture.feed),
(200, fork.to_string()),
]);
let state = tempfile::tempdir().unwrap();
let cfg = test_hub_config(hub, state.path().to_path_buf());
assert!(head(&cfg, TEST_BRAIN_ID).unwrap().verified);
assert!(head(&cfg, TEST_BRAIN_ID).is_err());
server.join().unwrap();
}
#[test]
fn alias_and_canonical_id_share_one_identity_checkpoint() {
let trusted = signed_remote_fixture();
let attacker = signed_remote_fixture();
let (hub, server) = scripted_json_hub(vec![
(200, trusted.card),
(200, trusted.feed),
(200, attacker.card),
]);
let state = tempfile::tempdir().unwrap();
let cfg = test_hub_config(hub, state.path().to_path_buf());
assert!(head(&cfg, "trusted-slug").unwrap().verified);
let error = head(&cfg, TEST_BRAIN_ID).unwrap_err().to_string();
assert!(
error.contains("equivocation")
|| error.contains("pinned")
|| error.contains("identity"),
"{error}"
);
server.join().unwrap();
}
#[test]
fn concurrent_aliases_cannot_establish_conflicting_tofu_pins() {
let alpha = signed_remote_fixture();
let beta = signed_remote_fixture();
let alpha_card = alpha.card.clone();
let alpha_feed = alpha.feed.clone();
let beta_card = beta.card.clone();
let beta_feed = beta.feed.clone();
let (hub, server) = routed_json_hub(3, move |path| {
if path.contains("/alpha/feed?") {
(200, alpha_feed.clone())
} else if path.contains("/beta/feed?") {
(200, beta_feed.clone())
} else if path.ends_with("/alpha") {
(200, alpha_card.clone())
} else if path.ends_with("/beta") {
(200, beta_card.clone())
} else {
(500, r#"{"error":"unexpected path"}"#.to_string())
}
});
let state = tempfile::tempdir().unwrap();
let cfg = test_hub_config(hub, state.path().to_path_buf());
let alpha_cfg = cfg.clone();
let beta_cfg = cfg;
let first = std::thread::spawn(move || head(&alpha_cfg, "alpha"));
let second = std::thread::spawn(move || head(&beta_cfg, "beta"));
let results = [first.join().unwrap(), second.join().unwrap()];
assert_eq!(
results.iter().filter(|result| result.is_ok()).count(),
1,
"only one alias identity may establish canonical TOFU: {results:?}"
);
assert_eq!(results.iter().filter(|result| result.is_err()).count(), 1);
server.join().unwrap();
}
#[cfg(unix)]
#[test]
fn trust_transaction_survives_an_ancestor_swap_without_writing_outside() {
use std::os::unix::fs::symlink;
let fixture = signed_remote_fixture();
let card = json!({
"id": TEST_BRAIN_ID,
"headSeq": 0,
"feedHash": Value::Null,
"identity": fixture.identity,
})
.to_string();
let work = tempfile::tempdir().unwrap();
let outside = tempfile::tempdir().unwrap();
let state = work.path().join("state");
let moved = work.path().join("state-held");
let swap_state = state.clone();
let swap_moved = moved.clone();
let outside_path = outside.path().to_path_buf();
let (hub, server) = routed_json_hub(1, move |_| {
std::fs::rename(&swap_state, &swap_moved).unwrap();
symlink(&outside_path, &swap_state).unwrap();
(200, card.clone())
});
let cfg = test_hub_config(hub, state);
let verified = verified_remote_head(&cfg, TEST_BRAIN_ID, false).unwrap();
assert_eq!(verified.head.seq, 0);
assert_eq!(std::fs::read_dir(outside.path()).unwrap().count(), 0);
assert!(std::fs::read_dir(moved.join("trust"))
.unwrap()
.flatten()
.any(|entry| entry.path().extension().is_some_and(|ext| ext == "json")));
server.join().unwrap();
}
#[test]
fn self_custody_push_refuses_an_unrelated_key_before_requesting_an_upload() {
let remote = signed_remote_fixture();
let unrelated = signed_remote_fixture().key;
let (hub, server) = scripted_json_hub(vec![(200, remote.card), (200, remote.feed)]);
let state = tempfile::tempdir().unwrap();
let mut cfg = test_hub_config(hub, state.path().to_path_buf());
cfg.brain_key = Some(unrelated);
let error = sync_push(
&cfg,
TEST_BRAIN_ID,
&[("DB.md".to_string(), "signed local content".to_string())],
)
.unwrap_err()
.to_string();
assert!(
error.contains("not the verified current brain identity"),
"{error}"
);
server.join().unwrap();
}
#[test]
fn rotation_ignores_a_forged_2xx_body_and_requires_verified_postcondition() {
let remote = signed_remote_fixture();
let new = signed_remote_fixture().key;
let state = tempfile::tempdir().unwrap();
let new_file = state.path().join("new.key");
std::fs::write(
&new_file,
format!("{}\n", URL_SAFE_NO_PAD.encode(&new.pkcs8)),
)
.unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt as _;
std::fs::set_permissions(&new_file, std::fs::Permissions::from_mode(0o600)).unwrap();
}
let forged = json!({
"brain": TEST_BRAIN_ID,
"identity": {
"fingerprint": new.multikey.trim_start_matches("ed25519:"),
"publicKeySpki": new.public_key_spki,
}
})
.to_string();
let (hub, server) = scripted_json_hub(vec![
(200, remote.card.clone()),
(200, remote.feed.clone()),
(200, forged),
(200, remote.card),
(200, remote.feed),
]);
let cfg = test_hub_config(hub, state.path().to_path_buf());
let error = rotate_brain_key(&cfg, TEST_BRAIN_ID, &remote.key, &new_file)
.unwrap_err()
.to_string();
assert!(
error.contains("without committing the verified new identity"),
"{error}"
);
server.join().unwrap();
}
#[test]
fn resolve_record_is_derived_from_signed_pack_not_a_query_response() {
let record_id = "01j5qc3v9k4ym8rwbn2tqe6f7e";
let raw = format!(
"---\ntype: client\nid: {record_id}\nsummary: Signed truth\n---\n# Signed truth\n"
);
let pack = build_store_pack(&[
(
"DB.md".to_string(),
"---\ntype: db-md\nscope: company\nowner: Test\n---\n".to_string(),
),
("records/clients/truth.md".to_string(), raw.clone()),
])
.unwrap();
let by_id = resolve_from_verified_pack(
"01j5qc3v9k4ym8rwbn2tqe6f7d",
&AddressTarget::Id(record_id.to_string()),
pack.clone(),
)
.unwrap();
assert_eq!(by_id["document"]["summary"], "Signed truth");
assert_eq!(by_id["document"]["body"], "# Signed truth\n");
assert_eq!(
by_id["document"]["contentSha"],
content_sha256(raw.as_bytes())
);
let by_path = resolve_from_verified_pack(
"01j5qc3v9k4ym8rwbn2tqe6f7d",
&AddressTarget::Path("records/clients/truth.md".to_string()),
pack,
)
.unwrap();
assert_eq!(by_path["document"]["id"], record_id);
assert_eq!(by_path["document"]["path"], "records/clients/truth.md");
}
#[test]
fn canonical_store_pack_matches_the_cross_language_zip32_golden() {
let unsorted = vec![
("records/a.md".to_string(), "alpha\n".to_string()),
("DB.md".to_string(), "# db\n".to_string()),
];
let sorted = vec![
("DB.md".to_string(), "# db\n".to_string()),
("records/a.md".to_string(), "alpha\n".to_string()),
];
let pack = build_store_pack(&unsorted).unwrap();
assert_eq!(pack.len(), 219);
assert_eq!(
content_sha256(&pack),
"972fb2045becaa21588baaf4b349e62a430687fa2c21167b53f4ca0efa6c9408"
);
assert_eq!(pack, build_store_pack(&sorted).unwrap());
assert!(!pack.windows(4).any(|bytes| bytes == b"PK\x07\x08"));
assert!(!pack.windows(4).any(|bytes| bytes == b"PK\x06\x06"));
assert!(!pack.windows(4).any(|bytes| bytes == b"PK\x06\x07"));
assert_eq!(
parse_store_pack(pack).unwrap(),
vec![
("DB.md".to_string(), b"# db\n".to_vec()),
("records/a.md".to_string(), b"alpha\n".to_vec()),
]
);
}
#[test]
fn canonical_store_pack_validates_every_path_before_writing() {
let duplicate = vec![
("DB.md".to_string(), "first".to_string()),
("DB.md".to_string(), "second".to_string()),
];
assert!(build_store_pack(&duplicate)
.unwrap_err()
.to_string()
.contains("duplicate path"));
assert!(matches!(
build_store_pack(&[("../escape.md".to_string(), "no".to_string())]),
Err(LinkError::UnsafePath { .. })
));
}
#[test]
fn zip64_preflight_rejects_an_entry_count_bomb_before_zip_parsing() {
const COUNT: u64 = MAX_PUSH_FILES as u64 + 1;
let mut bytes = vec![0_u8];
let zip64_offset = bytes.len() as u64;
bytes.extend_from_slice(b"PK\x06\x06");
bytes.extend_from_slice(&44_u64.to_le_bytes());
bytes.extend_from_slice(&[0_u8; 12]);
bytes.extend_from_slice(&COUNT.to_le_bytes());
bytes.extend_from_slice(&COUNT.to_le_bytes());
bytes.extend_from_slice(&1_u64.to_le_bytes());
bytes.extend_from_slice(&0_u64.to_le_bytes());
bytes.extend_from_slice(b"PK\x06\x07");
bytes.extend_from_slice(&0_u32.to_le_bytes());
bytes.extend_from_slice(&zip64_offset.to_le_bytes());
bytes.extend_from_slice(&1_u32.to_le_bytes());
bytes.extend_from_slice(b"PK\x05\x06");
bytes.extend_from_slice(&0_u16.to_le_bytes());
bytes.extend_from_slice(&0_u16.to_le_bytes());
bytes.extend_from_slice(&u16::MAX.to_le_bytes());
bytes.extend_from_slice(&u16::MAX.to_le_bytes());
bytes.extend_from_slice(&u32::MAX.to_le_bytes());
bytes.extend_from_slice(&u32::MAX.to_le_bytes());
bytes.extend_from_slice(&0_u16.to_le_bytes());
let error = preflight_zip_entry_count(&bytes, MAX_PUSH_FILES)
.unwrap_err()
.to_string();
assert!(error.contains("invalid file count"), "{error}");
}
#[test]
fn zip_preflight_rejects_a_fake_trailing_eocd_before_zip_parsing() {
const COUNT: u64 = MAX_PUSH_FILES as u64 + 1;
let mut bytes = vec![0_u8];
let zip64_offset = bytes.len() as u64;
bytes.extend_from_slice(b"PK\x06\x06");
bytes.extend_from_slice(&44_u64.to_le_bytes());
bytes.extend_from_slice(&[0_u8; 12]);
bytes.extend_from_slice(&COUNT.to_le_bytes());
bytes.extend_from_slice(&COUNT.to_le_bytes());
bytes.extend_from_slice(&1_u64.to_le_bytes());
bytes.extend_from_slice(&0_u64.to_le_bytes());
bytes.extend_from_slice(b"PK\x06\x07");
bytes.extend_from_slice(&0_u32.to_le_bytes());
bytes.extend_from_slice(&zip64_offset.to_le_bytes());
bytes.extend_from_slice(&1_u32.to_le_bytes());
bytes.extend_from_slice(b"PK\x05\x06");
bytes.extend_from_slice(&0_u16.to_le_bytes());
bytes.extend_from_slice(&0_u16.to_le_bytes());
bytes.extend_from_slice(&u16::MAX.to_le_bytes());
bytes.extend_from_slice(&u16::MAX.to_le_bytes());
bytes.extend_from_slice(&u32::MAX.to_le_bytes());
bytes.extend_from_slice(&u32::MAX.to_le_bytes());
bytes.extend_from_slice(&0_u16.to_le_bytes());
let fake_eocd = bytes.len() as u32;
bytes.extend_from_slice(b"PK\x05\x06");
bytes.extend_from_slice(&0_u16.to_le_bytes());
bytes.extend_from_slice(&0_u16.to_le_bytes());
bytes.extend_from_slice(&1_u16.to_le_bytes());
bytes.extend_from_slice(&1_u16.to_le_bytes());
bytes.extend_from_slice(&0_u32.to_le_bytes());
bytes.extend_from_slice(&fake_eocd.to_le_bytes());
bytes.extend_from_slice(&0_u16.to_le_bytes());
let error = preflight_zip_entry_count(&bytes, MAX_PUSH_FILES)
.unwrap_err()
.to_string();
assert!(error.contains("central directory"), "{error}");
}
#[test]
fn strict_http_status_handling_rejects_redirects_without_panicking() {
let error = ensure_ok(
HubResponse {
status: 302,
body: Some(json!({"redirect": "/elsewhere"})),
},
"mutation",
)
.unwrap_err();
assert!(matches!(error, LinkError::Http { status: 302, .. }));
let error = ensure_raw_ok(
RawHubResponse {
status: 302,
body: br#"{"redirect":"/elsewhere"}"#.to_vec(),
},
"feed",
)
.unwrap_err();
assert!(matches!(error, LinkError::Http { status: 302, .. }));
}
#[cfg(unix)]
#[test]
fn collect_push_files_refuses_external_symlink_and_nested_store() {
use std::os::unix::fs::symlink;
let root = tempfile::tempdir().unwrap();
std::fs::write(
root.path().join("DB.md"),
"---\ntype: db-md\nscope: company\nowner: Test\n---\n",
)
.unwrap();
std::fs::create_dir_all(root.path().join("records/notes")).unwrap();
let external = tempfile::tempdir().unwrap();
let secret = external.path().join("secret.md");
std::fs::write(&secret, "TOP SECRET").unwrap();
symlink(&secret, root.path().join("records/notes/secret.md")).unwrap();
let store = Store::open_strict(root.path()).unwrap();
let err = collect_push_files(&store).unwrap_err().to_string();
assert!(err.contains("cannot push"), "{err}");
assert!(
!err.contains("TOP SECRET"),
"external bytes must never leak"
);
std::fs::remove_file(root.path().join("records/notes/secret.md")).unwrap();
let nested = root.path().join("records/nested");
std::fs::create_dir_all(&nested).unwrap();
std::fs::write(
nested.join("DB.md"),
"---\ntype: db-md\nscope: research\nowner: Nested\n---\n",
)
.unwrap();
let err = collect_push_files(&store).unwrap_err().to_string();
assert!(err.contains("nested db.md store"), "{err}");
}
#[cfg(unix)]
#[test]
fn remote_push_uses_opened_root_after_path_replacement() {
use std::os::unix::fs::symlink;
let sandbox = tempfile::tempdir().unwrap();
let root = sandbox.path().join("store");
std::fs::create_dir_all(root.join("records/notes")).unwrap();
std::fs::write(root.join("DB.md"), "---\ntype: db-md\n---\n").unwrap();
std::fs::write(
root.join("records/notes/owned.md"),
"---\ntype: note\nsummary: owned\n---\nowned upload\n",
)
.unwrap();
let store = Store::open_strict(&root).unwrap();
let detached = sandbox.path().join("detached");
std::fs::rename(&root, &detached).unwrap();
let replacement = sandbox.path().join("replacement");
std::fs::create_dir_all(replacement.join("records/notes")).unwrap();
std::fs::write(replacement.join("DB.md"), "---\ntype: db-md\n---\n").unwrap();
std::fs::write(
replacement.join("records/notes/secret.md"),
"---\ntype: note\nsummary: secret\n---\nreplacement sentinel\n",
)
.unwrap();
symlink(&replacement, &root).unwrap();
let files = collect_push_files(&store).unwrap();
let wire_text = files
.iter()
.map(|(path, content)| format!("{path}\n{content}"))
.collect::<Vec<_>>()
.join("\n");
assert!(wire_text.contains("owned upload"));
assert!(!wire_text.contains("replacement sentinel"));
assert!(!wire_text.contains("records/notes/secret.md"));
let remote = signed_remote_fixture();
let (hub, server) = scripted_json_hub(vec![
(200, remote.card),
(200, remote.feed),
(200, json!({"ok": true}).to_string()),
]);
let state = tempfile::tempdir().unwrap();
let cfg = test_hub_config(hub, state.path().to_path_buf());
let pushed = sync_push(&cfg, TEST_BRAIN_ID, &files).unwrap();
assert_eq!(pushed, json!({"ok": true}));
server.join().unwrap();
}
#[test]
fn signed_feed_item_verifies_identity_hash_and_signature() {
use ring::rand::SystemRandom;
use ring::signature::{Ed25519KeyPair, KeyPair};
const PREFIX: &[u8] = &[
0x30, 0x2a, 0x30, 0x05, 0x06, 0x03, 0x2b, 0x65, 0x70, 0x03, 0x21, 0x00,
];
let pkcs8 = Ed25519KeyPair::generate_pkcs8(&SystemRandom::new()).unwrap();
let pair = Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap();
let mut spki = PREFIX.to_vec();
spki.extend_from_slice(pair.public_key().as_ref());
let public_key = URL_SAFE_NO_PAD.encode(&spki);
let fingerprint = URL_SAFE_NO_PAD.encode(Sha256::digest(&spki));
let mut entry = FeedEntry {
v: 1,
seq: 1,
ts: "2026-07-14T00:00:00.000Z".to_string(),
brain: format!("ed25519:{fingerprint}"),
public_key: public_key.clone(),
kind: "push".to_string(),
op: "snapshot".to_string(),
pack_sha256: "a".repeat(64),
files: vec![FeedFile {
path: "DB.md".to_string(),
sha256: "b".repeat(64),
bytes: 3,
}],
removed: vec![],
prev_entry_hash: None,
sig: String::new(),
};
let unsigned = UnsignedFeedEntry {
v: entry.v,
seq: entry.seq,
ts: &entry.ts,
brain: &entry.brain,
public_key: &entry.public_key,
kind: &entry.kind,
op: &entry.op,
pack_sha256: &entry.pack_sha256,
files: &entry.files,
removed: &entry.removed,
prev_entry_hash: &entry.prev_entry_hash,
};
entry.sig =
URL_SAFE_NO_PAD.encode(pair.sign(&serde_json::to_vec(&unsigned).unwrap()).as_ref());
let mut exact = serde_json::to_vec(&entry).unwrap();
exact.push(b'\n');
let item = FeedItem {
hash: format!("{:x}", Sha256::digest(&exact)),
entry,
};
let identity = FeedIdentity {
fingerprint,
public_key_spki: public_key,
previous: Vec::new(),
rotations: Vec::new(),
};
assert!(verify_feed_item(&item, &identity).is_ok());
let mut tampered = item;
tampered.entry.pack_sha256 = "c".repeat(64);
assert!(verify_feed_item(&tampered, &identity).is_err());
}
#[test]
fn a_self_custody_entry_verifies_like_any_hub_entry() {
let rng = ring::rand::SystemRandom::new();
let pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng).unwrap();
let pair = ring::signature::Ed25519KeyPair::from_pkcs8(pkcs8.as_ref()).unwrap();
let (spki, multikey) = public_identity_for(&pair);
let key = AgentSigningKey {
pkcs8: pkcs8.as_ref().to_vec(),
multikey: multikey.clone(),
public_key_spki: spki.clone(),
};
let files = vec![WireFeedFile {
path: "DB.md".to_string(),
sha256: "a".repeat(64),
bytes: 3,
}];
let raw = self_custody_entry(
&key,
1,
"2026-07-23T12:00:00.000Z".to_string(),
&"c".repeat(64),
&files,
None,
)
.unwrap();
let entry: FeedEntry = serde_json::from_str(&raw).unwrap();
let hash = format!("{:x}", Sha256::digest(format!("{raw}\n").as_bytes()));
let item = FeedItem { hash, entry };
let identity = FeedIdentity {
fingerprint: multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: spki,
previous: Vec::new(),
rotations: Vec::new(),
};
assert!(verify_feed_item(&item, &identity).is_ok());
}
#[test]
fn identity_rotation_requires_old_key_signature_and_preserves_the_pin() {
let rng = ring::rand::SystemRandom::new();
let old_pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng).unwrap();
let old = ring::signature::Ed25519KeyPair::from_pkcs8(old_pkcs8.as_ref()).unwrap();
let new_pkcs8 = ring::signature::Ed25519KeyPair::generate_pkcs8(&rng).unwrap();
let new = ring::signature::Ed25519KeyPair::from_pkcs8(new_pkcs8.as_ref()).unwrap();
let (old_spki, old_multikey) = public_identity_for(&old);
let (new_spki, new_multikey) = public_identity_for(&new);
let unsigned = serde_json::to_string(&UnsignedRotation {
v: 1,
op: "rotate",
brain: &old_multikey,
public_key: &old_spki,
new_brain: &new_multikey,
new_public_key: &new_spki,
prior_head_seq: 1,
prior_feed_hash: Some(&"a".repeat(64)),
ts: "2026-07-30T12:00:00.000Z".to_string(),
})
.unwrap();
let signature = URL_SAFE_NO_PAD.encode(old.sign(unsigned.as_bytes()).as_ref());
let rotation = format!(
"{},\"sig\":\"{}\"}}",
&unsigned[..unsigned.len() - 1],
signature
);
let identity = FeedIdentity {
fingerprint: new_multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: new_spki,
previous: vec![PreviousIdentity {
fingerprint: old_multikey.trim_start_matches("ed25519:").to_string(),
public_key_spki: old_spki,
}],
rotations: vec![rotation],
};
let pin = TrustState {
v: 2,
origin: "https://hub.example".to_string(),
requested: "brain".to_string(),
brain: "brain".to_string(),
home: None,
anchor: old_multikey.clone(),
current: old_multikey.clone(),
head_seq: 1,
feed_hash: Some("a".repeat(64)),
rotations: Vec::new(),
};
assert_eq!(
verify_identity_chain(&identity, Some(&pin)).unwrap(),
old_multikey
);
let mut accepted = pin.clone();
accepted.current = new_multikey.clone();
accepted.rotations = identity.rotations.clone();
let alternate_unsigned = serde_json::to_string(&UnsignedRotation {
v: 1,
op: "rotate",
brain: &old_multikey,
public_key: &identity.previous[0].public_key_spki,
new_brain: &new_multikey,
new_public_key: &identity.public_key_spki,
prior_head_seq: 1,
prior_feed_hash: Some(&"a".repeat(64)),
ts: "2026-07-30T12:00:01.000Z".to_string(),
})
.unwrap();
let alternate_signature =
URL_SAFE_NO_PAD.encode(old.sign(alternate_unsigned.as_bytes()).as_ref());
let mut rewritten = identity.clone();
rewritten.rotations[0] = format!(
"{},\"sig\":\"{}\"}}",
&alternate_unsigned[..alternate_unsigned.len() - 1],
alternate_signature
);
assert!(
verify_identity_chain(&rewritten, Some(&accepted)).is_err(),
"an alternate valid statement must not rewrite accepted history"
);
let mut stale_entry = FeedEntry {
v: 1,
seq: 2,
ts: "2026-07-30T12:01:00.000Z".to_string(),
brain: pin.current.clone(),
public_key: identity.previous[0].public_key_spki.clone(),
kind: "push".to_string(),
op: "snapshot".to_string(),
pack_sha256: "b".repeat(64),
files: Vec::new(),
removed: Vec::new(),
prev_entry_hash: pin.feed_hash.clone(),
sig: String::new(),
};
let stale_unsigned = UnsignedFeedEntry {
v: stale_entry.v,
seq: stale_entry.seq,
ts: &stale_entry.ts,
brain: &stale_entry.brain,
public_key: &stale_entry.public_key,
kind: &stale_entry.kind,
op: &stale_entry.op,
pack_sha256: &stale_entry.pack_sha256,
files: &stale_entry.files,
removed: &stale_entry.removed,
prev_entry_hash: &stale_entry.prev_entry_hash,
};
stale_entry.sig = URL_SAFE_NO_PAD.encode(
old.sign(&serde_json::to_vec(&stale_unsigned).unwrap())
.as_ref(),
);
let mut stale_exact = serde_json::to_vec(&stale_entry).unwrap();
stale_exact.push(b'\n');
let stale_item = FeedItem {
hash: content_sha256(&stale_exact),
entry: stale_entry,
};
assert!(
reject_retired_signer_after_checkpoint(&identity, Some(&accepted), &stale_item)
.is_err(),
"a key retired before the checkpoint must never append after it"
);
assert!(
verify_feed_item(&stale_item, &identity).is_err(),
"an old key must never append after its signed rotation boundary"
);
let mut missing = identity.clone();
missing.rotations.clear();
assert!(verify_identity_chain(&missing, Some(&pin)).is_err());
let mut tampered = identity;
tampered.rotations[0] = tampered.rotations[0].replace("\"sig\":\"", "\"sig\":\"A");
assert!(verify_identity_chain(&tampered, Some(&pin)).is_err());
}
#[cfg(unix)]
#[test]
fn key_creation_refuses_a_planted_symlink_without_touching_its_target() {
use std::os::unix::fs::symlink;
let dir = tempfile::tempdir().unwrap();
let target = dir.path().join("valuable.txt");
let planted = dir.path().join("agent.key");
std::fs::write(&target, "do not overwrite").unwrap();
symlink(&target, &planted).unwrap();
assert!(matches!(
generate_agent_key(&planted),
Err(LinkError::BadAgentKey { .. })
));
assert_eq!(std::fs::read_to_string(target).unwrap(), "do not overwrite");
}
#[cfg(unix)]
#[test]
fn key_creation_refuses_a_symlinked_parent_without_writing_through_it() {
use std::os::unix::fs::symlink;
let root = tempfile::tempdir().unwrap();
let outside = tempfile::tempdir().unwrap();
symlink(outside.path(), root.path().join("redirect")).unwrap();
assert!(generate_agent_key(&root.path().join("redirect/agent.key")).is_err());
assert!(!outside.path().join("agent.key").exists());
}
#[test]
fn address_bare_brain_with_and_without_sigil() {
for raw in ["@acme-ops", "acme-ops"] {
let a = Address::parse(raw).expect(raw);
assert_eq!(a.brain, "acme-ops");
assert_eq!(a.target, None);
}
}
#[test]
fn address_ulid_target_parses_as_id() {
let a = Address::parse("@acme/01j5qc3v9k4ym8rwbn2tqe6f7d").unwrap();
assert_eq!(a.brain, "acme");
assert_eq!(
a.target,
Some(AddressTarget::Id("01j5qc3v9k4ym8rwbn2tqe6f7d".to_string()))
);
}
#[test]
fn address_md_path_target_parses_as_path() {
let a = Address::parse("@acme/records/clients/lumio.md").unwrap();
assert_eq!(
a.target,
Some(AddressTarget::Path("records/clients/lumio.md".to_string()))
);
}
#[test]
fn address_rejects_malformed_forms() {
for raw in [
"",
"@",
"@/x",
"@acme/",
"@acme/../etc/passwd",
"@acme/records/.hidden.md",
"@ACME", "@acme/notes/x.txt", "@a b", ] {
assert!(Address::parse(raw).is_err(), "should reject {raw:?}");
}
}
#[test]
fn safe_paths_accept_store_shapes_and_reject_escapes() {
for ok in [
"DB.md",
"assets.jsonl",
"records/clients/lumio.md",
"sources/emails/2026/07/x.md",
] {
assert!(safe_store_rel_path(ok), "should accept {ok:?}");
}
for bad in [
"",
"/etc/passwd",
"../up.md",
"records/../../up.md",
"records//x.md",
".dbmd/config",
"records/.hidden/x.md",
"records/a b.md",
"records\\win.md",
] {
assert!(!safe_store_rel_path(bad), "should reject {bad:?}");
}
}
#[cfg(unix)]
#[test]
fn opened_destination_capability_survives_an_ancestor_path_swap() {
use std::os::unix::fs::symlink;
let work = tempfile::tempdir().unwrap();
let outside = tempfile::tempdir().unwrap();
let original = work.path().join("destination");
let moved = work.path().join("destination-moved");
let directory = open_or_create_dir_nofollow(&original).unwrap();
std::fs::rename(&original, &moved).unwrap();
symlink(outside.path(), &original).unwrap();
write_pull_entries_beneath_dir(
&directory,
&[("records/note.md".to_string(), b"held inode".to_vec())],
)
.unwrap();
assert_eq!(
std::fs::read(moved.join("records/note.md")).unwrap(),
b"held inode"
);
assert!(!outside.path().join("records/note.md").exists());
}
#[test]
fn hub_config_flag_beats_file_and_requires_some_source() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir_all(dir.path().join(".dbmd")).unwrap();
std::fs::write(
dir.path().join(CONFIG_REL_PATH),
"# toolkit state\nhub = https://file.example.com\nunknown = ignored\n",
)
.unwrap();
let from_flag = hub_config(Some("https://flag.example.com/"), dir.path()).unwrap();
assert_eq!(from_flag.hub, "https://flag.example.com");
let from_file = hub_config(None, dir.path()).unwrap();
assert_eq!(from_file.hub, "https://file.example.com");
let none = hub_config(None, tempfile::tempdir().unwrap().path());
assert!(matches!(none, Err(LinkError::NoHub)));
}
#[test]
fn https_guard_allows_loopback_only_for_plain_http() {
assert!(assert_safe_hub("https://hub.example.com").is_ok());
assert!(assert_safe_hub("http://localhost:3000").is_ok());
assert!(assert_safe_hub("http://127.0.0.1:3000").is_ok());
assert!(assert_safe_hub("http://[::1]:3000").is_ok());
assert!(matches!(
assert_safe_hub("http://hub.example.com"),
Err(LinkError::UnsafeHub { .. })
));
assert!(matches!(
assert_safe_hub("hub.example.com"),
Err(LinkError::UnsafeHub { .. })
));
assert!(matches!(
assert_safe_hub("http://localhost:80@127.0.0.1:1"),
Err(LinkError::UnsafeHub { .. })
));
assert!(matches!(
assert_safe_hub("https://hub.example.com@attacker.example"),
Err(LinkError::UnsafeHub { .. })
));
assert!(matches!(
assert_safe_hub("https://hub.example.com/base"),
Err(LinkError::UnsafeHub { .. })
));
}
#[test]
fn registry_ssrf_classifier_rejects_local_private_and_documentation_ranges() {
for blocked in [
"127.0.0.1",
"10.0.0.1",
"100.64.0.1",
"169.254.169.254",
"172.16.0.1",
"192.168.0.1",
"192.88.99.1",
"198.18.0.1",
"203.0.113.1",
"::1",
"fe80::1",
"fd00::1",
"2001:db8::1",
"2001:1::1",
"2002:7f00:1::",
"3fff::1",
] {
assert!(
!is_public_registry_ip(blocked.parse().unwrap()),
"must block {blocked}"
);
}
assert!(is_public_registry_ip("1.1.1.1".parse().unwrap()));
assert!(is_public_registry_ip(
"2606:4700:4700::1111".parse().unwrap()
));
assert!(is_public_registry_ip("3fff:1000::1".parse().unwrap()));
}
#[test]
fn registry_dns_answer_is_pinned_and_cannot_rebind_or_change_authority() {
use ureq::Resolver as _;
let pinned: std::net::SocketAddr = "1.1.1.1:443".parse().unwrap();
let resolver = PinnedRegistryResolver {
netloc: "home.example:443".to_string(),
addresses: vec![pinned],
};
assert_eq!(resolver.resolve("home.example:443").unwrap(), vec![pinned]);
assert!(resolver.resolve("127.0.0.1:443").is_err());
assert_eq!(
resolver.resolve("home.example:443").unwrap(),
vec![pinned],
"subsequent connects reuse the validated answer instead of DNS"
);
}
#[test]
fn production_object_urls_and_store_selected_hubs_cannot_reach_private_ips() {
let cfg = HubConfig {
hub: "https://hub.example".to_string(),
key: None,
agent_key: None,
brain_key: None,
state_dir: tempfile::tempdir().unwrap().keep(),
store_selected: false,
};
assert!(
presigned_agent(&cfg, "https://127.0.0.1/private").is_err(),
"a production hub must not turn its presigned URL into an SSRF primitive"
);
let store_selected = HubConfig {
hub: "https://127.0.0.1".to_string(),
store_selected: true,
..cfg
};
assert!(
hub_agent(&store_selected).is_err(),
"bytes in a cloned store must not select a private-network hub"
);
}
#[test]
fn presigned_download_ceiling_cannot_overflow_at_u64_max() {
assert_eq!(
one_past_bounded_limit(MAX_PACK_BYTES),
Some(MAX_PACK_BYTES + 1),
"the presigned reader consumes exactly one refusal byte beyond its fixed pack cap"
);
assert_eq!(
presigned_download_read_limit(),
MAX_PACK_BYTES + 1,
"the presigned reader is capped by the client constant, not a hub response"
);
assert_eq!(
one_past_bounded_limit(u64::MAX),
None,
"the former attacker-controlled u64::MAX + 1 shape must fail without overflow"
);
}
#[test]
fn https_guard_matches_the_scheme_case_insensitively() {
assert!(assert_safe_hub("HTTPS://hub.example.com").is_ok());
assert!(assert_safe_hub("Https://hub.example.com").is_ok());
assert!(matches!(
assert_safe_hub("HTTP://hub.example.com"),
Err(LinkError::UnsafeHub { .. })
));
}
#[test]
fn clean_key_refuses_paste_artifacts_without_echoing() {
assert_eq!(clean_key(" vc_account_abc ").unwrap(), "vc_account_abc");
for bad in ["vc account", "vc\naccount", "ключ", ""] {
let err = clean_key(bad).unwrap_err();
assert!(matches!(err, LinkError::BadKey));
assert!(
!err.to_string().contains(bad.trim()) || bad.trim().is_empty(),
"error must not echo the key"
);
}
}
fn dead_hub() -> HubConfig {
HubConfig {
hub: "http://127.0.0.1:9".to_string(),
key: Some("k".to_string()),
agent_key: None,
brain_key: None,
state_dir: PathBuf::from("."),
store_selected: false,
}
}
#[test]
fn request_retries_a_connection_failure_before_sending() {
use std::io::{Read as _, Write as _};
use std::net::TcpListener;
use std::thread;
use std::time::Duration;
let probe = TcpListener::bind("127.0.0.1:0").unwrap();
let address = probe.local_addr().unwrap();
drop(probe);
let server = thread::spawn(move || {
thread::sleep(Duration::from_millis(40));
let listener = TcpListener::bind(address).unwrap();
let (mut stream, _) = listener.accept().unwrap();
let mut request_bytes = [0_u8; 1024];
let _ = stream.read(&mut request_bytes).unwrap();
stream
.write_all(
b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}",
)
.unwrap();
});
let cfg = HubConfig {
hub: format!("http://{address}"),
key: None,
agent_key: None,
brain_key: None,
state_dir: tempfile::tempdir().unwrap().keep(),
store_selected: false,
};
let response = request(&cfg, "GET", "/retry", None, Auth::None).unwrap();
assert_eq!(response.status, 200);
assert_eq!(response.body, Some(json!({ "ok": true })));
server.join().unwrap();
}
#[test]
fn endpoint_cap_refuses_a_body_before_json_parsing() {
let (hub, server) = scripted_json_hub(vec![(200, "x".repeat(2_048))]);
let cfg = HubConfig {
hub,
key: None,
agent_key: None,
brain_key: None,
state_dir: tempfile::tempdir().unwrap().keep(),
store_selected: false,
};
assert!(matches!(
request_capped(&cfg, "GET", "/bounded", None, Auth::None, 1_024),
Err(LinkError::ResponseTooLarge { .. })
));
server.join().unwrap();
}
#[test]
fn overall_deadline_stops_a_dribbled_response_body() {
use std::io::{Read as _, Write as _};
use std::net::TcpListener;
use std::time::{Duration, Instant};
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}/dribble", listener.local_addr().unwrap());
let server = std::thread::spawn(move || {
let (mut stream, _) = listener.accept().unwrap();
let mut request = [0_u8; 1024];
let _ = stream.read(&mut request);
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 32\r\nConnection: close\r\n\r\n")
.unwrap();
for byte in [b'x'; 32] {
if stream.write_all(&[byte]).is_err() {
break;
}
std::thread::sleep(Duration::from_millis(40));
}
});
let http = agent_builder_with_timeout(Duration::from_millis(150)).build();
let started = Instant::now();
let response = http.get(&url).call().unwrap();
let mut body = Vec::new();
let error = response
.into_reader()
.read_to_end(&mut body)
.expect_err("per-read progress must not reset the overall deadline");
assert!(
started.elapsed() < Duration::from_millis(700),
"dribbled body exceeded the wall-clock budget: {error}"
);
server.join().unwrap();
}
#[test]
fn overall_deadline_stops_a_stalled_upload() {
use std::net::TcpListener;
use std::time::{Duration, Instant};
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let url = format!("http://{}/upload", listener.local_addr().unwrap());
let server = std::thread::spawn(move || {
let (_stream, _) = listener.accept().unwrap();
std::thread::sleep(Duration::from_millis(600));
});
let http = agent_builder_with_timeout(Duration::from_millis(150)).build();
let body = vec![0x5a; 32 * 1024 * 1024];
let started = Instant::now();
let error = http
.put(&url)
.send_bytes(&body)
.expect_err("stalled request-body writes must time out");
assert!(
started.elapsed() < Duration::from_millis(700),
"stalled upload exceeded the wall-clock budget: {error}"
);
server.join().unwrap();
}
#[test]
fn verb_entry_gates_accept_the_hub_ref_shapes() {
for ok in ["acme-ops", "a", "01j5qc3v9k4ym8rwbn2tqe6f7d"] {
assert!(require_safe_ref(ok).is_ok(), "brain ref {ok:?}");
assert!(require_valid_handle(ok).is_ok(), "handle {ok:?}");
assert!(require_safe_grant_id(ok).is_ok(), "grant id {ok:?}");
}
}
#[test]
fn raw_ref_verbs_refuse_url_reshaping_brain_refs_before_any_request() {
let cfg = dead_hub();
for bad in ["../up", "a/b", "a?x=1", "a#frag", "a%2e%2e", "A", "a b", ""] {
assert!(
matches!(
sync_pull(&cfg, bad, None),
Err(LinkError::BadAddress { .. })
),
"sync_pull must refuse {bad:?}"
);
assert!(
matches!(sync_push(&cfg, bad, &[]), Err(LinkError::BadAddress { .. })),
"sync_push must refuse {bad:?}"
);
assert!(
matches!(
grant_issue(&cfg, bad, "maya@example.com", Capability::Read, None, None),
Err(LinkError::BadAddress { .. })
),
"grant_issue must refuse {bad:?}"
);
assert!(
matches!(grant_list(&cfg, bad), Err(LinkError::BadAddress { .. })),
"grant_list must refuse {bad:?}"
);
assert!(
matches!(
grant_revoke(&cfg, bad, "01j5qc3v9k4ym8rwbn2tqe6f7f"),
Err(LinkError::BadAddress { .. })
),
"grant_revoke must refuse brain {bad:?}"
);
assert!(
matches!(head(&cfg, bad), Err(LinkError::BadAddress { .. })),
"head must refuse {bad:?}"
);
}
}
#[test]
fn grant_revoke_refuses_url_reshaping_grant_ids() {
let cfg = dead_hub();
for bad in ["../01j", "a/b", "id?x=1", "id#frag", "ID", ""] {
assert!(
matches!(
grant_revoke(&cfg, "acme", bad),
Err(LinkError::BadGrantId { .. })
),
"grant_revoke must refuse grant id {bad:?}"
);
}
}
#[test]
fn propose_refuses_url_reshaping_handles_and_oversize_bodies_before_upload() {
let cfg = dead_hub();
for bad in ["../up", "a/b", "a?x=1", "a#frag", "A", ""] {
assert!(
matches!(
propose(&cfg, bad, "intake", "hi"),
Err(LinkError::BadAddress { .. })
),
"propose must refuse handle {bad:?}"
);
}
let oversize = "a".repeat(MAX_PROPOSE_BYTES as usize + 1);
assert!(matches!(
propose(&cfg, "acme-site", "intake", &oversize),
Err(LinkError::ProposeTooLarge { .. })
));
assert!(matches!(
propose(&cfg, "acme-site", "intake", "hi"),
Err(LinkError::Transport { .. })
));
}
#[test]
fn resolve_refuses_a_hand_built_unsafe_address() {
let cfg = dead_hub();
for brain in ["../up", "a/b", "a?x", "a#f"] {
let addr = Address {
brain: brain.to_string(),
target: None,
};
assert!(
matches!(resolve(&cfg, &addr), Err(LinkError::BadAddress { .. })),
"resolve must refuse brain {brain:?}"
);
}
for target in [
AddressTarget::Id("01j5qc3v9k4ym8rwbn2tqe6f7d?id=other".to_string()),
AddressTarget::Id("01J5QC3V9K4YM8RWBN2TQE6F7D".to_string()), AddressTarget::Path("../up.md".to_string()),
AddressTarget::Path("records/x.md#frag".to_string()),
] {
let addr = Address {
brain: "acme".to_string(),
target: Some(target.clone()),
};
assert!(
matches!(resolve(&cfg, &addr), Err(LinkError::BadAddress { .. })),
"resolve must refuse target {target:?}"
);
}
}
}