#![cfg_attr(test, allow(unused_crate_dependencies))]
use ed25519_dalek::VerifyingKey;
use eyre::Report;
use eyre::eyre;
use hex::decode as hex_decode;
use mac_address::get_mac_address;
use quinn::{ClientConfig, Endpoint};
use rustls::{Certificate, RootCertStore, ServerName};
use sha2::{Digest, Sha256};
use std::convert::TryFrom;
use std::fs;
use std::net::ToSocketAddrs;
use std::path::PathBuf;
use std::sync::Arc;
use tokio::net::TcpStream;
use tokio_rustls::TlsConnector;
use volli_core::{DEFAULT_QUIC_PORT, DEFAULT_TCP_PORT, Message};
use volli_transport::{QuicTransport, TcpTransport, Transport};
#[derive(Clone, Debug)]
pub enum Protocol {
Quic,
Tcp,
}
#[derive(Debug, Clone)]
pub enum Role {
Agent,
Coordinator,
}
#[derive(Clone, Debug)]
pub struct AgentConfig {
pub host: String,
pub quic_port: u16,
pub tcp_port: u16,
pub protocol: Option<Protocol>,
pub token: String,
pub fingerprint: String,
pub cert: Vec<u8>,
pub role: Role,
}
impl Default for AgentConfig {
fn default() -> Self {
Self {
host: "127.0.0.1".into(),
quic_port: DEFAULT_QUIC_PORT,
tcp_port: DEFAULT_TCP_PORT,
protocol: None,
token: String::new(),
fingerprint: String::new(),
cert: Vec::new(),
role: Role::Agent,
}
}
}
fn default_agent_dir() -> PathBuf {
let mut base = volli_core::config_dir();
base.push("profiles");
base.push("agent");
base
}
pub fn agent_dir(profile: Option<&str>) -> PathBuf {
let mut dir = default_agent_dir();
if let Some(p) = profile {
dir.push(p);
}
dir
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
pub struct PeerEntry {
pub host: String,
pub quic_port: u16,
pub tcp_port: u16,
#[serde(default)]
pub last_ok: Option<u64>,
#[serde(default)]
pub last_fail: Option<u64>,
}
pub fn load_peers(profile: &str) -> Result<Vec<PeerEntry>, Report> {
let path = agent_dir(Some(profile)).join("peers.json");
match fs::read_to_string(path) {
Ok(s) => Ok(serde_json::from_str(&s)?),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Vec::new()),
Err(e) => Err(e.into()),
}
}
pub fn save_peers(profile: &str, peers: &[PeerEntry]) -> Result<(), Report> {
let dir = agent_dir(Some(profile));
fs::create_dir_all(&dir)?;
fs::write(dir.join("peers.json"), serde_json::to_string(peers)?)?;
Ok(())
}
pub fn add_peer(profile: &str, peer: PeerEntry) -> Result<(), Report> {
let mut peers = load_peers(profile)?;
if !peers.iter().any(|p| {
p.host == peer.host && p.tcp_port == peer.tcp_port && p.quic_port == peer.quic_port
}) {
peers.push(peer);
save_peers(profile, &peers)?;
}
Ok(())
}
fn now_secs() -> u64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs()
}
pub fn save_state(profile: &str, secret: &str) -> Result<(), Report> {
let dir = agent_dir(Some(profile));
fs::create_dir_all(&dir)?;
fs::write(dir.join("agent_state"), secret)?;
Ok(())
}
pub fn load_state(profile: &str) -> Result<Option<String>, Report> {
let path = agent_dir(Some(profile)).join("agent_state");
match fs::read_to_string(path) {
Ok(s) => Ok(Some(s)),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
Err(e) => Err(e.into()),
}
}
pub fn list_profiles() -> Result<Vec<String>, Report> {
let base = default_agent_dir();
let mut profiles = Vec::new();
if base.exists() {
for entry in fs::read_dir(base)? {
let entry = entry?;
if entry.path().is_dir() {
if let Some(name) = entry.file_name().to_str() {
profiles.push(name.to_string());
}
}
}
}
profiles.sort();
Ok(profiles)
}
pub fn delete_profile(profile: &str) -> Result<(), Report> {
let dir = agent_dir(Some(profile));
if dir.exists() {
fs::remove_dir_all(dir)?;
}
Ok(())
}
pub fn profile_exists(profile: &str) -> bool {
agent_dir(Some(profile)).exists()
}
pub fn rename_profile(old: &str, new: &str) -> Result<(), Report> {
let src = agent_dir(Some(old));
let dst = agent_dir(Some(new));
if !src.exists() {
return Err(eyre!("profile not found"));
}
if dst.exists() {
return Err(eyre!("profile exists"));
}
fs::create_dir_all(dst.parent().unwrap())?;
fs::rename(src, dst)?;
Ok(())
}
#[derive(serde::Serialize, serde::Deserialize)]
pub struct AgentProfileExport {
pub name: String,
pub secret: String,
#[serde(default)]
pub peers: Vec<PeerEntry>,
}
pub fn export_profile(profile: &str) -> Result<String, Report> {
let secret = load_state(profile)?.ok_or_else(|| eyre!("profile not found"))?;
let peers = load_peers(profile).unwrap_or_default();
let exp = AgentProfileExport {
name: profile.to_string(),
secret,
peers,
};
Ok(serde_yaml::to_string(&exp)?)
}
pub fn import_profile(yaml: &str, name: Option<&str>, force: bool) -> Result<String, Report> {
let mut exp: AgentProfileExport = serde_yaml::from_str(yaml)?;
if let Some(n) = name {
exp.name = n.to_string();
}
if profile_exists(&exp.name) && !force {
return Err(eyre!("profile exists"));
}
save_state(&exp.name, &exp.secret)?;
if !exp.peers.is_empty() {
save_peers(&exp.name, &exp.peers)?;
}
Ok(exp.name)
}
fn configure_client(cert: &[u8], alpn: &str) -> Result<ClientConfig, Report> {
let mut roots = rustls::RootCertStore::empty();
roots.add(&Certificate(cert.to_vec()))?;
let mut crypto = rustls::ClientConfig::builder()
.with_safe_defaults()
.with_root_certificates(roots)
.with_no_client_auth();
crypto.alpn_protocols = vec![alpn.as_bytes().to_vec()];
Ok(ClientConfig::new(Arc::new(crypto)))
}
pub fn apply_secret(cfg: &mut AgentConfig, profile: &str, secret: &str) -> Result<(), Report> {
let bs = volli_core::BootstrapSecret::decode(secret)?;
let fp = hex::encode(Sha256::digest(&bs.cert));
cfg.host = bs.host.clone();
cfg.quic_port = bs.quic_port;
cfg.tcp_port = bs.tcp_port;
cfg.token = volli_core::token::encode_token(&bs.token)?;
cfg.fingerprint = fp;
cfg.cert = bs.cert;
let peer = PeerEntry {
host: cfg.host.clone(),
quic_port: cfg.quic_port,
tcp_port: cfg.tcp_port,
last_ok: None,
last_fail: None,
};
let _ = add_peer(profile, peer);
Ok(())
}
pub async fn run(
mut config: AgentConfig,
profile: String,
on_first_connect: Option<Box<dyn FnOnce() + Send>>,
) -> Result<(), Report> {
let proto_pref = config.protocol.take();
let mut backoff = 1u64;
let mut connect_cb = on_first_connect;
let mut peers = load_peers(&profile).unwrap_or_default();
if peers.is_empty() {
peers.push(PeerEntry {
host: config.host.clone(),
quic_port: config.quic_port,
tcp_port: config.tcp_port,
last_ok: None,
last_fail: None,
});
save_peers(&profile, &peers).ok();
}
let mut idx = 0usize;
loop {
let peer = peers.get(idx).cloned().unwrap();
config.host = peer.host.clone();
config.quic_port = peer.quic_port;
config.tcp_port = peer.tcp_port;
let res = match proto_pref.as_ref().unwrap_or(&Protocol::Quic) {
Protocol::Quic => match connect_quic(&config).await {
Ok((tr, p)) => {
peers[idx].last_ok = Some(now_secs());
save_peers(&profile, &peers).ok();
handle_agent(tr, &config, &p, &mut connect_cb).await
}
Err(e) => {
tracing::warn!("quic connect error: {}", e);
match connect_tcp(&config).await {
Ok((tr, p)) => {
peers[idx].last_ok = Some(now_secs());
save_peers(&profile, &peers).ok();
handle_agent(tr, &config, &p, &mut connect_cb).await
}
Err(e) => Err(e),
}
}
},
Protocol::Tcp => match connect_tcp(&config).await {
Ok((tr, p)) => {
peers[idx].last_ok = Some(now_secs());
save_peers(&profile, &peers).ok();
handle_agent(tr, &config, &p, &mut connect_cb).await
}
Err(e) => Err(e),
},
};
match res {
Ok(_) => {
backoff = 1;
idx = 0;
}
Err(e) => {
tracing::error!("connection error: {}", e);
peers[idx].last_fail = Some(now_secs());
save_peers(&profile, &peers).ok();
backoff = (backoff * 2).min(32);
idx = (idx + 1) % peers.len();
}
}
tokio::time::sleep(std::time::Duration::from_secs(backoff)).await;
}
}
async fn handle_agent(
mut transport: Box<dyn Transport>,
cfg: &AgentConfig,
peer: &str,
on_first_connect: &mut Option<Box<dyn FnOnce() + Send>>,
) -> Result<(), Report> {
transport
.send(&Message::Auth {
token: cfg.token.clone(),
})
.await?;
let mut authed = false;
while let Some(msg) = transport.recv().await? {
match msg {
Message::AuthOk => {
authed = true;
}
Message::Hello {
coord_id,
nonce,
sig,
} => {
if !authed {
return Err(eyre!("handshake before auth"));
}
let pk_bytes = hex_decode(&coord_id)?;
let arr: [u8; 32] = pk_bytes
.as_slice()
.try_into()
.map_err(|_| eyre!("bad coord id"))?;
let pk = VerifyingKey::from_bytes(&arr)?;
volli_core::handshake::verify_nonce(&pk, &nonce, &sig)?;
transport
.send(&Message::Welcome {
coord_id,
nonce,
sig,
})
.await?;
tracing::info!(target: "connection", %peer, role=?cfg.role, "authenticated");
if let Some(cb) = on_first_connect.take() {
cb();
}
}
Message::AuthErr => return Err(eyre!("authentication failed")),
Message::Ping => {
tracing::info!(target: "connection", %peer, "received ping");
let mac = get_mac_address()
.ok()
.flatten()
.map(|m| m.to_string())
.unwrap_or_default();
tracing::info!(target: "connection", %peer, "sending pong");
transport.send(&Message::Pong { mac }).await?;
}
_ => {}
}
}
Ok(())
}
pub async fn connect_tcp(cfg: &AgentConfig) -> Result<(Box<dyn Transport>, String), Report> {
let addr = format!("{}:{}", cfg.host, cfg.tcp_port);
let mut addrs = addr.to_socket_addrs()?;
let addr = addrs
.find(|a| a.is_ipv4())
.or_else(|| addrs.next())
.ok_or_else(|| eyre!("invalid addr"))?;
let stream = TcpStream::connect(addr).await?;
let alpn = match cfg.role {
Role::Agent => "volli/agent",
Role::Coordinator => "volli/coord",
};
let mut roots = RootCertStore::empty();
roots.add(&Certificate(cfg.cert.clone()))?;
let mut root = rustls::ClientConfig::builder()
.with_safe_defaults()
.with_root_certificates(roots)
.with_no_client_auth();
root.alpn_protocols = vec![alpn.as_bytes().to_vec()];
let connector = TlsConnector::from(Arc::new(root));
let tls = connector
.connect(ServerName::try_from("volli")?, stream)
.await?;
if let Some(certs) = tls.get_ref().1.peer_certificates() {
if let Some(cert) = certs.first() {
let hash = Sha256::digest(&cert.0);
if hex::encode(hash) != cfg.fingerprint {
return Err(eyre!("server fingerprint mismatch"));
}
}
}
let peer = tls.get_ref().0.peer_addr()?.to_string();
Ok((Box::new(TcpTransport::new(tls)), peer))
}
pub async fn connect_quic(cfg: &AgentConfig) -> Result<(Box<dyn Transport>, String), Report> {
let addr = format!("{}:{}", cfg.host, cfg.quic_port);
let mut addrs = addr.to_socket_addrs()?;
let addr = addrs
.find(|a| a.is_ipv4())
.or_else(|| addrs.next())
.ok_or_else(|| eyre!("invalid addr"))?;
let mut endpoint = Endpoint::client("0.0.0.0:0".parse()?)?;
let alpn = match cfg.role {
Role::Agent => "volli/agent",
Role::Coordinator => "volli/coord",
};
let quinn_cfg = configure_client(&cfg.cert, alpn)?;
endpoint.set_default_client_config(quinn_cfg);
let connection = endpoint.connect(addr, "volli")?.await?;
if let Some(identity) = connection.peer_identity() {
if let Ok(certs) = identity.downcast::<Vec<Certificate>>() {
if let Some(cert) = certs.first() {
let hash = Sha256::digest(&cert.0);
if hex::encode(hash) != cfg.fingerprint {
return Err(eyre!("server fingerprint mismatch"));
}
}
}
}
let peer = connection.remote_address().to_string();
let (send, recv) = connection.open_bi().await?;
Ok((Box::new(QuicTransport::new(send, recv)), peer))
}