use std::io::{BufRead, BufReader, Read};
use std::path::{Path, PathBuf};
use std::process::{Child, ChildStdout, Command, Output, Stdio};
use std::sync::mpsc;
use std::time::Duration;
use serde_json::Value;
const CHANNEL: &str = "fleet.telemetry";
struct Fx {
tmp: tempfile::TempDir,
}
impl Fx {
fn new() -> Self {
let tmp = tempfile::tempdir().unwrap();
let cfg = tmp.path().join("config.toml");
std::fs::write(&cfg, "").unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(&cfg, std::fs::Permissions::from_mode(0o600)).unwrap();
}
Self { tmp }
}
fn state(&self) -> PathBuf {
self.tmp.path().join("state")
}
fn base(&self) -> Command {
let mut c = Command::new(env!("CARGO_BIN_EXE_net-mesh"));
c.env_remove("NET_MESH_CONFIG")
.env_remove("NET_MESH_PROFILE")
.arg("--config")
.arg(self.tmp.path().join("config.toml"));
c
}
fn exec(&self, args: &[&str], stateful: bool) -> Output {
let mut cmd = self.base();
cmd.args(["--output", "json"]).args(args);
if stateful {
cmd.arg("--state-dir").arg(self.state());
}
let mut child = cmd
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.unwrap();
for _ in 0..400 {
if child.try_wait().unwrap().is_some() {
return child.wait_with_output().unwrap();
}
std::thread::sleep(Duration::from_millis(50));
}
let _ = child.kill();
panic!(
"{args:?} did not exit within 20s: {:?}",
child.wait_with_output()
);
}
fn run(&self, args: &[&str]) -> Output {
self.exec(args, true)
}
fn json(&self, args: &[&str]) -> Value {
let out = self.run(args);
assert!(
out.status.success(),
"{args:?} failed: {}",
String::from_utf8_lossy(&out.stderr)
);
serde_json::from_slice(&out.stdout).unwrap()
}
fn json_stateless(&self, args: &[&str]) -> Value {
let out = self.exec(args, false);
assert!(
out.status.success(),
"{args:?} failed: {}",
String::from_utf8_lossy(&out.stderr)
);
serde_json::from_slice(&out.stdout).unwrap()
}
fn spawn_up(&self, extra: &[&str]) -> Child {
self.base()
.args(["--output", "ndjson", "up", "--bind", "127.0.0.1:0"])
.args(extra)
.arg("--state-dir")
.arg(self.state())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.stdin(Stdio::null())
.spawn()
.unwrap()
}
fn up(&self, extra: &[&str]) -> Up {
let mut child = self.spawn_up(extra);
let mut stdout = BufReader::new(child.stdout.take().unwrap());
let ready = read_row(&mut stdout, &mut child);
assert_eq!(ready["event"], "ready", "{ready}");
Up { child, ready }
}
}
struct Up {
child: Child,
ready: Value,
}
impl Drop for Up {
fn drop(&mut self) {
let _ = self.child.kill();
let _ = self.child.wait();
}
}
fn read_row(stdout: &mut BufReader<ChildStdout>, child: &mut Child) -> Value {
let (tx, rx) = mpsc::channel();
std::thread::scope(|s| {
s.spawn(|| {
let mut line = String::new();
let _ = stdout.read_line(&mut line);
let _ = tx.send(line);
});
match rx.recv_timeout(Duration::from_secs(30)) {
Ok(line) if !line.trim().is_empty() => serde_json::from_str(&line).unwrap(),
_ => {
let _ = child.kill();
let mut err = String::new();
if let Some(mut e) = child.stderr.take() {
let _ = e.read_to_string(&mut err);
}
panic!("no row from up; stderr: {err}");
}
}
})
}
fn stderr_of(out: &Output) -> String {
String::from_utf8_lossy(&out.stderr).to_string()
}
fn channel_root(fx: &Fx, dir: &Path) -> (PathBuf, String) {
let path = dir.join("channel-root.toml");
let made = fx.json_stateless(&["identity", "generate", "--out", path.to_str().unwrap()]);
(path, made["public_key_hex"].as_str().unwrap().to_string())
}
#[test]
fn a_device_joins_with_a_channel_credential_minted_from_an_offline_grant() {
let operator = Fx::new();
let keys = operator.tmp.path().join("keys");
std::fs::create_dir_all(&keys).unwrap();
let (root, root_hex) = channel_root(&operator, &keys);
let first = operator.up(&["--enroll", "--no-port-mapping", "--no-relay"]);
let node_hex = first.ready["enrollment"]["issuer"]
.as_str()
.unwrap()
.to_string();
drop(first);
let grant = keys.join("fleet.grant");
let issued = operator.json_stateless(&[
"channel",
"issue-grant",
"--root-identity",
root.to_str().unwrap(),
"--issuer",
&node_hex,
"--channel",
CHANNEL,
"--out",
grant.to_str().unwrap(),
]);
assert_eq!(issued["root"], root_hex.as_str(), "{issued}");
assert_eq!(issued["rights"], "publish,subscribe");
let grant_text = std::fs::read_to_string(&grant).unwrap();
let root_text = std::fs::read_to_string(&root).unwrap();
let seed = root_text
.lines()
.find_map(|l| l.trim().strip_prefix("seed_hex"))
.unwrap()
.trim_start_matches(['=', ' '])
.trim()
.trim_matches('"');
assert!(!grant_text.contains(seed));
let again = operator.exec(
&[
"channel",
"issue-grant",
"--root-identity",
root.to_str().unwrap(),
"--issuer",
&node_hex,
"--channel",
CHANNEL,
"--out",
grant.to_str().unwrap(),
],
false,
);
assert!(!again.status.success());
let op = operator.up(&[
"--enroll",
"--no-port-mapping",
"--no-relay",
"--channel-grant",
grant.to_str().unwrap(),
]);
let channels = &op.ready["enrollment"]["channels"];
assert_eq!(channels[0]["channel"], CHANNEL, "{}", op.ready);
assert_eq!(channels[0]["root"], root_hex.as_str());
let refused = operator.run(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"subscribe",
]);
assert!(!refused.status.success(), "{refused:?}");
assert!(
stderr_of(&refused).contains("channel serve"),
"{}",
stderr_of(&refused)
);
let refused = operator.run(&[
"invite",
"create",
"--channel",
"fleet.other",
"--channel-rights",
"publish",
]);
assert!(stderr_of(&refused).contains("no grant for channel"));
assert!(!operator
.run(&["invite", "create", "--channel", CHANNEL])
.status
.success());
let refused = operator.run(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"admin",
]);
assert!(!refused.status.success());
let stranger = "11".repeat(32);
operator.json(&["channel", "serve", CHANNEL, "--token-root", &stranger]);
assert!(!operator
.run(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"subscribe"
])
.status
.success());
let served = operator.json(&["channel", "serve", CHANNEL, "--token-root", &root_hex]);
assert_eq!(served["served"], true, "{served}");
assert_eq!(served["token_roots"][0], root_hex.as_str());
let created = operator.json(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"subscribe",
]);
assert_eq!(created["channel"]["channel"], CHANNEL, "{created}");
assert_eq!(created["channel"]["rights"], "subscribe");
assert_eq!(created["channel"]["root"], root_hex.as_str());
assert_eq!(
created["channel"]["credential_expires_at"], issued["not_after"],
"{created}"
);
assert_ne!(
created["channel"]["credential_expires_at"],
created["expires_at"]
);
assert!(created["channel"]["delegation"]
.as_str()
.unwrap()
.starts_with("none"));
assert_eq!(created["channel"]["publisher"], "this node (the issuer)");
for flag in ["--channel-ttl", "--channel-depth", "--channel-publisher"] {
let refused = operator.run(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"subscribe",
flag,
"1",
]);
assert!(!refused.status.success(), "{flag} must be refused");
}
let token = created["token"].as_str().unwrap().to_string();
let inspected = operator.json_stateless(&["invite", "inspect", &token]);
assert_eq!(inspected["channel"]["channel"], CHANNEL, "{inspected}");
assert_eq!(
inspected["relations"],
serde_json::json!(["mesh", "channel"])
);
let agent = Fx::new();
let joined = agent.json(&["join", &token, "--yes"]);
assert_eq!(joined["state"], "joined", "{joined}");
assert_eq!(joined["channel"]["credential"], "stored", "{joined}");
assert_eq!(joined["channel"]["rights"], "subscribe");
assert_eq!(
joined["channel"]["credential_expires_at"], issued["not_after"],
"{joined}"
);
let publish = operator.json(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"publish",
]);
assert_eq!(publish["channel"]["rights"], "publish", "{publish}");
drop(op);
let op = operator.up(&[
"--enroll",
"--no-port-mapping",
"--no-relay",
"--channel-grant",
grant.to_str().unwrap(),
]);
operator.json(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"subscribe",
]);
drop(op);
std::fs::write(operator.state().join("channels.json"), b"{not json").unwrap();
let mut child = operator.spawn_up(&["--enroll", "--no-port-mapping", "--no-relay"]);
let mut exited = None;
for _ in 0..400 {
if let Some(status) = child.try_wait().unwrap() {
exited = Some(status);
break;
}
std::thread::sleep(Duration::from_millis(50));
}
let status = exited.unwrap_or_else(|| {
let _ = child.kill();
panic!("up started over a corrupt channels.json")
});
assert!(!status.success());
let mut err = String::new();
child
.stderr
.take()
.unwrap()
.read_to_string(&mut err)
.unwrap();
assert!(err.contains("channels.json"), "{err}");
}
#[test]
fn a_grant_for_another_node_is_refused_at_start() {
let operator = Fx::new();
let keys = operator.tmp.path().join("keys");
std::fs::create_dir_all(&keys).unwrap();
let (root, _) = channel_root(&operator, &keys);
let grant = keys.join("elsewhere.grant");
operator.json_stateless(&[
"channel",
"issue-grant",
"--root-identity",
root.to_str().unwrap(),
"--issuer",
&"22".repeat(32),
"--channel",
CHANNEL,
"--out",
grant.to_str().unwrap(),
]);
let mut child = operator.spawn_up(&[
"--enroll",
"--no-port-mapping",
"--no-relay",
"--channel-grant",
grant.to_str().unwrap(),
]);
let status = child.wait().unwrap();
assert!(!status.success());
let mut err = String::new();
child
.stderr
.take()
.unwrap()
.read_to_string(&mut err)
.unwrap();
assert!(err.contains("another identity"), "{err}");
let mut file: Value = serde_json::from_slice(&std::fs::read(&grant).unwrap()).unwrap();
file["channel"] = Value::from("fleet.other");
std::fs::write(&grant, serde_json::to_vec(&file).unwrap()).unwrap();
let mut child = operator.spawn_up(&[
"--enroll",
"--no-port-mapping",
"--no-relay",
"--channel-grant",
grant.to_str().unwrap(),
]);
assert!(!child.wait().unwrap().success());
let mut err = String::new();
child
.stderr
.take()
.unwrap()
.read_to_string(&mut err)
.unwrap();
assert!(err.contains("not for the channel it names"), "{err}");
}
fn free_mesh_port() -> u16 {
loop {
let tcp = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let port = tcp.local_addr().unwrap().port();
if std::net::UdpSocket::bind(("127.0.0.1", port)).is_ok() {
return port;
}
}
}
impl Fx {
fn up_on(&self, bind: &str, extra: &[&str]) -> Up {
let mut child = self
.base()
.args(["--output", "ndjson", "up", "--bind", bind])
.args(extra)
.arg("--state-dir")
.arg(self.state())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.stdin(Stdio::null())
.spawn()
.unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
let ready = read_row(&mut stdout, &mut child);
assert_eq!(ready["event"], "ready", "{ready}");
Up { child, ready }
}
}
fn operator_with_grant(operator: &Fx) -> (Vec<String>, String) {
let keys = operator.tmp.path().join("keys");
std::fs::create_dir_all(&keys).unwrap();
let (root, root_hex) = channel_root(operator, &keys);
let first = operator.up(&["--enroll", "--no-port-mapping", "--no-relay"]);
let node_hex = first.ready["enrollment"]["issuer"]
.as_str()
.unwrap()
.to_string();
drop(first);
let grant = keys.join("fleet.grant");
operator.json_stateless(&[
"channel",
"issue-grant",
"--root-identity",
root.to_str().unwrap(),
"--issuer",
&node_hex,
"--channel",
CHANNEL,
"--out",
grant.to_str().unwrap(),
]);
let args = [
"--enroll",
"--no-port-mapping",
"--no-relay",
"--channel-grant",
grant.to_str().unwrap(),
]
.map(str::to_string)
.to_vec();
(args, root_hex)
}
fn strs(args: &[String]) -> Vec<&str> {
args.iter().map(String::as_str).collect()
}
fn wait_for_channel(fx: &Fx, what: &str, done: impl Fn(&Value) -> bool) -> Value {
let mut last = Value::Null;
for _ in 0..180 {
last = fx.json(&["channel", "status"]);
if done(&last["joined"]) {
return last;
}
std::thread::sleep(Duration::from_millis(500));
}
panic!("{what}: never got there; last status {last}");
}
#[test]
fn a_subscribing_device_is_acked_resubscribes_and_leaves() {
let operator = Fx::new();
let (op_args, root_hex) = operator_with_grant(&operator);
let bind = format!("127.0.0.1:{}", free_mesh_port());
let op = operator.up_on(&bind, &strs(&op_args));
operator.json(&["channel", "serve", CHANNEL, "--token-root", &root_hex]);
let invite = |fx: &Fx| {
let created = fx.json(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"subscribe",
]);
created["token"].as_str().unwrap().to_string()
};
let agent = Fx::new();
agent.json(&["join", &invite(&operator), "--yes"]);
let node = agent.up(&[]);
let ch = &node.ready["joined"]["channel"];
assert_eq!(ch["state"], "active", "{}", node.ready);
assert_eq!(ch["subscribed"], true, "{}", node.ready);
assert_eq!(ch["rights"], "subscribe");
assert_eq!(ch["root"], root_hex.as_str());
assert!(ch.get("publish_installed").is_none(), "subscribe only");
drop(op);
let op = operator.up_on(&bind, &strs(&op_args));
wait_for_channel(&agent, "resubscribe after operator restart", |j| {
j["resubscribes"].as_u64() >= Some(1) && j["subscribed"] == true
});
let left = agent.json(&["channel", "leave"]);
assert_eq!(left["newly_left"], true, "{left}");
assert_eq!(left["unsubscribed"], true, "{left}");
let status = agent.json(&["channel", "status"]);
assert_eq!(status["joined"]["state"], "left", "{status}");
assert!(status["joined"].get("subscribed").is_none(), "{status}");
assert_eq!(agent.json(&["channel", "leave"])["newly_left"], false);
drop(node);
let node = agent.up(&[]);
let ch = &node.ready["joined"]["channel"];
assert_eq!(ch["state"], "left", "{}", node.ready);
assert!(ch.get("subscribed").is_none(), "{}", node.ready);
assert_eq!(node.ready["joined"]["attached"], true, "{}", node.ready);
drop(node);
let second = Fx::new();
second.json(&["join", &invite(&operator), "--yes"]);
let running = second.up(&[]);
assert_eq!(running.ready["joined"]["channel"]["subscribed"], true);
let gone = second.json(&["leave"]);
assert_eq!(gone["state"], "left", "{gone}");
assert_eq!(gone["channel_unsubscribed"], true, "{gone}");
drop(running);
drop(op);
}
#[test]
fn a_publishing_device_is_ready_only_under_its_own_trust_and_leaves_exactly() {
let operator = Fx::new();
let (op_args, root_hex) = operator_with_grant(&operator);
let _op = operator.up(&strs(&op_args));
let created = operator.json(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"publish",
]);
let agent = Fx::new();
agent.json(&["join", created["token"].as_str().unwrap(), "--yes"]);
let node = agent.up(&[]);
let ch = &node.ready["joined"]["channel"];
assert_eq!(ch["publish_installed"], true, "{}", node.ready);
assert_eq!(ch["publish_ready"], false, "no local trust yet");
assert!(ch.get("subscribed").is_none(), "publish only");
let publish = |fx: &Fx| fx.run(&["channel", "publish", CHANNEL, "--data", "reading 42"]);
let open: Value = serde_json::from_slice(&publish(&agent).stdout).unwrap();
assert!(open["gate"].as_str().unwrap().starts_with("open"), "{open}");
assert!(agent.json(&["channel", "status"])["joined"]
.get("published_at")
.is_none());
agent.json(&[
"channel",
"serve",
CHANNEL,
"--token-root",
&"33".repeat(32),
]);
std::thread::sleep(Duration::from_millis(2500));
let status = agent.json(&["channel", "status"]);
assert_eq!(status["joined"]["publish_ready"], false, "{status}");
let denied = publish(&agent);
assert!(!denied.status.success());
assert!(
stderr_of(&denied).contains("publish denied by channel ACL"),
"{}",
stderr_of(&denied)
);
agent.json(&["channel", "serve", CHANNEL, "--token-root", &root_hex]);
wait_for_channel(&agent, "publish ready under local trust", |j| {
j["publish_ready"] == true
});
let passed = agent.json(&["channel", "publish", CHANNEL, "--data", "reading 42"]);
assert_eq!(passed["gate"], "passed", "{passed}");
let status = agent.json(&["channel", "status"]);
assert!(
status["joined"]["published_at"].as_u64().is_some(),
"{status}"
);
let left = agent.json(&["channel", "leave"]);
assert_eq!(left["publish_removed"], true, "{left}");
assert_eq!(left["publish_stop"], "confirmed", "{left}");
let status = agent.json(&["channel", "status"]);
assert_eq!(status["joined"]["publish_installed"], false, "{status}");
assert_eq!(status["joined"]["publish_ready"], false, "{status}");
assert!(stderr_of(&publish(&agent)).contains("publish denied by channel ACL"));
assert_eq!(status["served"][0]["channel"], CHANNEL, "{status}");
drop(node);
}
#[test]
fn a_joined_device_adds_leaves_and_rejoins_a_channel_with_standalone_links() {
let operator = Fx::new();
let (op_args, root_hex) = operator_with_grant(&operator);
let _op = operator.up(&strs(&op_args));
operator.json(&["channel", "serve", CHANNEL, "--token-root", &root_hex]);
let agent = Fx::new();
let mesh_only = operator.json(&["invite", "create"]);
agent.json(&["join", mesh_only["token"].as_str().unwrap(), "--yes"]);
let node = agent.up(&[]);
assert!(
node.ready["joined"].get("channel").is_none(),
"{}",
node.ready
);
let link = |fx: &Fx| {
let created = fx.json(&["channel", "invite", CHANNEL, "--rights", "subscribe"]);
assert_eq!(created["standalone"], true, "{created}");
created["token"].as_str().unwrap().to_string()
};
let first = link(&operator);
let inspected = operator.json_stateless(&["invite", "inspect", &first]);
assert_eq!(
inspected["relations"],
serde_json::json!(["channel"]),
"{inspected}"
);
assert!(!agent.run(&["channel", "join", &first]).status.success());
let joined = agent.json(&["channel", "join", &first, "--yes"]);
assert_eq!(joined["state"], "active", "{joined}");
assert_eq!(joined["subscribed"], true, "{joined}");
let second = link(&operator);
let refused = agent.run(&["channel", "join", &second, "--yes"]);
assert!(!refused.status.success());
assert!(stderr_of(&refused).contains("already holds an active credential"));
drop(node);
let node = agent.up(&[]);
wait_for_channel_list(
&agent,
"standalone channel resubscribed after restart",
|l| l.len() == 1 && l[0]["subscribed"] == true && l[0]["state"] == "active",
);
let left = agent.json(&["channel", "leave"]);
assert_eq!(left["newly_left"], true, "{left}");
assert_eq!(left["unsubscribed"], true, "{left}");
let spent = agent.run(&["channel", "join", &first, "--yes"]);
assert!(
stderr_of(&spent).contains("left that channel membership"),
"{}",
stderr_of(&spent)
);
let rejoined = agent.json(&["channel", "join", &second, "--yes"]);
assert_eq!(rejoined["subscribed"], true, "{rejoined}");
let status = agent.json(&["channel", "status"]);
let states: Vec<&str> = status["standalone"]
.as_array()
.unwrap()
.iter()
.map(|l| l["state"].as_str().unwrap())
.collect();
assert_eq!(
states.iter().filter(|s| **s == "left").count(),
1,
"{status}"
);
assert_eq!(
states.iter().filter(|s| **s == "active").count(),
1,
"{status}"
);
assert_eq!(agent.json(&["channel", "leave"])["newly_left"], true);
assert_eq!(agent.json(&["channel", "leave"])["newly_left"], false);
assert_eq!(agent.json(&["node", "status"])["link"]["attached"], true);
drop(node);
}
fn wait_for_channel_list(fx: &Fx, what: &str, done: impl Fn(&Vec<Value>) -> bool) -> Value {
let mut last = Value::Null;
for _ in 0..120 {
last = fx.json(&["channel", "status"]);
if last["standalone"].as_array().is_some_and(&done) {
return last;
}
std::thread::sleep(Duration::from_millis(500));
}
panic!("{what}: never got there; last status {last}");
}
#[test]
fn an_offline_channel_leave_completes_locally_and_restart_honours_it() {
let operator = Fx::new();
let (op_args, root_hex) = operator_with_grant(&operator);
let _op = operator.up(&strs(&op_args));
operator.json(&["channel", "serve", CHANNEL, "--token-root", &root_hex]);
let created = operator.json(&[
"invite",
"create",
"--channel",
CHANNEL,
"--channel-rights",
"subscribe",
]);
let agent = Fx::new();
agent.json(&["join", created["token"].as_str().unwrap(), "--yes"]);
let node = agent.up(&[]);
assert_eq!(node.ready["joined"]["channel"]["subscribed"], true);
drop(node);
let left = agent.json(&["channel", "leave"]);
assert_eq!(left["runtime"], "not_running", "{left}");
assert_eq!(left["newly_left"], true, "{left}");
assert_eq!(left["unsubscribed"], false, "{left}");
assert!(
left["unsubscribe_detail"]
.as_str()
.unwrap()
.starts_with("unconfirmed"),
"{left}"
);
assert_eq!(agent.json(&["channel", "leave"])["newly_left"], false);
let node = agent.up(&[]);
let ch = &node.ready["joined"]["channel"];
assert_eq!(ch["state"], "left", "{}", node.ready);
assert!(ch.get("subscribed").is_none(), "{}", node.ready);
drop(node);
}