#![cfg(unix)]
#![forbid(unsafe_code)]
use haematite_cli::serve::transport::Listener;
use std::error::Error;
use std::os::unix::{
fs::PermissionsExt,
net::{UnixListener, UnixStream},
};
type TestResult = Result<(), Box<dyn Error>>;
#[test]
fn ownership_is_exclusive_and_reusable_after_release() -> TestResult {
let dir = tempfile::tempdir()?;
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700))?;
let path = dir.path().join("haematite.sock");
let first = Listener::bind(&path)?;
assert!(Listener::bind(&path).is_err());
let _client = UnixStream::connect(&path)?;
drop(first);
assert!(!path.exists());
assert!(dir.path().join("haematite.sock.lock").exists());
let second = Listener::bind(&path)?;
let _client = UnixStream::connect(&path)?;
drop(second);
Ok(())
}
#[test]
fn stale_socket_is_recovered_but_live_foreign_listener_is_preserved() -> TestResult {
let dir = tempfile::tempdir()?;
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700))?;
let path = dir.path().join("haematite.sock");
let foreign = UnixListener::bind(&path)?;
assert!(Listener::bind(&path).is_err());
assert!(UnixStream::connect(&path).is_ok());
drop(foreign);
let recovered = Listener::bind(&path)?;
assert!(UnixStream::connect(&path).is_ok());
drop(recovered);
Ok(())
}
#[test]
fn non_socket_entries_and_non_private_directory_are_refused() -> TestResult {
let dir = tempfile::tempdir()?;
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700))?;
let path = dir.path().join("haematite.sock");
std::fs::write(&path, b"preserve")?;
assert!(Listener::bind(&path).is_err());
assert_eq!(std::fs::read(&path)?, b"preserve");
std::fs::remove_file(&path)?;
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o750))?;
assert!(Listener::bind(&path).is_err());
assert!(!path.exists());
Ok(())
}
fn name_fixture() -> Result<tempfile::TempDir, std::io::Error> {
let dir = tempfile::Builder::new().prefix("n").tempdir_in("/tmp")?;
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700))?;
Ok(dir)
}
fn name_anchor(endpoint: &std::path::Path) -> std::path::PathBuf {
let mut anchor = endpoint.as_os_str().to_os_string();
anchor.push(".owner.lock");
anchor.into()
}
#[test]
fn serve_name_claim_refuses_distinct_sockets_before_storage_and_survives_socket_cleanup()
-> TestResult {
use haematite_cli::serve::cli::name_is_live;
use std::os::unix::fs::MetadataExt;
let dir = name_fixture()?;
let name = format!(
"ownership-{}",
dir.path()
.file_name()
.ok_or("missing fixture name")?
.to_string_lossy()
);
let data = dir.path().join("first");
let endpoint = haematite_cli::serve::runtime::resolve(&data, Some(&name), None, true)?;
let uid = std::fs::metadata(endpoint.socket.parent().ok_or("missing namespace")?)?.uid();
let mut artifacts = RuntimeArtifacts {
paths: vec![name_anchor(&endpoint.socket)],
};
let first_config = serve_config(
dir.path(),
"first",
&name,
Some(&dir.path().join("first.sock")),
)?;
let mut first = NameServe::start(&first_config)?;
let entry_path = registry_path(endpoint.socket.parent().ok_or("missing namespace")?, &name)?;
artifacts.paths.push(entry_path.clone());
let mut entry = haematite_cli::serve::registry::read(&entry_path)?;
assert_eq!(entry.pid.get(), first.child.id());
entry.pid = std::num::NonZeroU32::new(u32::MAX).ok_or("missing bogus pid")?;
std::fs::write(&entry_path, serde_json::to_vec(&entry)?)?;
assert_eq!(haematite_cli::serve::registry::read(&entry_path)?, entry);
assert!(name_is_live(&endpoint.socket, uid)?);
let anchor_inode = std::fs::symlink_metadata(name_anchor(&endpoint.socket))?.ino();
let second_config = serve_config(
dir.path(),
"second",
&name,
Some(&dir.path().join("second.sock")),
)?;
let refused = std::process::Command::new(env!("CARGO_BIN_EXE_haematite"))
.args(["serve", "--config"])
.arg(&second_config)
.arg("--create")
.output()?;
assert!(!refused.status.success());
assert!(String::from_utf8(refused.stderr)?.contains("bind service socket:"));
assert!(!dir.path().join("second").try_exists()?);
assert!(!dir.path().join("second.sock").try_exists()?);
assert_eq!(haematite_cli::serve::registry::read(&entry_path)?, entry);
assert!(name_is_live(&endpoint.socket, uid)?);
assert_hello(&entry.socket)?;
first.kill()?;
assert!(!name_is_live(&endpoint.socket, uid)?);
assert_eq!(haematite_cli::serve::registry::read(&entry_path)?, entry);
assert_eq!(
std::fs::symlink_metadata(name_anchor(&endpoint.socket))?.ino(),
anchor_inode
);
let default_config = serve_config(dir.path(), "second", &name, None)?;
artifacts.paths.push(endpoint.socket.clone());
let mut socket_anchor = endpoint.socket.as_os_str().to_os_string();
socket_anchor.push(".lock");
artifacts.paths.push(socket_anchor.into());
let mut second = NameServe::start(&default_config)?;
assert!(name_is_live(&endpoint.socket, uid)?);
assert_eq!(
haematite_cli::serve::registry::read(&entry_path)?.socket,
endpoint.socket
);
assert_hello(&endpoint.socket)?;
second.kill()?;
assert!(!name_is_live(&endpoint.socket, uid)?);
assert_eq!(
std::fs::symlink_metadata(name_anchor(&endpoint.socket))?.ino(),
anchor_inode
);
artifacts.finish()?;
dir.close()?;
Ok(())
}
#[test]
fn name_probe_changes_no_entry_or_socket_and_retains_no_claim() -> TestResult {
use haematite_cli::serve::cli::name_is_live;
use std::os::unix::fs::{MetadataExt, OpenOptionsExt};
let dir = name_fixture()?;
let endpoint = dir.path().join("name");
let uid = std::fs::metadata(dir.path())?.uid();
assert!(!name_is_live(&endpoint, uid)?);
assert_eq!(std::fs::read_dir(dir.path())?.count(), 0);
let claim = std::fs::OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.mode(0o600)
.open(name_anchor(&endpoint))?;
claim.try_lock()?;
let socket = UnixListener::bind(&endpoint)?;
let entry = dir.path().join("entry.json");
let bytes = br#"{"name":"store","socket":"/unused","data_dir":"/unused","pid":4294967295,"version":"1"}"#;
std::fs::write(&entry, bytes)?;
let before_anchor = std::fs::symlink_metadata(name_anchor(&endpoint))?;
let before_socket = std::fs::symlink_metadata(&endpoint)?;
assert!(name_is_live(&endpoint, uid)?);
assert_eq!(std::fs::read(&entry)?, bytes);
assert_eq!(
std::fs::symlink_metadata(&endpoint)?.ino(),
before_socket.ino()
);
assert_eq!(
std::fs::symlink_metadata(name_anchor(&endpoint))?.ino(),
before_anchor.ino()
);
assert_eq!(std::fs::read(name_anchor(&endpoint))?, b"");
socket.set_nonblocking(true)?;
assert_eq!(
socket
.accept()
.err()
.ok_or("probe connected to the socket")?
.kind(),
std::io::ErrorKind::WouldBlock
);
claim.unlock()?;
assert!(!name_is_live(&endpoint, uid)?);
assert_eq!(std::fs::read(&entry)?, bytes);
assert_eq!(
std::fs::symlink_metadata(&endpoint)?.ino(),
before_socket.ino()
);
claim.try_lock()?;
claim.unlock()?;
drop(claim);
drop(socket);
dir.close()?;
Ok(())
}
#[test]
fn name_anchor_shape_owner_and_parent_failures_are_named() -> TestResult {
use haematite_cli::serve::cli::name_is_live;
use std::os::unix::fs::{MetadataExt, OpenOptionsExt};
let dir = name_fixture()?;
let endpoint = dir.path().join("name");
let anchor = name_anchor(&endpoint);
let uid = std::fs::metadata(dir.path())?.uid();
let foreign_uid = uid.checked_add(1).ok_or("uid overflow")?;
assert!(
name_is_live(&endpoint, foreign_uid)
.err()
.ok_or("foreign parent accepted")?
.contains("name_owner_probe_failed")
);
assert!(!anchor.try_exists()?);
drop(
std::fs::OpenOptions::new()
.write(true)
.create_new(true)
.mode(0o600)
.open(&anchor)?,
);
for mode in [0o644, 0o660, 0o4600] {
std::fs::set_permissions(&anchor, std::fs::Permissions::from_mode(mode))?;
assert!(
name_is_live(&endpoint, uid)
.err()
.ok_or("anchor mode accepted")?
.contains("name_owner_probe_failed")
);
}
std::fs::set_permissions(&anchor, std::fs::Permissions::from_mode(0o600))?;
std::fs::write(&anchor, b"unexpected content")?;
assert!(
name_is_live(&endpoint, uid)
.err()
.ok_or("anchor content accepted")?
.contains("name_owner_probe_failed")
);
assert_eq!(std::fs::read(&anchor)?, b"unexpected content");
std::fs::remove_file(&anchor)?;
std::os::unix::fs::symlink(&endpoint, &anchor)?;
assert!(
name_is_live(&endpoint, uid)
.err()
.ok_or("anchor symlink accepted")?
.contains("name_owner_probe_failed")
);
std::fs::remove_file(&anchor)?;
std::fs::create_dir(&anchor)?;
assert!(
name_is_live(&endpoint, uid)
.err()
.ok_or("anchor directory accepted")?
.contains("name_owner_probe_failed")
);
std::fs::remove_dir(&anchor)?;
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o750))?;
assert!(
name_is_live(&endpoint, uid)
.err()
.ok_or("nonprivate parent accepted")?
.contains("name_owner_probe_failed")
);
std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o700))?;
assert!(!anchor.try_exists()?);
dir.close()?;
Ok(())
}
fn serve_config(
dir: &std::path::Path,
leaf: &str,
name: &str,
socket: Option<&std::path::Path>,
) -> Result<std::path::PathBuf, Box<dyn Error>> {
let path = dir.join(format!("{leaf}.json"));
std::fs::write(
&path,
serde_json::to_vec(&serde_json::json!({
"data_dir": dir.join(leaf), "name": name, "socket": socket,
"branch_cache_bytes_per_shard": 1_048_576, "max_frame": 65_536,
"max_clients": 2, "max_page": 8, "timeout_ms": 1000,
"create": {"shards": 1, "cache_bytes_total": 1_048_576, "executor_threads": 1}
}))?,
)?;
Ok(path)
}
fn registry_path(
root: &std::path::Path,
name: &str,
) -> Result<std::path::PathBuf, std::fmt::Error> {
use sha2::{Digest, Sha256};
use std::fmt::Write;
let mut filename = String::with_capacity(69);
for byte in Sha256::digest(name.as_bytes()) {
write!(filename, "{byte:02x}")?;
}
filename.push_str(".json");
Ok(root.join(filename))
}
fn assert_hello(socket: &std::path::Path) -> TestResult {
let mut stream = UnixStream::connect(socket)?;
let bound = std::num::NonZeroU32::new(65_536).ok_or("missing frame bound")?;
let request = serde_json::to_vec(
&serde_json::json!({"id":"owner", "operation":{"method":"hello", "params":{
"protocol": haematite_cli::serve::protocol::PROTOCOL,
"capabilities": [haematite_cli::serve::protocol::RECEIPT_CAPABILITY],
"max_frame": bound,
}}}),
)?;
haematite_cli::serve::frame::write(&mut stream, &request, bound)?;
let response =
haematite_cli::serve::frame::read(&mut stream, bound)?.ok_or("missing hello response")?;
let response: serde_json::Value = serde_json::from_slice(&response)?;
assert_eq!(response["id"], "owner");
assert_eq!(response["result"]["protocol"], "haem/1", "{response}");
Ok(())
}
struct NameServe {
child: std::process::Child,
stderr: Option<std::process::ChildStderr>,
reaped: bool,
}
impl NameServe {
fn start(config: &std::path::Path) -> Result<Self, Box<dyn Error>> {
use std::io::BufRead;
let child = std::process::Command::new(env!("CARGO_BIN_EXE_haematite"))
.args(["serve", "--config"])
.arg(config)
.arg("--create")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::piped())
.spawn()?;
let mut process = Self {
child,
stderr: None,
reaped: false,
};
let pipe = process
.child
.stderr
.take()
.ok_or("missing service stderr")?;
let mut pipe = std::io::BufReader::new(pipe);
let mut transcript = String::new();
loop {
let mut line = String::new();
if pipe.read_line(&mut line)? == 0 {
return Err(format!("service exited before readiness: {transcript}").into());
}
transcript.push_str(&line);
if line.starts_with("haem/1 ready on ") {
break;
}
}
process.stderr = Some(pipe.into_inner());
Ok(process)
}
fn kill(&mut self) -> TestResult {
self.child.kill()?;
let status = self.child.wait()?;
self.reaped = true;
assert!(!status.success());
Ok(())
}
}
impl Drop for NameServe {
fn drop(&mut self) {
if !self.reaped {
if let Err(error) = self.child.kill() {
eprintln!("name fixture kill: {error}");
}
if let Err(error) = self.child.wait() {
eprintln!("name fixture reap: {error}");
}
}
}
}
struct RuntimeArtifacts {
paths: Vec<std::path::PathBuf>,
}
impl RuntimeArtifacts {
fn finish(&mut self) -> TestResult {
for path in &self.paths {
match std::fs::remove_file(path) {
Ok(()) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => {
return Err(format!(
"owned runtime fixture cleanup {}: {error}",
path.display()
)
.into());
}
}
}
self.paths.clear();
Ok(())
}
}
impl Drop for RuntimeArtifacts {
fn drop(&mut self) {
if let Err(error) = self.finish() {
eprintln!("{error}");
}
}
}