use std::fmt;
use std::fs::{self, File, Metadata, OpenOptions};
use std::io::{self, Read, Write};
use std::num::NonZeroU32;
use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
use std::os::unix::net::SocketAddr;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use super::runtime::{self, Endpoint};
const MOST_BYTES: u64 = 65_536;
static TEMPORARIES: AtomicU64 = AtomicU64::new(0);
#[derive(Debug)]
pub struct Refusal {
pub code: &'static str,
detail: String,
}
impl Refusal {
pub(super) fn new(code: &'static str, detail: impl fmt::Display) -> Self {
Self {
code,
detail: detail.to_string(),
}
}
}
impl fmt::Display for Refusal {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(formatter, "{}: {}", self.code, self.detail)
}
}
impl std::error::Error for Refusal {}
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct Entry {
pub name: String,
pub socket: PathBuf,
pub data_dir: PathBuf,
pub pid: NonZeroU32,
pub version: String,
}
impl Entry {
pub fn current(endpoint: &Endpoint) -> Result<Self, Refusal> {
let data_dir = fs::canonicalize(&endpoint.data_dir)
.map_err(|error| Refusal::new("registry_publication_failed", error))?;
if !fs::metadata(&data_dir)
.map_err(|error| Refusal::new("registry_publication_failed", error))?
.is_dir()
{
return Err(Refusal::new(
"registry_publication_failed",
"data location is not a directory",
));
}
let pid = NonZeroU32::new(std::process::id())
.ok_or_else(|| Refusal::new("registry_publication_failed", "process id is zero"))?;
let entry = Self {
name: endpoint.name.clone(),
socket: endpoint.socket.clone(),
data_dir,
pid,
version: env!("CARGO_PKG_VERSION").to_owned(),
};
entry
.validate()
.map_err(|error| Refusal::new("registry_publication_failed", error))?;
Ok(entry)
}
fn validate(&self) -> Result<(), &'static str> {
if self.name.is_empty()
|| self.name.len() > 128
|| matches!(self.name.as_str(), "." | "..")
|| self.name.contains(['/', '\\', '\0'])
{
return Err("invalid logical store name");
}
if !valid_path(&self.socket) || SocketAddr::from_pathname(&self.socket).is_err() {
return Err("invalid Unix endpoint");
}
if !valid_path(&self.data_dir)
|| self.data_dir.file_name().is_none()
|| self
.data_dir
.components()
.any(|part| !matches!(part, Component::RootDir | Component::Normal(_)))
{
return Err("invalid canonical data directory");
}
if self.version.is_empty()
|| self.version.len() > 128
|| !self
.version
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'-' | b'+'))
{
return Err("invalid haematite version");
}
Ok(())
}
}
fn valid_path(path: &Path) -> bool {
path.is_absolute() && path.to_str().is_some_and(|value| !value.contains('\0'))
}
fn entry_path(directory: &Path, name: &str) -> PathBuf {
const HEX: &[u8; 16] = b"0123456789abcdef";
let digest = Sha256::digest(name.as_bytes());
let mut filename = String::with_capacity(69);
for byte in digest {
filename.push(char::from(HEX[usize::from(byte >> 4)]));
filename.push(char::from(HEX[usize::from(byte & 15)]));
}
filename.push_str(".json");
directory.join(filename)
}
fn private_directory(directory: &Path) -> io::Result<Metadata> {
let metadata = fs::symlink_metadata(directory)?;
if !directory.is_absolute()
|| !metadata.is_dir()
|| metadata.permissions().mode() & 0o7777 != 0o700
{
return Err(io::Error::new(
io::ErrorKind::PermissionDenied,
"registry directory must be a non-symlink absolute mode-0700 directory",
));
}
Ok(metadata)
}
pub fn read(path: &Path) -> Result<Entry, Refusal> {
let refused = |error: &dyn fmt::Display| {
Refusal::new(
"invalid_registry_entry",
format!("{}: {error}", path.display()),
)
};
let parent = path
.parent()
.ok_or_else(|| refused(&"entry has no parent"))?;
let directory = private_directory(parent).map_err(|error| refused(&error))?;
let named = fs::symlink_metadata(path).map_err(|error| refused(&error))?;
if !named.is_file() {
return Err(refused(&"entry is not a regular file"));
}
let mut file = File::open(path).map_err(|error| refused(&error))?;
let metadata = file.metadata().map_err(|error| refused(&error))?;
if !metadata.is_file()
|| metadata.uid() != directory.uid()
|| metadata.permissions().mode() & 0o7777 != 0o600
|| metadata.len() > MOST_BYTES
{
return Err(refused(
&"entry must be an owned mode-0600 regular file within the size bound",
));
}
let mut bytes = Vec::new();
Read::by_ref(&mut file)
.take(MOST_BYTES + 1)
.read_to_end(&mut bytes)
.map_err(|error| refused(&error))?;
if u64::try_from(bytes.len()).map_err(|error| refused(&error))? > MOST_BYTES {
return Err(refused(&"entry exceeds the size bound"));
}
let entry: Entry = serde_json::from_slice(&bytes).map_err(|error| refused(&error))?;
entry.validate().map_err(|error| refused(&error))?;
if path != entry_path(parent, &entry.name) {
return Err(refused(&"entry name does not match its filename"));
}
Ok(entry)
}
pub fn read_directory(directory: &Path) -> Result<Vec<Entry>, Refusal> {
private_directory(directory).map_err(|error| {
Refusal::new(
"registry_unavailable",
format!("{}: {error}", directory.display()),
)
})?;
let entries = fs::read_dir(directory).map_err(|error| {
Refusal::new(
"registry_unavailable",
format!("{}: {error}", directory.display()),
)
})?;
let mut records = Vec::new();
for entry in entries {
let entry = entry.map_err(|error| Refusal::new("registry_unavailable", error))?;
let path = entry.path();
if path
.extension()
.is_some_and(|extension| extension == "json")
{
records.push(read(&path)?);
}
}
records.sort_by(|left, right| left.name.cmp(&right.name));
Ok(records)
}
struct Prepared {
temporary: Option<PathBuf>,
destination: PathBuf,
identity: (u64, u64),
entry: Entry,
}
impl Prepared {
fn new(directory: &Path, entry: Entry) -> Result<Self, Refusal> {
let error = |cause: &dyn fmt::Display| {
Refusal::new(
"registry_publication_failed",
format!("{}: {cause}", directory.display()),
)
};
entry.validate().map_err(|cause| error(&cause))?;
let owner = private_directory(directory)
.map_err(|cause| error(&cause))?
.uid();
let bytes = serde_json::to_vec(&entry).map_err(|cause| error(&cause))?;
if u64::try_from(bytes.len()).map_err(|cause| error(&cause))? > MOST_BYTES {
return Err(error(&"entry exceeds size bound"));
}
let sequence = TEMPORARIES
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
value.checked_add(1)
})
.map_err(|cause| error(&cause))?;
let temporary = directory.join(format!(".entry-{}-{sequence}", std::process::id()));
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&temporary)
.map_err(|cause| error(&cause))?;
let metadata = match file.metadata() {
Ok(metadata) => metadata,
Err(cause) => {
drop(file);
return Err(match fs::remove_file(&temporary) {
Ok(()) => error(&cause),
Err(cleanup) => error(&format!(
"metadata: {cause}; registry_cleanup_failed: {cleanup}"
)),
});
}
};
let mut prepared = Self {
temporary: Some(temporary),
destination: entry_path(directory, &entry.name),
identity: (metadata.dev(), metadata.ino()),
entry,
};
let written = if metadata.uid() == owner
&& metadata.is_file()
&& metadata.permissions().mode() & 0o7777 == 0o600
{
file.write_all(&bytes)
} else {
Err(io::Error::new(
io::ErrorKind::PermissionDenied,
"temporary entry ownership or mode is invalid",
))
};
drop(file);
if let Err(cause) = written {
return Err(match prepared.clean_temporary() {
Ok(()) => error(&cause),
Err(cleanup) => error(&format!("write: {cause}; {cleanup}")),
});
}
Ok(prepared)
}
fn clean_temporary(&mut self) -> Result<(), Refusal> {
if let Some(path) = self.temporary.take() {
remove_same(&path, self.identity)?;
}
Ok(())
}
fn publish(mut self) -> Result<Publication, Refusal> {
let temporary = self.temporary.as_ref().ok_or_else(|| {
Refusal::new("registry_publication_failed", "temporary entry is absent")
})?;
if let Err(cause) = fs::rename(temporary, &self.destination) {
return Err(match self.clean_temporary() {
Ok(()) => Refusal::new("registry_publication_failed", cause),
Err(cleanup) => Refusal::new(
"registry_publication_failed",
format!("rename: {cause}; {cleanup}"),
),
});
}
self.temporary.take();
Ok(Publication {
path: self.destination.clone(),
identity: self.identity,
entry: self.entry.clone(),
active: true,
})
}
}
impl Drop for Prepared {
fn drop(&mut self) {
if let Err(error) = self.clean_temporary() {
eprintln!("{error}");
}
}
}
fn remove_same(path: &Path, identity: (u64, u64)) -> Result<(), Refusal> {
let refused = |cause: &dyn fmt::Display| {
Refusal::new(
"registry_cleanup_failed",
format!("{}: {cause}", path.display()),
)
};
let metadata = fs::symlink_metadata(path).map_err(|cause| refused(&cause))?;
if !metadata.is_file() || (metadata.dev(), metadata.ino()) != identity {
return Err(refused(&"owned entry was replaced"));
}
fs::remove_file(path).map_err(|cause| refused(&cause))
}
pub(super) struct Publication {
path: PathBuf,
identity: (u64, u64),
entry: Entry,
active: bool,
}
impl Publication {
pub(super) fn publish(endpoint: &Endpoint) -> Result<Self, Refusal> {
let namespace = namespace(endpoint)
.map_err(|cause| Refusal::new("registry_publication_failed", cause))?;
let directory = namespace.socket.parent().ok_or_else(|| {
Refusal::new(
"registry_publication_failed",
"runtime endpoint has no parent",
)
})?;
Self::publish_in(directory, Entry::current(endpoint)?)
}
pub(super) fn publish_in(directory: &Path, entry: Entry) -> Result<Self, Refusal> {
Prepared::new(directory, entry)?.publish()
}
pub(super) fn finish(&mut self) -> Result<(), Refusal> {
if !self.active {
return Ok(());
}
self.active = false;
let actual =
read(&self.path).map_err(|cause| Refusal::new("registry_cleanup_failed", cause))?;
if actual != self.entry {
return Err(Refusal::new(
"registry_cleanup_failed",
"owned entry contents changed",
));
}
remove_same(&self.path, self.identity)
}
}
impl Drop for Publication {
fn drop(&mut self) {
if let Err(error) = self.finish() {
eprintln!("{error}");
}
}
}
fn namespace(endpoint: &Endpoint) -> Result<Endpoint, runtime::Refusal> {
#[cfg(test)]
if let Some((root, uid)) = super::cli::endpoint_tests::input() {
return runtime::resolve_input(
&endpoint.data_dir,
Some(&endpoint.name),
None,
false,
&root,
uid,
);
}
runtime::resolve(&endpoint.data_dir, Some(&endpoint.name), None, false)
}
#[cfg(test)]
mod tests {
use std::error::Error;
use std::fs;
use std::io::{self, BufRead, Read, Write};
use std::os::unix::fs::PermissionsExt;
use std::path::Path;
use std::process::{Child, ChildStdout, Command, Stdio};
use super::{Entry, Prepared, Publication, entry_path, read, read_directory};
use crate::serve::cli::endpoint_tests::{Fixture, with_fixture};
use crate::serve::runtime::Endpoint;
use crate::serve::transport::Listener;
type TestResult = Result<(), Box<dyn Error>>;
fn endpoint(root: &Path, name: &str) -> Endpoint {
Endpoint {
name: name.to_owned(),
socket: root.join(format!("{name}.sock")),
data_dir: root.join("database"),
}
}
fn directory(fixture: &Fixture) -> Result<std::path::PathBuf, io::Error> {
let directory = fixture.directory.path().join("haem");
fs::create_dir(&directory)?;
fs::set_permissions(&directory, fs::Permissions::from_mode(0o700))?;
Ok(directory)
}
#[test]
fn publication_precedes_readiness_and_contains_the_five_fields() -> TestResult {
with_fixture(|fixture| {
let directory = directory(fixture)?;
let endpoint = endpoint(fixture.directory.path(), "ready");
let mut listener = Listener::bind(&endpoint.socket)?;
listener.publish_in(&directory, Entry::current(&endpoint)?)?;
let entry = read(&entry_path(&directory, &endpoint.name))?;
assert_eq!(entry.name, endpoint.name);
assert_eq!(entry.socket, endpoint.socket);
assert_eq!(entry.data_dir, fs::canonicalize(&fixture.data_dir)?);
assert_eq!(entry.pid.get(), std::process::id());
assert_eq!(entry.version, env!("CARGO_PKG_VERSION"));
let path = entry_path(&directory, &endpoint.name);
let value: serde_json::Value = serde_json::from_slice(&fs::read(&path)?)?;
assert_eq!(value.as_object().ok_or("entry is not an object")?.len(), 5);
assert_eq!(fs::metadata(&path)?.permissions().mode() & 0o7777, 0o600);
listener.finish()?;
assert!(!path.try_exists()?);
assert!(Listener::bind(&endpoint.socket).is_err());
drop(listener);
drop(Listener::bind(&endpoint.socket)?);
Ok(())
})
}
#[test]
fn a_reader_sees_the_complete_prior_entry_until_atomic_rename() -> TestResult {
with_fixture(|fixture| {
let directory = directory(fixture)?;
let endpoint = endpoint(fixture.directory.path(), "atomic");
let prior = Entry::current(&endpoint)?;
let path = entry_path(&directory, &prior.name);
fs::write(&path, serde_json::to_vec(&prior)?)?;
fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?;
let mut replacement = prior.clone();
replacement.version = "0.1.1".to_owned();
let pending = Prepared::new(&directory, replacement.clone())?;
assert_eq!(read(&path)?, prior);
let mut publication = pending.publish()?;
assert_eq!(read(&path)?, replacement);
assert_eq!(read_directory(&directory)?, vec![replacement]);
publication.finish()?;
Ok(())
})
}
#[test]
fn clean_exit_removes_only_our_entry_and_kill_leaves_it_complete() -> TestResult {
with_fixture(|fixture| {
let directory = directory(fixture)?;
let other = Entry::current(&endpoint(fixture.directory.path(), "other"))?;
let mut other_owner = Publication::publish_in(&directory, other.clone())?;
let child_path = entry_path(&directory, "child");
let child_socket = fixture.directory.path().join("child.sock");
let mut clean = Process::start(fixture.directory.path())?;
let clean_entry = read(&child_path)?;
assert_eq!(clean_entry.pid.get(), clean.pid()?);
assert!(Listener::bind(&child_socket).is_err());
clean.finish()?;
assert!(!child_path.try_exists()?);
assert_eq!(read(&entry_path(&directory, "other"))?, other);
drop(Listener::bind(&child_socket)?);
let mut killed = Process::start(fixture.directory.path())?;
let last = read(&child_path)?;
killed.kill()?;
assert_eq!(read(&child_path)?, last);
drop(Listener::bind(&child_socket)?);
assert_eq!(read(&entry_path(&directory, "other"))?, other);
other_owner.finish()?;
Ok(())
})
}
#[test]
fn publication_and_cleanup_failures_are_named_and_preserve_replacements() -> TestResult {
with_fixture(|fixture| {
let directory = directory(fixture)?;
let endpoint = endpoint(fixture.directory.path(), "failure");
let entry = Entry::current(&endpoint)?;
let path = entry_path(&directory, &entry.name);
fs::create_dir(&path)?;
let mut listener = Listener::bind(&endpoint.socket)?;
let mut ready = false;
let result = listener.publish_in(&directory, entry.clone()).map(|()| {
ready = true;
});
let refused = result.err().ok_or("publication unexpectedly succeeded")?;
assert_eq!(refused.code, "registry_publication_failed");
assert!(!ready);
assert!(path.is_dir());
assert_eq!(fs::read_dir(&directory)?.count(), 1);
fs::remove_dir(&path)?;
listener.publish_in(&directory, entry)?;
let saved = directory.join("saved");
fs::rename(&path, &saved)?;
fs::create_dir(&path)?;
let refused = listener.finish().err().ok_or("replacement was removed")?;
assert_eq!(refused.code, "registry_cleanup_failed");
assert!(path.is_dir());
assert!(saved.try_exists()?);
fs::remove_dir(&path)?;
fs::rename(&saved, &path)?;
fs::remove_file(&path)?;
Ok(())
})
}
#[test]
fn malformed_entries_are_refused_at_the_reader_boundary() -> TestResult {
with_fixture(|fixture| {
let directory = directory(fixture)?;
let entry = Entry::current(&endpoint(fixture.directory.path(), "validate"))?;
let path = entry_path(&directory, &entry.name);
let valid = serde_json::to_value(entry)?;
for (field, replacement) in [
("name", serde_json::json!("../escape")),
("name", serde_json::json!("another")),
("socket", serde_json::json!("relative.sock")),
("data_dir", serde_json::json!("relative")),
("pid", serde_json::json!(0)),
("pid", serde_json::json!("1")),
("version", serde_json::json!("")),
("extra", serde_json::json!(true)),
] {
let mut value = valid.clone();
value[field] = replacement;
fs::write(&path, serde_json::to_vec(&value)?)?;
fs::set_permissions(&path, fs::Permissions::from_mode(0o600))?;
let refused = read(&path).err().ok_or("invalid entry was accepted")?;
assert_eq!(refused.code, "invalid_registry_entry");
assert!(refused.to_string().contains(&path.display().to_string()));
}
fs::write(&path, b"{")?;
assert_eq!(
read(&path).err().ok_or("partial entry was accepted")?.code,
"invalid_registry_entry"
);
Ok(())
})
}
#[test]
fn missing_namespace_is_named_and_the_next_publication_recreates_it() -> TestResult {
with_fixture(|fixture| {
let directory = fixture.directory.path().join("haem");
assert_eq!(
read_directory(&directory)
.err()
.ok_or("missing namespace was accepted")?
.code,
"registry_unavailable"
);
let endpoint = endpoint(fixture.directory.path(), "recreate");
let mut listener = Listener::bind(&endpoint.socket)?;
listener.publish(&endpoint)?;
assert_eq!(read_directory(&directory)?.len(), 1);
listener.finish()?;
Ok(())
})
}
#[test]
fn explicit_socket_cannot_skip_a_refused_namespace_lookup() -> TestResult {
with_fixture(|fixture| {
let endpoint = endpoint(fixture.directory.path(), "explicit");
let mut listener = Listener::bind(&endpoint.socket)?;
fs::set_permissions(fixture.directory.path(), fs::Permissions::from_mode(0o500))?;
let result = listener.publish(&endpoint);
fs::set_permissions(fixture.directory.path(), fs::Permissions::from_mode(0o700))?;
let refused = result.err().ok_or("invalid namespace was accepted")?;
assert_eq!(refused.code, "registry_publication_failed");
assert!(refused.to_string().contains("invalid_runtime_location"));
assert!(!fixture.directory.path().join("haem").try_exists()?);
Ok(())
})
}
#[test]
fn registry_child() -> TestResult {
let Some(root) = std::env::var_os("HAEM_REGISTRY_TEST_CHILD") else {
return Ok(());
};
let root = Path::new(&root);
let endpoint = endpoint(root, "child");
let mut listener = Listener::bind(&endpoint.socket)?;
listener.publish_in(&root.join("haem"), Entry::current(&endpoint)?)?;
println!("haematite-registry-ready");
io::stdout().flush()?;
let mut command = String::new();
io::stdin().read_line(&mut command)?;
if command != "finish\n" {
return Err("child did not receive its finish signal".into());
}
listener.finish()?;
Ok(())
}
struct Process {
child: Option<Child>,
output: io::BufReader<ChildStdout>,
}
impl Process {
fn start(root: &Path) -> Result<Self, Box<dyn Error>> {
let mut child = Command::new(std::env::current_exe()?)
.args([
"--exact",
"serve::registry::tests::registry_child",
"--nocapture",
])
.env("HAEM_REGISTRY_TEST_CHILD", root)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()?;
let Some(output) = child.stdout.take() else {
let kill = child.kill();
let wait = child.wait();
return Err(format!(
"child stdout is absent; cleanup kill: {kill:?}; cleanup wait: {wait:?}"
)
.into());
};
let mut process = Self {
child: Some(child),
output: io::BufReader::new(output),
};
loop {
let mut line = String::new();
match process.output.read_line(&mut line) {
Ok(0) => {
return Err(match process.wait(true) {
Ok(()) => "child exited before readiness".into(),
Err(error) => format!("child exited before readiness: {error}").into(),
});
}
Ok(_) if line.contains("haematite-registry-ready") => return Ok(process),
Ok(_) => {}
Err(error) => return Err(format!("read child readiness: {error}").into()),
}
}
}
fn pid(&self) -> Result<u32, Box<dyn Error>> {
Ok(self.child.as_ref().ok_or("child already exited")?.id())
}
fn finish(&mut self) -> TestResult {
let child = self.child.as_mut().ok_or("child already exited")?;
child
.stdin
.take()
.ok_or("child stdin is absent")?
.write_all(b"finish\n")?;
self.wait(true)
}
fn kill(&mut self) -> TestResult {
self.child.as_mut().ok_or("child already exited")?.kill()?;
self.wait(false)
}
fn wait(&mut self, clean: bool) -> TestResult {
let child = self.child.as_mut().ok_or("child already exited")?;
let status = child.wait()?;
let mut errors = String::new();
child
.stderr
.take()
.ok_or("child stderr is absent")?
.read_to_string(&mut errors)?;
self.child.take();
if clean && !status.success() {
return Err(format!("child exited {status}: {errors}").into());
}
if !clean && status.success() {
return Err("killed child exited successfully".into());
}
if !errors.is_empty() {
return Err(format!("child stderr: {errors}").into());
}
Ok(())
}
}
impl Drop for Process {
fn drop(&mut self) {
if let Some(child) = self.child.as_mut() {
match child.try_wait() {
Ok(Some(_)) => {}
Ok(None) => {
if let Err(error) = child.kill() {
eprintln!("registry child cleanup kill: {error}");
}
if let Err(error) = child.wait() {
eprintln!("registry child cleanup wait: {error}");
}
}
Err(error) => {
eprintln!("registry child cleanup status: {error}");
if let Err(error) = child.kill() {
eprintln!("registry child cleanup kill: {error}");
}
if let Err(error) = child.wait() {
eprintln!("registry child cleanup wait: {error}");
}
}
}
if let Some(mut stderr) = child.stderr.take() {
let mut errors = String::new();
match stderr.read_to_string(&mut errors) {
Ok(_) if !errors.is_empty() => {
eprintln!("registry child cleanup stderr: {errors}");
}
Ok(_) => {}
Err(error) => eprintln!("registry child cleanup stderr read: {error}"),
}
}
}
}
}
}