635 lines
25 KiB
Rust
635 lines
25 KiB
Rust
//! Minimal privileged helper for hive-c0re.
|
|
//!
|
|
//! Runs as root. Exposes a narrow unix socket at `/run/hive/priv.sock`
|
|
//! that accepts `PrivRequest` JSON lines and executes only the
|
|
//! operations that genuinely require root. All coordination logic,
|
|
//! broker, HTTP, and scheduling stay in the unprivileged hive-c0re
|
|
//! process.
|
|
//!
|
|
//! **Security model**: every request is validated against a strict
|
|
//! container-name allowlist before any filesystem or process operation.
|
|
//! Only containers whose names match the hive convention (`h-*`,
|
|
//! the manager container included) are accepted. Every variant maps to a single known operation — no
|
|
//! arbitrary command pass-through.
|
|
//!
|
|
//! **Socket activation**: when systemd passes the listener socket via
|
|
//! `LISTEN_FDS=1` + `LISTEN_PID=<self>`, the inherited fd 3 is used
|
|
//! instead of binding a fresh socket.
|
|
//!
|
|
//! **Layout**: this file holds the socket loop and the one-arm-per-op
|
|
//! dispatch table. Each op family's handlers and validation live in their
|
|
//! own module: [`container`] (lifecycle), [`container_config`] (nspawn flags,
|
|
//! systemd drop-ins), [`rootfs`] (the salvage of a container's root before it
|
|
//! turns ephemeral, with [`salvage_io`] for its space, reflink and lock calls),
|
|
//! [`prestart`] (the check of a container's root that runs before each start,
|
|
//! as a sub-command), [`storage`] (btrfs subvolumes, quotas, snapshots, disk
|
|
//! measurement), [`files`] (pause marker, socket dir, legacy tmpfiles),
|
|
//! [`infra`] (gateway nginx, infra containers) and [`closure`] (the swarm-signature
|
|
//! check before an apply, and the cache-facts refresh). Helpers more than one family uses are in [`shared`].
|
|
|
|
mod closure;
|
|
mod container;
|
|
mod container_config;
|
|
mod files;
|
|
mod infra;
|
|
mod prestart;
|
|
mod rootfs;
|
|
mod salvage_io;
|
|
mod shared;
|
|
mod storage;
|
|
mod userns;
|
|
|
|
use anyhow::{Context as _, Result, bail};
|
|
use hive_priv_sock::{PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse};
|
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
|
use tokio::net::unix::OwnedWriteHalf;
|
|
use tokio::net::{UnixListener, UnixStream};
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
// `hive-priv check-rootfs <container>` is a container unit's `ExecStartPre=`,
|
|
// not the daemon: it answers with an exit code and never touches the socket.
|
|
let args: Vec<String> = std::env::args().skip(1).collect();
|
|
if let Some(code) = prestart::run(&args) {
|
|
std::process::exit(code);
|
|
}
|
|
|
|
tracing_subscriber::fmt()
|
|
.with_env_filter(
|
|
tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
|
|
)
|
|
// This is a systemd-managed daemon — stdout always goes to journald,
|
|
// never a human terminal, and journald doesn't strip ANSI escapes:
|
|
// they land in victorialogs as raw byte-array spam otherwise.
|
|
.with_ansi(false)
|
|
.init();
|
|
|
|
let listener = socket_listener()?;
|
|
if let Err(e) = files::remove_legacy_tmpfiles() {
|
|
tracing::warn!(error = %format!("{e:#}"), "legacy tmpfiles cleanup failed");
|
|
}
|
|
tracing::info!("hive-priv listening");
|
|
|
|
loop {
|
|
match listener.accept().await {
|
|
Ok((stream, _)) => {
|
|
tokio::spawn(handle(stream));
|
|
}
|
|
Err(e) => {
|
|
tracing::error!(error = %e, "accept failed");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Refuse to run unless systemd handed this process its socket: `LISTEN_FDS`
|
|
/// is at least 1 and `LISTEN_PID` is `own_pid`. A `LISTEN_PID` naming another
|
|
/// process means the variables were inherited from a parent, not meant for us.
|
|
fn check_socket_activation(
|
|
listen_fds: Option<&str>,
|
|
listen_pid: Option<&str>,
|
|
own_pid: u32,
|
|
) -> Result<()> {
|
|
let listen_fds: Option<i32> = listen_fds.and_then(|s| s.parse().ok());
|
|
let listen_pid: Option<u32> = listen_pid.and_then(|s| s.parse().ok());
|
|
|
|
let activated = matches!(listen_fds, Some(n) if n >= 1) && listen_pid == Some(own_pid);
|
|
if !activated {
|
|
bail!(
|
|
"hive-priv requires systemd socket activation (expected LISTEN_FDS>=1 + \
|
|
LISTEN_PID=<self> for {PRIV_SOCK}); run it via the hive-priv.socket unit, \
|
|
not directly"
|
|
);
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
fn socket_listener() -> Result<UnixListener> {
|
|
// hive-priv is ALWAYS socket-activated by the `hive-priv.socket` unit
|
|
// (fd 3 via LISTEN_FDS). There is intentionally no self-bind fallback,
|
|
// so dev and prod take the same path; see docs/4-implementation/trust-boundary/boundary-mechanism.md.
|
|
check_socket_activation(
|
|
std::env::var("LISTEN_FDS").ok().as_deref(),
|
|
std::env::var("LISTEN_PID").ok().as_deref(),
|
|
std::process::id(),
|
|
)?;
|
|
|
|
// SAFETY: systemd has passed us a ready UnixListener on fd 3.
|
|
let std_listener = unsafe {
|
|
use std::os::unix::io::FromRawFd;
|
|
std::os::unix::net::UnixListener::from_raw_fd(3)
|
|
};
|
|
std_listener
|
|
.set_nonblocking(true)
|
|
.context("set socket non-blocking")?;
|
|
let listener =
|
|
tokio::net::UnixListener::from_std(std_listener).context("wrap systemd socket")?;
|
|
tracing::info!("using systemd-activated socket");
|
|
Ok(listener)
|
|
}
|
|
|
|
async fn handle(stream: UnixStream) {
|
|
let (reader, mut writer) = stream.into_split();
|
|
let mut lines = BufReader::new(reader).lines();
|
|
while let Ok(Some(line)) = lines.next_line().await {
|
|
let resp = dispatch(&line, &mut writer).await;
|
|
// Write the terminal PrivResponse as a PrivEvent::Done. Wire-identical
|
|
// to a bare PrivResponse (untagged), so old hive-c0re callers that
|
|
// deserialise directly to PrivResponse continue to work.
|
|
let event = PrivEvent::Done(resp);
|
|
let mut json = serde_json::to_string(&event).unwrap_or_else(|e| {
|
|
format!("{{\"ok\":false,\"stdout\":\"\",\"stderr\":\"\",\"error\":\"serialise failed: {e}\"}}")
|
|
});
|
|
json.push('\n');
|
|
if let Err(e) = writer.write_all(json.as_bytes()).await {
|
|
tracing::warn!(error = %e, "write response failed");
|
|
break;
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn dispatch(line: &str, writer: &mut OwnedWriteHalf) -> PrivResponse {
|
|
response_of(run(line, writer).await)
|
|
}
|
|
|
|
/// The wire answer for one request's outcome: the handler's output on
|
|
/// success, otherwise the whole error chain and no output.
|
|
fn response_of(outcome: Result<(String, String)>) -> PrivResponse {
|
|
match outcome {
|
|
Ok((stdout, stderr)) => PrivResponse {
|
|
ok: true,
|
|
stdout,
|
|
stderr,
|
|
error: None,
|
|
},
|
|
Err(e) => PrivResponse {
|
|
ok: false,
|
|
stdout: String::new(),
|
|
stderr: String::new(),
|
|
error: Some(format!("{e:#}")),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Parse one request line and execute it.
|
|
///
|
|
/// Split out of [`dispatch`] so the two failure modes collapse into one
|
|
/// `Result` instead of nested matches building the same struct.
|
|
async fn run(line: &str, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
|
|
let req = serde_json::from_str::<PrivRequest>(line).context("parse request")?;
|
|
exec(req, writer).await
|
|
}
|
|
|
|
/// Execute a validated `PrivRequest`. Returns `(stdout, stderr)` on success.
|
|
/// For streaming ops (`CreateContainer`/`UpdateContainer` with `stream: true`)
|
|
/// output lines are forwarded to `writer` as `PrivEvent::Line` messages and
|
|
/// the returned strings are empty.
|
|
// One match arm per priv op — a flat 1:1 dispatch table. Every arm either
|
|
// delegates directly or validates then delegates; an op whose handling is
|
|
// more than that gets its own named function instead, so the match's length
|
|
// tracks the op count, not complexity.
|
|
#[allow(clippy::too_many_lines)]
|
|
async fn exec(req: PrivRequest, writer: &mut OwnedWriteHalf) -> Result<(String, String)> {
|
|
match req {
|
|
PrivRequest::StartContainer { ref name } => container::start_container(name).await,
|
|
|
|
PrivRequest::StartFailureLog {
|
|
ref name,
|
|
since_unix,
|
|
} => container::start_failure_log(name, since_unix).await,
|
|
|
|
PrivRequest::StopContainer { ref name } => container::stop_container(name).await,
|
|
|
|
PrivRequest::KillContainer { ref name } => container::kill_container(name).await,
|
|
|
|
PrivRequest::UpdateContainer {
|
|
ref name,
|
|
stream,
|
|
ref toplevel,
|
|
} => container::container_toplevel_action("update", name, toplevel, stream, writer).await,
|
|
|
|
PrivRequest::CreateContainer {
|
|
ref name,
|
|
stream,
|
|
ref toplevel,
|
|
} => container::container_toplevel_action("create", name, toplevel, stream, writer).await,
|
|
|
|
PrivRequest::RealiseClosure {
|
|
ref agent,
|
|
ref store_path,
|
|
} => closure::realise_closure(agent, store_path).await,
|
|
|
|
PrivRequest::RefreshCacheFacts => closure::refresh_cache_facts().await,
|
|
|
|
PrivRequest::DestroyContainer { ref name } => container::destroy_container(name).await,
|
|
|
|
PrivRequest::ListContainers => container::container_run(&["list"]).await,
|
|
|
|
PrivRequest::WriteNspawnFlags {
|
|
container,
|
|
binds,
|
|
isolation,
|
|
load_credentials,
|
|
hive_facts,
|
|
private_users,
|
|
nix_builds,
|
|
} => {
|
|
container_config::handle_write_nspawn_flags_blocking(
|
|
container,
|
|
binds,
|
|
isolation,
|
|
load_credentials,
|
|
*hive_facts,
|
|
private_users,
|
|
nix_builds,
|
|
)
|
|
.await
|
|
}
|
|
|
|
PrivRequest::EnsureAgentSocketDir {
|
|
ref name,
|
|
private_users,
|
|
} => files::ensure_agent_socket_dir(name, private_users),
|
|
|
|
PrivRequest::WriteResourceLimits {
|
|
ref container,
|
|
ref memory_max,
|
|
ref cpu_quota,
|
|
cpu_weight,
|
|
io_weight,
|
|
} => container_config::write_resource_limits(
|
|
container, memory_max, cpu_quota, cpu_weight, io_weight,
|
|
),
|
|
|
|
PrivRequest::RemoveServiceDropin { ref container } => {
|
|
container_config::remove_service_dropin(container)
|
|
}
|
|
|
|
PrivRequest::DaemonReload => container_config::daemon_reload().await,
|
|
|
|
PrivRequest::ReloadGatewayNginx => infra::sync_gateway_nginx().await,
|
|
|
|
PrivRequest::SetAgentPaused {
|
|
ref agent_name,
|
|
paused,
|
|
} => files::exec_set_agent_paused(agent_name, paused),
|
|
|
|
PrivRequest::ControlInfraContainer { container, action } => {
|
|
infra::control_infra_container(container, action).await
|
|
}
|
|
|
|
PrivRequest::EnsureAgentSubvolume { ref agent_name } => {
|
|
storage::exec_ensure_agent_subvolume(agent_name).await
|
|
}
|
|
|
|
PrivRequest::DeleteAgentSubvolume { ref agent_name } => {
|
|
storage::exec_delete_agent_subvolume(agent_name).await
|
|
}
|
|
|
|
PrivRequest::EnsureBtrfsQuota => storage::ensure_btrfs_quota().await,
|
|
|
|
PrivRequest::ReadSubvolumeUsage { ref agent_name } => {
|
|
storage::exec_read_subvolume_usage(agent_name).await
|
|
}
|
|
|
|
PrivRequest::SetSubvolumeQuota {
|
|
ref agent_name,
|
|
limit_bytes,
|
|
} => storage::exec_set_subvolume_quota(agent_name, limit_bytes).await,
|
|
|
|
PrivRequest::MeasureAgentDisk { ref agent_name } => {
|
|
storage::exec_measure_agent_disk(agent_name).await
|
|
}
|
|
|
|
PrivRequest::UpgradeAgentSubvolume { ref agent_name } => {
|
|
storage::exec_upgrade_agent_subvolume(agent_name).await
|
|
}
|
|
|
|
PrivRequest::SnapshotAgentSubvolume {
|
|
ref agent_name,
|
|
ref snapshot_name,
|
|
} => storage::exec_snapshot_agent_subvolume(agent_name, snapshot_name).await,
|
|
|
|
PrivRequest::DeleteAgentSnapshot {
|
|
ref agent_name,
|
|
ref snapshot_name,
|
|
} => storage::exec_delete_agent_snapshot(agent_name, snapshot_name).await,
|
|
|
|
PrivRequest::SendAgentSnapshotToFile {
|
|
ref agent_name,
|
|
ref snapshot_name,
|
|
ref parent_snapshot_name,
|
|
ref dest_file_name,
|
|
} => {
|
|
storage::exec_send_agent_snapshot_to_file(
|
|
agent_name,
|
|
snapshot_name,
|
|
parent_snapshot_name.as_deref(),
|
|
dest_file_name,
|
|
)
|
|
.await
|
|
}
|
|
|
|
PrivRequest::SyncAgentTmpfiles => {
|
|
files::remove_legacy_tmpfiles()?;
|
|
Ok((String::new(), String::new()))
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use hive_priv_sock::{PrivEvent, PrivRequest, PrivResponse};
|
|
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
|
|
use tokio::net::UnixStream;
|
|
|
|
use super::{check_socket_activation, dispatch, handle, response_of};
|
|
|
|
/// An older hive-c0re's `SyncAgentTmpfiles` still decodes, payload and
|
|
/// all, so it gets the legacy cleanup rather than a parse error. Control:
|
|
/// an unknown op does not decode.
|
|
#[test]
|
|
fn legacy_sync_agent_tmpfiles_request_still_decodes() {
|
|
let legacy =
|
|
r#"{"op":"sync_agent_tmpfiles","agents":[{"name":"atlas","uid":1000,"gid":994}]}"#;
|
|
assert!(matches!(
|
|
serde_json::from_str::<PrivRequest>(legacy),
|
|
Ok(PrivRequest::SyncAgentTmpfiles)
|
|
));
|
|
assert!(serde_json::from_str::<PrivRequest>(r#"{"op":"sync_agent_tmpfilez"}"#).is_err());
|
|
}
|
|
|
|
// ---- socket activation ----
|
|
|
|
/// Only systemd's own handover starts the daemon: at least one fd, and
|
|
/// `LISTEN_PID` naming this process. Every other shape is refused, and the
|
|
/// refusal says how to run it.
|
|
#[test]
|
|
fn only_a_handover_addressed_to_this_process_starts_the_daemon() {
|
|
check_socket_activation(Some("1"), Some("42"), 42).expect("the normal handover");
|
|
check_socket_activation(Some("3"), Some("42"), 42).expect("more fds is still activation");
|
|
|
|
for (fds, pid, why) in [
|
|
(None, None, "neither variable set"),
|
|
(Some("1"), None, "no LISTEN_PID"),
|
|
(None, Some("42"), "no LISTEN_FDS"),
|
|
(Some("0"), Some("42"), "zero fds"),
|
|
(Some("-1"), Some("42"), "negative fds"),
|
|
(Some("many"), Some("42"), "unparsable LISTEN_FDS"),
|
|
(Some("1"), Some("pid"), "unparsable LISTEN_PID"),
|
|
(Some("1"), Some("41"), "LISTEN_PID of another process"),
|
|
(Some("1"), Some("-42"), "negative LISTEN_PID"),
|
|
(Some("1"), Some(""), "empty LISTEN_PID"),
|
|
] {
|
|
let err = check_socket_activation(fds, pid, 42).expect_err(why);
|
|
let msg = format!("{err:#}");
|
|
assert!(
|
|
msg.contains("requires systemd socket activation"),
|
|
"{why}: {msg}"
|
|
);
|
|
assert!(msg.contains("/run/hive/priv.sock"), "{why}: {msg}");
|
|
}
|
|
}
|
|
|
|
// ---- dispatch: parsing, validation, error mapping ----
|
|
|
|
/// Success carries the handler's output and no error; failure carries the
|
|
/// whole context chain and no output.
|
|
#[test]
|
|
fn the_outcome_maps_to_the_wire_response() {
|
|
let ok = response_of(Ok(("out".to_owned(), "err".to_owned())));
|
|
assert!(ok.ok);
|
|
assert_eq!((ok.stdout.as_str(), ok.stderr.as_str()), ("out", "err"));
|
|
assert!(ok.error.is_none());
|
|
|
|
let failed = response_of(Err(anyhow::anyhow!("root cause").context("outer step")));
|
|
assert!(!failed.ok);
|
|
assert!(failed.stdout.is_empty() && failed.stderr.is_empty());
|
|
assert_eq!(failed.error.as_deref(), Some("outer step: root cause"));
|
|
}
|
|
|
|
/// A write half nobody reads, for ops that would stream.
|
|
fn sink() -> (tokio::net::unix::OwnedWriteHalf, UnixStream) {
|
|
let (ours, theirs) = UnixStream::pair().unwrap();
|
|
(ours.into_split().1, theirs)
|
|
}
|
|
|
|
/// The failed response for `line`, asserting nothing else came back.
|
|
async fn refused(line: &str) -> String {
|
|
let (mut w, _keep) = sink();
|
|
let resp = dispatch(line, &mut w).await;
|
|
assert!(!resp.ok, "{line} was accepted");
|
|
assert!(resp.stdout.is_empty() && resp.stderr.is_empty(), "{line}");
|
|
resp.error.expect("a failure carries its error")
|
|
}
|
|
|
|
/// A line that is not a request, or names no op, is answered with the
|
|
/// parse error and runs nothing.
|
|
#[tokio::test]
|
|
async fn malformed_and_unknown_requests_are_refused_as_parse_errors() {
|
|
for line in [
|
|
"",
|
|
"not json",
|
|
"{",
|
|
"[]",
|
|
"null",
|
|
r#"{"op":"rm_rf"}"#,
|
|
r#"{"op":"start_container"}"#,
|
|
r#"{"name":"bob"}"#,
|
|
r#"{"op":"start_container","name":7}"#,
|
|
r#"{"op":"control_infra_container","container":"Forge","action":"stop"}"#,
|
|
] {
|
|
let msg = refused(line).await;
|
|
assert!(msg.starts_with("parse request"), "{line}: {msg}");
|
|
}
|
|
}
|
|
|
|
/// Every op that names an agent or container refuses a bad name from the
|
|
/// request line itself, before any command runs or path is built. A
|
|
/// control (a valid name on the same op) is in the lifecycle and storage
|
|
/// tests; here the point is that each wire op reaches its validator.
|
|
#[tokio::test]
|
|
async fn every_named_op_refuses_a_bad_name_from_the_wire() {
|
|
let store = "/nix/store/0123456789abcdfghijklmnpqrsvwxyz-nixos-system-bob";
|
|
let bad = "../x";
|
|
let named = |op: &str, field: &str, extra: &str| {
|
|
format!(r#"{{"op":"{op}","{field}":"{bad}"{extra}}}"#)
|
|
};
|
|
let mut cases: Vec<(String, &str)> = Vec::new();
|
|
for (op, field) in [
|
|
("start_container", "name"),
|
|
("stop_container", "name"),
|
|
("kill_container", "name"),
|
|
("destroy_container", "name"),
|
|
("ensure_agent_socket_dir", "name"),
|
|
("ensure_agent_subvolume", "agent_name"),
|
|
("delete_agent_subvolume", "agent_name"),
|
|
("read_subvolume_usage", "agent_name"),
|
|
("measure_agent_disk", "agent_name"),
|
|
("upgrade_agent_subvolume", "agent_name"),
|
|
] {
|
|
cases.push((named(op, field, ""), "invalid name"));
|
|
}
|
|
cases.push((
|
|
named("start_failure_log", "name", r#","since_unix":0"#),
|
|
"invalid name",
|
|
));
|
|
for op in ["update_container", "create_container"] {
|
|
let extra = format!(r#","toplevel":"{store}""#);
|
|
cases.push((named(op, "name", &extra), "invalid name"));
|
|
}
|
|
cases.push((
|
|
format!(r#"{{"op":"realise_closure","agent":"{bad}","store_path":"{store}"}}"#),
|
|
"invalid name",
|
|
));
|
|
cases.push((
|
|
named("set_agent_paused", "agent_name", r#","paused":true"#),
|
|
"invalid name",
|
|
));
|
|
cases.push((
|
|
named(
|
|
"set_subvolume_quota",
|
|
"agent_name",
|
|
r#","limit_bytes":null"#,
|
|
),
|
|
"invalid name",
|
|
));
|
|
for op in ["snapshot_agent_subvolume", "delete_agent_snapshot"] {
|
|
let extra = r#","snapshot_name":"hive-x""#;
|
|
cases.push((named(op, "agent_name", extra), "invalid name"));
|
|
}
|
|
cases.push((
|
|
named(
|
|
"send_agent_snapshot_to_file",
|
|
"agent_name",
|
|
r#","snapshot_name":"hive-x","parent_snapshot_name":null,"dest_file_name":"out""#,
|
|
),
|
|
"invalid name",
|
|
));
|
|
// The container-name ops take the `h-` system name.
|
|
for name in ["h-../x", "evil", "hive-forge", "h-"] {
|
|
cases.push((
|
|
format!(r#"{{"op":"remove_service_dropin","container":"{name}"}}"#),
|
|
if name.starts_with("h-") {
|
|
"invalid name"
|
|
} else {
|
|
"not managed by hive"
|
|
},
|
|
));
|
|
}
|
|
cases.push((
|
|
r#"{"op":"write_resource_limits","container":"evil","memory_max":"1G","cpu_quota":"100%"}"#.to_owned(),
|
|
"not managed by hive",
|
|
));
|
|
cases.push((
|
|
r#"{"op":"write_nspawn_flags","container":"evil","binds":[],"isolation":{"bridge":"br","gateway_ip":"10.0.0.1"}}"#.to_owned(),
|
|
"not managed by hive",
|
|
));
|
|
|
|
for (line, want) in &cases {
|
|
let msg = refused(line).await;
|
|
assert!(msg.contains(want), "{line}: {msg}");
|
|
}
|
|
}
|
|
|
|
/// Snapshot labels and the export file name are validated from the wire
|
|
/// too, for a valid agent: a label must carry the `hive-` prefix and the
|
|
/// credential charset, so no label or file name can climb out of the
|
|
/// agent's directory.
|
|
#[tokio::test]
|
|
async fn snapshot_labels_and_export_names_are_validated_from_the_wire() {
|
|
let snap = |op: &str, label: &str| {
|
|
format!(r#"{{"op":"{op}","agent_name":"atlas","snapshot_name":"{label}"}}"#)
|
|
};
|
|
for op in ["snapshot_agent_subvolume", "delete_agent_snapshot"] {
|
|
for label in ["", "x", "../hive-x", "hive-../x", "hive-a/b", "hive-a.b"] {
|
|
let msg = refused(&snap(op, label)).await;
|
|
assert!(msg.contains("invalid"), "{op} {label:?}: {msg}");
|
|
}
|
|
}
|
|
let export = |label: &str, parent: &str, dest: &str| {
|
|
format!(
|
|
r#"{{"op":"send_agent_snapshot_to_file","agent_name":"atlas","snapshot_name":"{label}","parent_snapshot_name":{parent},"dest_file_name":"{dest}"}}"#
|
|
)
|
|
};
|
|
for (label, parent, dest) in [
|
|
("../x", "null", "out"),
|
|
("hive-x", r#""../x""#, "out"),
|
|
("hive-x", "null", "../out"),
|
|
("hive-x", "null", "a/b"),
|
|
("hive-x", "null", ""),
|
|
("hive-x", "null", "out.img"),
|
|
] {
|
|
let msg = refused(&export(label, parent, dest)).await;
|
|
assert!(msg.contains("invalid"), "{label} {parent} {dest}: {msg}");
|
|
}
|
|
}
|
|
|
|
/// The error is the whole anyhow chain, so a caller sees the cause and
|
|
/// the context in one string.
|
|
#[tokio::test]
|
|
async fn the_error_carries_the_whole_context_chain() {
|
|
let msg = refused(r#"{"op":"start_container","name":7}"#).await;
|
|
assert!(msg.starts_with("parse request: "), "{msg}");
|
|
assert!(msg.contains("invalid type"), "{msg}");
|
|
}
|
|
|
|
// ---- handle: the whole connection ----
|
|
|
|
/// Every response line of a connection, in order, once the server closes.
|
|
async fn serve(input: &[u8]) -> Vec<PrivResponse> {
|
|
let (mut client, server) = UnixStream::pair().unwrap();
|
|
let task = tokio::spawn(handle(server));
|
|
client.write_all(input).await.unwrap();
|
|
client.shutdown().await.unwrap();
|
|
let mut raw = String::new();
|
|
client.read_to_string(&mut raw).await.unwrap();
|
|
task.await.unwrap();
|
|
raw.lines()
|
|
.map(|l| {
|
|
// The wire is bare-`PrivResponse` compatible (the event is
|
|
// untagged), and also a `Done` to a new client.
|
|
let bare: PrivResponse = serde_json::from_str(l).unwrap();
|
|
match serde_json::from_str::<PrivEvent>(l).unwrap() {
|
|
PrivEvent::Done(done) => assert_eq!(done.ok, bare.ok),
|
|
PrivEvent::Line(_) => panic!("unexpected stream line: {l}"),
|
|
}
|
|
bare
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
/// One connection answers each request line with one response line, in
|
|
/// order, and a refused request does not close the connection.
|
|
#[tokio::test]
|
|
async fn a_connection_answers_each_line_and_survives_a_refusal() {
|
|
let resps =
|
|
serve(b"garbage\n{\"op\":\"start_container\",\"name\":\"../x\"}\n{\"op\":\"nope\"}\n")
|
|
.await;
|
|
assert_eq!(resps.len(), 3);
|
|
let errs: Vec<_> = resps.iter().map(|r| r.error.as_deref().unwrap()).collect();
|
|
assert!(errs[0].starts_with("parse request"), "{errs:?}");
|
|
assert!(errs[1].contains("invalid name"), "{errs:?}");
|
|
assert!(errs[2].starts_with("parse request"), "{errs:?}");
|
|
assert!(resps.iter().all(|r| !r.ok));
|
|
}
|
|
|
|
/// A connection that sends non-UTF-8 or nothing gets no response and
|
|
/// nothing runs. A line cut off by EOF is still one line: it fails to
|
|
/// parse and is answered with the parse error, never executed.
|
|
#[tokio::test]
|
|
async fn a_broken_request_closes_the_connection_without_running_anything() {
|
|
assert!(serve(b"\xff\xfe\n").await.is_empty());
|
|
assert!(serve(b"").await.is_empty());
|
|
let cut = serve(b"{\"op\":\"start_cont").await;
|
|
assert_eq!(cut.len(), 1);
|
|
assert!(!cut[0].ok);
|
|
assert!(
|
|
cut[0]
|
|
.error
|
|
.as_deref()
|
|
.unwrap()
|
|
.starts_with("parse request")
|
|
);
|
|
}
|
|
}
|