sn-node: Yggdrasil config generation and admin socket client (genconf, status)

This commit is contained in:
2026-09-24 15:58:12 +02:00
parent 23dc906da0
commit c422deee7c
8 changed files with 1046 additions and 2 deletions
+15
View File
@@ -0,0 +1,15 @@
[package]
name = "sn-node"
description = "SpiderNet node helper: Yggdrasil overlay config generation and status queries"
version.workspace = true
edition.workspace = true
license.workspace = true
[dependencies]
tokio = { workspace = true }
serde_json = { workspace = true }
thiserror = { workspace = true }
clap = { workspace = true }
[lints]
workspace = true
+343
View File
@@ -0,0 +1,343 @@
//! Yggdrasil configuration generation for a `SpiderNet` node.
//!
//! Yggdrasil's own `-genconf` output is used as the base (so we never
//! hard-code version-specific defaults); the `SpiderNet` settings are then
//! patched into that JSON base as validated, typed values. The result is
//! written as JSON — Yggdrasil accepts JSON config via `-useconffile`.
use std::path::{Path, PathBuf};
use crate::error::NodeError;
use crate::yggdrasil::AdminEndpoint;
/// A validated Yggdrasil peering URI (`tcp://`, `tls://`, `quic://` or
/// `socks://`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PeerUri(String);
impl PeerUri {
pub fn parse(input: &str) -> Result<Self, NodeError> {
Ok(Self(validate_scheme(input, "peering")?))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
/// A validated listener URI a node binds for incoming peerings
/// (`tcp://[::]:1337` style).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListenUri(String);
impl ListenUri {
pub fn parse(input: &str) -> Result<Self, NodeError> {
let trimmed = input.trim();
if trimmed.starts_with("socks://") {
return Err(NodeError::InvalidUri {
uri: input.to_string(),
hint: "listeners bind an address: use tcp://, tls:// or quic://",
});
}
Ok(Self(validate_scheme(trimmed, "listen")?))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
const PEER_SCHEMES: [&str; 4] = ["tcp://", "tls://", "quic://", "socks://"];
fn validate_scheme(input: &str, kind: &'static str) -> Result<String, NodeError> {
let trimmed = input.trim();
let scheme = PEER_SCHEMES
.iter()
.find(|scheme| trimmed.starts_with(**scheme));
match scheme {
Some(scheme) if trimmed.len() > scheme.len() => Ok(trimmed.to_string()),
_ => Err(NodeError::InvalidUri {
uri: input.to_string(),
hint: match kind {
"peering" => "peer URIs must start with tcp://, tls://, quic:// or socks://",
_ => "listen URIs must start with tcp://, tls:// or quic://",
},
}),
}
}
/// A validated Yggdrasil public key: 64 hex characters.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PublicKeyHex(String);
impl PublicKeyHex {
pub fn parse(input: &str) -> Result<Self, NodeError> {
let trimmed = input.trim().to_lowercase();
let valid = trimmed.len() == 64 && trimmed.chars().all(|c| c.is_ascii_hexdigit());
if valid {
Ok(Self(trimmed))
} else {
Err(NodeError::InvalidPublicKey(input.to_string()))
}
}
pub fn as_str(&self) -> &str {
&self.0
}
}
/// Which TUN interface Yggdrasil should use.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum IfName {
/// `auto` — let Yggdrasil pick (default).
Auto,
/// `none` — headless router-only mode, no TUN device.
Headless,
Named(String),
}
/// Multicast (same-LAN) auto-peering mode.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum MulticastMode {
/// Leave Yggdrasil's generated default untouched.
PlatformDefault,
/// Disable multicast peering entirely.
Disabled,
/// Explicit interface entries (regex-matched).
Interfaces(Vec<MulticastInterface>),
}
/// One multicast peering entry, mirroring Yggdrasil's JSON shape.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MulticastInterface {
/// Interface name regex, e.g. `eth0` or `.*`.
pub regex: String,
pub beacon: bool,
pub listen: bool,
pub port: u16,
}
/// Everything `SpiderNet` wants to control in the Yggdrasil config.
/// Unlisted fields keep the values Yggdrasil's `-genconf` produced.
#[derive(Debug, Clone, Default)]
pub struct NodeSettings {
pub peers: Vec<PeerUri>,
pub listen: Vec<ListenUri>,
pub multicast: Option<MulticastMode>,
pub admin_listen: Option<AdminEndpoint>,
pub if_name: Option<IfName>,
/// Neighbor keys allowed to peer with us; empty means "allow all".
pub allowed_public_keys: Vec<PublicKeyHex>,
pub private_key_path: Option<PathBuf>,
}
/// Run `yggdrasil -genconf -json` and patch the `SpiderNet` settings into
/// the generated base config. Returns the pretty-printed JSON config.
pub fn generate_config(binary: &str, settings: &NodeSettings) -> Result<String, NodeError> {
let output = std::process::Command::new(binary)
.arg("-genconf")
.arg("-json")
.output()?;
if !output.status.success() {
let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
return Err(NodeError::YggdrasilBinary {
binary: binary.to_string(),
stderr,
});
}
let mut base: serde_json::Value = serde_json::from_slice(&output.stdout)?;
apply_to_base_config(settings, &mut base)?;
Ok(serde_json::to_string_pretty(&base)?)
}
/// Patch the `SpiderNet` settings into an existing (parsed) Yggdrasil base
/// config. Pure function so it can be tested without the binary.
pub fn apply_to_base_config(
settings: &NodeSettings,
base: &mut serde_json::Value,
) -> Result<(), NodeError> {
let obj = base
.as_object_mut()
.ok_or(NodeError::UnexpectedConfigShape)?;
if !settings.peers.is_empty() {
let peers: Vec<&str> = settings.peers.iter().map(PeerUri::as_str).collect();
obj.insert("Peers".to_string(), serde_json::json!(peers));
}
if !settings.listen.is_empty() {
let listen: Vec<&str> = settings.listen.iter().map(ListenUri::as_str).collect();
obj.insert("Listen".to_string(), serde_json::json!(listen));
}
match settings.multicast.as_ref() {
None | Some(MulticastMode::PlatformDefault) => {}
Some(MulticastMode::Disabled) => {
obj.insert("MulticastInterfaces".to_string(), serde_json::json!([]));
}
Some(MulticastMode::Interfaces(interfaces)) => {
let entries: Vec<serde_json::Value> = interfaces
.iter()
.map(|iface| {
serde_json::json!({
"Beacon": iface.beacon,
"Listen": iface.listen,
"Port": iface.port,
"Regex": iface.regex,
})
})
.collect();
obj.insert(
"MulticastInterfaces".to_string(),
serde_json::json!(entries),
);
}
}
if let Some(admin_listen) = settings.admin_listen.as_ref() {
obj.insert(
"AdminListen".to_string(),
serde_json::json!(admin_listen.config_value()),
);
}
if let Some(if_name) = settings.if_name.as_ref() {
let value = match if_name {
IfName::Auto => "auto".to_string(),
IfName::Headless => "none".to_string(),
IfName::Named(name) => name.clone(),
};
obj.insert("IfName".to_string(), serde_json::json!(value));
}
if !settings.allowed_public_keys.is_empty() {
let keys: Vec<&str> = settings
.allowed_public_keys
.iter()
.map(PublicKeyHex::as_str)
.collect();
obj.insert("AllowedPublicKeys".to_string(), serde_json::json!(keys));
}
if let Some(private_key_path) = settings.private_key_path.as_ref() {
obj.insert(
"PrivateKeyPath".to_string(),
serde_json::json!(private_key_path.display().to_string()),
);
}
Ok(())
}
/// Write the config to `path` and restrict permissions to the owner
/// (the file contains the node's private key — it must never be readable
/// by other users, and it must never be committed to any repository).
pub fn write_config(path: &Path, json: &str) -> Result<(), NodeError> {
use std::io::Write as _;
let mut file = std::fs::File::create(path)?;
file.write_all(json.as_bytes())?;
file.flush()?;
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600))?;
}
Ok(())
}
#[cfg(test)]
mod tests {
#![allow(clippy::unwrap_used, clippy::expect_used)]
use super::*;
#[test]
fn peer_uri_accepts_supported_schemes() {
for uri in [
"tcp://host:1337",
"tls://host:1337",
"quic://host:1337",
"socks://p:1/h:2",
] {
assert!(PeerUri::parse(uri).is_ok(), "{uri} should be valid");
}
}
#[test]
fn peer_uri_rejects_bare_or_unknown_schemes() {
for uri in ["host:1337", "http://host:80", "tcp://", ""] {
assert!(PeerUri::parse(uri).is_err(), "{uri} should be invalid");
}
}
#[test]
fn listen_uri_rejects_socks() {
assert!(ListenUri::parse("socks://[::]:1337").is_err());
assert!(ListenUri::parse("tcp://[::]:1337").is_ok());
}
#[test]
fn public_key_requires_64_hex_chars() {
assert!(PublicKeyHex::parse(&"a".repeat(64)).is_ok());
assert!(PublicKeyHex::parse(&"a".repeat(63)).is_err());
assert!(PublicKeyHex::parse(&"z".repeat(64)).is_err());
}
#[test]
fn admin_endpoint_parse_and_roundtrip() {
let tcp = AdminEndpoint::parse("tcp://localhost:9001").unwrap();
assert_eq!(
tcp,
AdminEndpoint::Tcp {
host: "localhost".into(),
port: 9001
}
);
assert_eq!(tcp.config_value(), "tcp://localhost:9001");
let unix = AdminEndpoint::parse("unix:///var/run/yggdrasil.sock").unwrap();
assert_eq!(unix.config_value(), "unix:///var/run/yggdrasil.sock");
assert!(AdminEndpoint::parse("http://localhost:9001").is_err());
assert!(AdminEndpoint::parse("tcp://localhost").is_err());
}
#[test]
fn patches_land_in_base_config() {
let mut base: serde_json::Value =
serde_json::from_str(r#"{"PrivateKey": "secret", "Peers": [], "Listen": []}"#).unwrap();
let settings = NodeSettings {
peers: vec![PeerUri::parse("tcp://[200:1234::1]:1337").unwrap()],
listen: vec![ListenUri::parse("tls://[::]:1337").unwrap()],
multicast: Some(MulticastMode::Disabled),
admin_listen: Some(AdminEndpoint::parse("tcp://localhost:9001").unwrap()),
if_name: Some(IfName::Headless),
allowed_public_keys: vec![PublicKeyHex::parse(&"ab".repeat(32)).unwrap()],
private_key_path: Some(PathBuf::from("/etc/spidernet/ygg.key")),
};
apply_to_base_config(&settings, &mut base).unwrap();
assert_eq!(base["Peers"].as_array().unwrap().len(), 1);
assert_eq!(base["Listen"][0], "tls://[::]:1337");
assert_eq!(base["MulticastInterfaces"].as_array().unwrap().len(), 0);
assert_eq!(base["AdminListen"], "tcp://localhost:9001");
assert_eq!(base["IfName"], "none");
assert_eq!(base["AllowedPublicKeys"][0], "ab".repeat(32));
assert_eq!(base["PrivateKeyPath"], "/etc/spidernet/ygg.key");
// The base's own keys survive the patch.
assert_eq!(base["PrivateKey"], "secret");
}
#[test]
fn multicast_interfaces_are_serialized_in_yggdrasil_shape() {
let mut base: serde_json::Value = serde_json::from_str("{}").unwrap();
let settings = NodeSettings {
multicast: Some(MulticastMode::Interfaces(vec![MulticastInterface {
regex: "eth0".to_string(),
beacon: true,
listen: true,
port: 1337,
}])),
..NodeSettings::default()
};
apply_to_base_config(&settings, &mut base).unwrap();
let entry = &base["MulticastInterfaces"][0];
assert_eq!(entry["Regex"], "eth0");
assert_eq!(entry["Beacon"], true);
assert_eq!(entry["Listen"], true);
assert_eq!(entry["Port"], 1337);
}
}
+34
View File
@@ -0,0 +1,34 @@
//! Error type for all sn-node operations.
#[derive(Debug, thiserror::Error)]
pub enum NodeError {
#[error("io error: {0}")]
Io(#[from] std::io::Error),
#[error("json error: {0}")]
Json(#[from] serde_json::Error),
#[error("yggdrasil binary `{binary}` failed: {stderr}")]
YggdrasilBinary { binary: String, stderr: String },
#[error("invalid uri `{uri}`: {hint}")]
InvalidUri { uri: String, hint: &'static str },
#[error("invalid admin endpoint `{0}`: expected `tcp://host:port` or `unix:///path`")]
InvalidAdminEndpoint(String),
#[error("invalid public key `{0}`: expected 64 hex characters")]
InvalidPublicKey(String),
#[error("generated config base has unexpected shape (expected a JSON object)")]
UnexpectedConfigShape,
#[error("admin protocol violation: {0}")]
AdminProtocol(String),
#[error("admin socket answered with an error: {0}")]
AdminFailed(String),
#[error("admin socket operation timed out")]
Timeout,
}
+28
View File
@@ -0,0 +1,28 @@
//! sn-node — `SpiderNet` node helper for the Yggdrasil overlay.
//!
//! `SpiderNet` is software for neighborhood fiber meshes: households peer
//! their internet uplinks and pool bandwidth. This crate is the node-side
//! glue for the overlay layer:
//!
//! - [`config`]: generate a Yggdrasil configuration file from validated,
//! typed `SpiderNet` settings (peers, listeners, admin socket, allowed
//! neighbor keys).
//! - [`yggdrasil`]: client for the Yggdrasil admin socket (JSON over
//! TCP/unix, default `localhost:9001`) to query node status (`getSelf`,
//! `getPeers`).
//!
//! Yggdrasil provides encrypted end-to-end routing between households.
//! It is **not** an anonymity network, and `SpiderNet` never claims to
//! provide anonymity.
//!
//! `SpiderNet` software carries no warranty and accepts no responsibility
//! for its usage or for transported content (AGPL-3.0, sections 15/16).
#![forbid(unsafe_code)]
#![warn(clippy::pedantic)]
// PoC scope: doc-annotation noise, not correctness. Revisit at release.
#![allow(clippy::must_use_candidate, clippy::missing_errors_doc)]
pub mod config;
pub mod error;
pub mod yggdrasil;
+229
View File
@@ -0,0 +1,229 @@
//! CLI entry point for sn-node.
//!
//! KEINE GARANTIE — this software is provided "as is" without warranty of
//! any kind (AGPL-3.0, sections 15/16). Users are solely responsible for
//! what they transmit or receive with it.
use std::path::PathBuf;
use std::time::Duration;
use clap::{Parser, Subcommand};
use sn_node::config::{
self, IfName, ListenUri, MulticastMode, NodeSettings, PeerUri, PublicKeyHex,
};
use sn_node::yggdrasil::{self, AdminEndpoint};
#[derive(Debug, Parser)]
#[command(
name = "sn-node",
about = "SpiderNet node helper: Yggdrasil config generation + overlay status"
)]
struct Args {
#[command(subcommand)]
command: Command,
}
#[derive(Debug, Subcommand)]
enum Command {
/// Generate a Yggdrasil config file for this SpiderNet node.
Genconf {
/// Peering URIs to connect to, comma separated (tcp://, tls://,
/// quic://, socks://).
#[arg(long = "peer", value_delimiter = ',')]
peers: Vec<String>,
/// Listener URIs for incoming peerings, comma separated
/// (e.g. `tcp://[::]:1337`).
#[arg(long = "listen", value_delimiter = ',')]
listen: Vec<String>,
/// Admin socket endpoint (tcp://host:port or unix:///path).
#[arg(long, default_value = "tcp://localhost:9001")]
admin: String,
/// Public keys of neighbors allowed to peer with us (64 hex
/// characters, comma separated). Empty: any node may peer.
#[arg(long = "allow-key", value_delimiter = ',')]
allow_key: Vec<String>,
/// Store the node's private key in an external file instead of
/// embedding it in the config.
#[arg(long)]
private_key_path: Option<PathBuf>,
/// TUN interface: `auto`, `none` (headless router-only), or a name.
#[arg(long, default_value = "auto")]
ifname: String,
/// Disable multicast (same-LAN) auto-peering.
#[arg(long)]
no_multicast: bool,
/// Path to the yggdrasil binary.
#[arg(long, default_value = "yggdrasil")]
yggdrasil_bin: String,
/// Output path for the generated config.
#[arg(short, long, default_value = "yggdrasil.conf")]
out: PathBuf,
},
/// Query the local Yggdrasil node via the admin socket.
Status {
/// Admin socket endpoint (tcp://host:port or unix:///path).
#[arg(long, default_value = "tcp://localhost:9001")]
admin: String,
/// Operation timeout in seconds.
#[arg(long, default_value_t = 3)]
timeout_secs: u64,
},
}
fn parse_ifname(input: &str) -> Result<IfName, String> {
match input.trim() {
"auto" => Ok(IfName::Auto),
"none" => Ok(IfName::Headless),
name if !name.is_empty() => Ok(IfName::Named(name.to_string())),
_ => Err("--ifname must be `auto`, `none` or an interface name".to_string()),
}
}
fn parse_uris<T>(
inputs: &[String],
what: &str,
parse: fn(&str) -> Result<T, sn_node::error::NodeError>,
) -> Result<Vec<T>, String> {
inputs
.iter()
.map(|input| parse(input).map_err(|err| format!("{what}: {err}")))
.collect()
}
#[tokio::main]
async fn main() {
if let Err(report) = run().await {
eprintln!("NO WARRANTY — SpiderNet sn-node (AGPL-3.0, no liability)");
eprintln!("error: {report}");
std::process::exit(1);
}
}
async fn run() -> Result<(), String> {
let args = Args::parse();
match args.command {
Command::Genconf {
peers,
listen,
admin,
allow_key,
private_key_path,
ifname,
no_multicast,
yggdrasil_bin,
out,
} => {
run_genconf(
peers,
listen,
admin,
allow_key,
private_key_path,
ifname,
no_multicast,
yggdrasil_bin,
out,
)
.await
}
Command::Status {
admin,
timeout_secs,
} => run_status(&admin, timeout_secs).await,
}
}
#[allow(clippy::too_many_arguments)]
async fn run_genconf(
peers: Vec<String>,
listen: Vec<String>,
admin: String,
allow_key: Vec<String>,
private_key_path: Option<PathBuf>,
ifname: String,
no_multicast: bool,
yggdrasil_bin: String,
out: PathBuf,
) -> Result<(), String> {
let peers = parse_uris(&peers, "peer", PeerUri::parse)?;
let listen = parse_uris(&listen, "listen", ListenUri::parse)?;
let admin_listen = AdminEndpoint::parse(&admin).map_err(|err| err.to_string())?;
let allowed_public_keys = allow_key
.iter()
.map(|key| PublicKeyHex::parse(key).map_err(|err| err.to_string()))
.collect::<Result<Vec<_>, String>>()?;
let if_name = parse_ifname(&ifname)?;
let settings = NodeSettings {
peers,
listen,
multicast: Some(if no_multicast {
MulticastMode::Disabled
} else {
MulticastMode::PlatformDefault
}),
admin_listen: Some(admin_listen),
if_name: Some(if_name),
allowed_public_keys,
private_key_path,
};
let json = config::generate_config(&yggdrasil_bin, &settings).map_err(|err| err.to_string())?;
config::write_config(&out, &json).map_err(|err| err.to_string())?;
println!("NO WARRANTY — SpiderNet sn-node (AGPL-3.0, no liability)");
println!();
println!("config written: {}", out.display());
println!(
"mode: start with `yggdrasil -useconffile {}`",
out.display()
);
println!();
println!(
"NOTE: the config contains this node's PRIVATE key. Keep it at \
permission 600, never commit it, never share it."
);
Ok(())
}
async fn run_status(admin: &str, timeout_secs: u64) -> Result<(), String> {
let endpoint = AdminEndpoint::parse(admin).map_err(|err| err.to_string())?;
let timeout = Duration::from_secs(timeout_secs);
let info = yggdrasil::AdminClient::get_self(&endpoint, timeout)
.await
.map_err(|err| err.to_string())?;
println!("NO WARRANTY — SpiderNet sn-node (AGPL-3.0, no liability)");
println!();
println!("address: {}", info.address);
if let Some(subnet) = info.subnet.as_ref() {
println!("subnet: {subnet}");
}
if let Some(version) = info.build_version.as_ref() {
let name = info.build_name.as_deref().unwrap_or("yggdrasil");
println!("version: {name} {version}");
}
println!("key: {}", info.key);
let peers = yggdrasil::AdminClient::get_peers(&endpoint, timeout)
.await
.map_err(|err| err.to_string())?;
println!("peers: {}", peers.len());
for peer in &peers {
match peer.remote.as_ref() {
Some(remote) => println!(" {} via {remote}", peer.address),
None => println!(" {}", peer.address),
}
}
Ok(())
}
+268
View File
@@ -0,0 +1,268 @@
//! Client for the Yggdrasil admin socket.
//!
//! Protocol (per the official docs): JSON stanza + `\n` over a TCP or unix
//! socket; the default endpoint is `localhost:9001`. A response always has
//! a `"status"` field (`"success"` or `"error"`), optionally a `"response"`
//! section and an `"error"` field. By default Yggdrasil closes the
//! connection after answering, so this client sends one request per
//! connection.
use std::net::Ipv6Addr;
use std::path::PathBuf;
use std::time::Duration;
use crate::error::NodeError;
/// Where the admin socket lives, in Yggdrasil endpoint syntax
/// (`tcp://host:port` or `unix:///path/to/socket`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AdminEndpoint {
Tcp { host: String, port: u16 },
Unix { path: PathBuf },
}
impl AdminEndpoint {
/// Yggdrasil's default admin endpoint: `tcp://localhost:9001`.
pub fn default_local() -> Self {
Self::Tcp {
host: "localhost".to_string(),
port: 9001,
}
}
pub fn parse(input: &str) -> Result<Self, NodeError> {
let trimmed = input.trim();
if let Some(rest) = trimmed.strip_prefix("tcp://") {
let (host, port) = rest
.rsplit_once(':')
.ok_or_else(|| NodeError::InvalidAdminEndpoint(trimmed.to_string()))?;
let port: u16 = port
.parse()
.map_err(|_| NodeError::InvalidAdminEndpoint(trimmed.to_string()))?;
if host.is_empty() {
return Err(NodeError::InvalidAdminEndpoint(trimmed.to_string()));
}
Ok(Self::Tcp {
host: host.to_string(),
port,
})
} else if let Some(rest) = trimmed.strip_prefix("unix://") {
if rest.is_empty() {
return Err(NodeError::InvalidAdminEndpoint(trimmed.to_string()));
}
Ok(Self::Unix {
path: PathBuf::from(rest),
})
} else {
Err(NodeError::InvalidAdminEndpoint(trimmed.to_string()))
}
}
/// Value for the `AdminListen` field of a Yggdrasil config file.
pub fn config_value(&self) -> String {
match self {
Self::Tcp { host, port } => format!("tcp://{host}:{port}"),
Self::Unix { path } => format!("unix://{}", path.display()),
}
}
}
/// The node's own Yggdrasil address — a validated IPv6 address inside the
/// Yggdrasil range.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct YggAddr(Ipv6Addr);
impl YggAddr {
pub const fn new(addr: Ipv6Addr) -> Self {
Self(addr)
}
pub const fn inner(self) -> Ipv6Addr {
self.0
}
}
impl std::fmt::Display for YggAddr {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.0)
}
}
/// One connection to the admin socket; one request per connection.
pub struct AdminClient {
stream: AdminStream,
}
enum AdminStream {
Tcp(tokio::net::TcpStream),
Unix(tokio::net::UnixStream),
}
async fn connect_stream(endpoint: &AdminEndpoint) -> Result<AdminStream, NodeError> {
match endpoint {
AdminEndpoint::Tcp { host, port } => tokio::net::TcpStream::connect((host.as_str(), *port))
.await
.map(AdminStream::Tcp)
.map_err(NodeError::Io),
AdminEndpoint::Unix { path } => tokio::net::UnixStream::connect(path)
.await
.map(AdminStream::Unix)
.map_err(NodeError::Io),
}
}
impl AdminClient {
pub async fn connect(endpoint: &AdminEndpoint, timeout: Duration) -> Result<Self, NodeError> {
let stream = tokio::time::timeout(timeout, connect_stream(endpoint))
.await
.map_err(|_| NodeError::Timeout)??;
Ok(Self { stream })
}
/// Send one request verb (e.g. `getself`, `getpeers`) and return the
/// raw response section.
pub async fn request(&mut self, verb: &str) -> Result<serde_json::Value, NodeError> {
// `verb` is a compile-time-known constant at every call site, not
// user input — no injection risk.
let request = format!("{{\"request\":\"{verb}\"}}\n");
let bytes = request.as_bytes();
match &mut self.stream {
AdminStream::Tcp(stream) => {
tokio::io::AsyncWriteExt::write_all(stream, bytes).await?;
let mut buffer = Vec::new();
tokio::io::AsyncReadExt::read_to_end(stream, &mut buffer).await?;
Self::parse_response(&buffer, verb)
}
AdminStream::Unix(stream) => {
tokio::io::AsyncWriteExt::write_all(stream, bytes).await?;
let mut buffer = Vec::new();
tokio::io::AsyncReadExt::read_to_end(stream, &mut buffer).await?;
Self::parse_response(&buffer, verb)
}
}
}
/// Query the node's own status (address, subnet, key, build info).
pub async fn get_self(
endpoint: &AdminEndpoint,
timeout: Duration,
) -> Result<SelfInfo, NodeError> {
let mut client = Self::connect(endpoint, timeout).await?;
let value = client.request("getself").await?;
SelfInfo::from_response(&value)
}
/// Query the node's active peer sessions.
pub async fn get_peers(
endpoint: &AdminEndpoint,
timeout: Duration,
) -> Result<Vec<PeerInfo>, NodeError> {
let mut client = Self::connect(endpoint, timeout).await?;
let value = client.request("getpeers").await?;
peers_from_response(&value)
}
fn parse_response(buffer: &[u8], verb: &str) -> Result<serde_json::Value, NodeError> {
let envelope: serde_json::Value = serde_json::from_slice(buffer)?;
let status = envelope
.get("status")
.and_then(serde_json::Value::as_str)
.ok_or_else(|| {
NodeError::AdminProtocol(format!("{verb}: response has no `status` field"))
})?;
if status != "success" {
let error = envelope
.get("error")
.and_then(serde_json::Value::as_str)
.unwrap_or("unknown error");
return Err(NodeError::AdminFailed(error.to_string()));
}
envelope
.get("response")
.cloned()
.ok_or_else(|| NodeError::AdminProtocol(format!("{verb}: response has no data")))
}
}
/// Information about the local Yggdrasil node (`getSelf`).
#[derive(Debug, Clone)]
pub struct SelfInfo {
pub address: YggAddr,
pub key: String,
pub subnet: Option<String>,
pub coords: Option<serde_json::Value>,
pub build_name: Option<String>,
pub build_version: Option<String>,
}
impl SelfInfo {
/// Parse the `getSelf` response section: exactly one record, keyed by
/// the node's IPv6 address.
pub fn from_response(value: &serde_json::Value) -> Result<Self, NodeError> {
let obj = value.as_object().ok_or_else(|| {
NodeError::AdminProtocol("getSelf: response section is not an object".to_string())
})?;
let (address, record) = obj.iter().next().ok_or_else(|| {
NodeError::AdminProtocol("getSelf: response contains no record".to_string())
})?;
let address = YggAddr::new(address.parse::<Ipv6Addr>().map_err(|_| {
NodeError::AdminProtocol(format!("getSelf: `{address}` is not an IPv6 address"))
})?);
let str_field = |name: &str| {
record
.get(name)
.and_then(serde_json::Value::as_str)
.map(str::to_string)
};
Ok(Self {
address,
key: str_field("key").ok_or_else(|| {
NodeError::AdminProtocol("getSelf: record has no `key`".to_string())
})?,
subnet: str_field("subnet"),
coords: record.get("coords").cloned(),
build_name: str_field("build_name"),
build_version: str_field("build_version"),
})
}
}
/// One active peer session (`getPeers`).
#[derive(Debug, Clone)]
pub struct PeerInfo {
pub address: YggAddr,
pub key: Option<String>,
/// The peering URI of the remote peer, when reported.
pub remote: Option<String>,
pub coords: Option<serde_json::Value>,
}
/// Parse the `getPeers` response section: records keyed by IPv6 address.
/// The first record typically refers to the current node itself.
pub fn peers_from_response(value: &serde_json::Value) -> Result<Vec<PeerInfo>, NodeError> {
let obj = value.as_object().ok_or_else(|| {
NodeError::AdminProtocol("getPeers: response section is not an object".to_string())
})?;
let mut peers = Vec::with_capacity(obj.len());
for (address, record) in obj {
let address = YggAddr::new(address.parse::<Ipv6Addr>().map_err(|_| {
NodeError::AdminProtocol(format!("getPeers: `{address}` is not an IPv6 address"))
})?);
peers.push(PeerInfo {
address,
key: record
.get("key")
.and_then(serde_json::Value::as_str)
.map(str::to_string),
remote: record
.get("remote")
.and_then(serde_json::Value::as_str)
.map(str::to_string),
coords: record.get("coords").cloned(),
});
}
Ok(peers)
}
+126
View File
@@ -0,0 +1,126 @@
//! Integration tests for the Yggdrasil admin socket client, against a
//! fake admin server speaking the documented protocol.
#![forbid(unsafe_code)]
#![allow(clippy::unwrap_used, clippy::expect_used)]
use std::time::Duration;
use sn_node::error::NodeError;
use sn_node::yggdrasil::{peers_from_response, AdminClient, AdminEndpoint, SelfInfo};
const SELF_RESPONSE: &str = concat!(
r#"{"request": {"request": "getself"}, "status": "success", "response": {"#,
r#""200:1111:2222:3333:4444:5555:6666:7777": {"#,
r#""address": "200:1111:2222:3333:4444:5555:6666:7777","#,
r#""key": "0000aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa1111","#,
r#""build_name": "yggdrasil", "build_version": "0.5.12","#,
r#""subnet": "300:1111:2222:3333::/64", "coords": [1, 2, 3]}}}"#
);
const PEERS_RESPONSE: &str = concat!(
r#"{"request": {"request": "getpeers"}, "status": "success", "response": {"#,
r#""200:1111:2222:3333:4444:5555:6666:7777": {"coords": [1]}, "#,
r#""200:9999:2222:3333:4444:5555:6666:7777": {"#,
r#""coords": [1, 2], "port": 1, "key": "ffff...", "#,
r#""remote": "tcp://[fe80::1%25eth0]:1337"}}}"#
);
const ERROR_RESPONSE: &str =
r#"{"request": {"request": "getself"}, "status": "error", "error": "boom"}"#;
/// Fake admin server: answers each connection's request line, then closes.
async fn fake_admin_server(response: &'static str) -> std::net::SocketAddr {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
loop {
let Ok((mut socket, _)) = listener.accept().await else {
break;
};
let mut line = Vec::new();
let mut byte = [0u8; 1];
while !line.ends_with(b"\n") {
if socket.read(&mut byte).await.unwrap_or(0) == 0 {
break;
}
line.push(byte[0]);
}
socket.write_all(response.as_bytes()).await.ok();
socket.shutdown().await.ok();
}
});
addr
}
fn tcp_endpoint(addr: std::net::SocketAddr) -> AdminEndpoint {
AdminEndpoint::Tcp {
host: addr.ip().to_string(),
port: addr.port(),
}
}
#[tokio::test]
async fn get_self_parses_address_subnet_and_version() {
let endpoint = tcp_endpoint(fake_admin_server(SELF_RESPONSE).await);
let info = SelfInfo::from_response(
&AdminClient::connect(&endpoint, Duration::from_secs(2))
.await
.unwrap()
.request("getself")
.await
.unwrap(),
)
.unwrap();
assert_eq!(
info.address.inner().to_string(),
"200:1111:2222:3333:4444:5555:6666:7777"
);
assert_eq!(info.subnet.as_deref(), Some("300:1111:2222:3333::/64"));
assert_eq!(info.build_version.as_deref(), Some("0.5.12"));
}
#[tokio::test]
async fn get_peers_lists_remote_uris() {
let endpoint = tcp_endpoint(fake_admin_server(PEERS_RESPONSE).await);
let value = AdminClient::connect(&endpoint, Duration::from_secs(2))
.await
.unwrap()
.request("getpeers")
.await
.unwrap();
let peers = peers_from_response(&value).unwrap();
assert_eq!(peers.len(), 2);
assert_eq!(
peers[1].remote.as_deref(),
Some("tcp://[fe80::1%25eth0]:1337")
);
}
#[tokio::test]
async fn admin_error_is_surfaced() {
let endpoint = tcp_endpoint(fake_admin_server(ERROR_RESPONSE).await);
let mut client = AdminClient::connect(&endpoint, Duration::from_secs(2))
.await
.unwrap();
let err = client.request("getself").await.unwrap_err();
assert!(matches!(err, NodeError::AdminFailed(ref text) if text == "boom"));
}
#[tokio::test]
async fn connection_refused_becomes_io_or_timeout_error() {
// Port 1 on localhost: virtually guaranteed to be closed.
let endpoint = AdminEndpoint::Tcp {
host: "127.0.0.1".into(),
port: 1,
};
let result = AdminClient::get_self(&endpoint, Duration::from_millis(500)).await;
assert!(matches!(
result,
Err(NodeError::Io(_)) | Err(NodeError::Timeout)
));
}