Files
AWG2_harness/traffic_core/src/status.rs
T

637 lines
22 KiB
Rust

use crate::{config::Config, docker, firewall, netns::build_transport, swap, traffic, wg};
use std::{
collections::BTreeMap,
fs,
process::Command,
time::{SystemTime, UNIX_EPOCH},
};
const ONLINE_HANDSHAKE_THRESHOLD_SECS: u64 = 180;
pub fn print_status(cfg: &Config) -> anyhow::Result<()> {
println!("System:");
let svc = Command::new("systemctl")
.args(["is-active", "awg2-harness"])
.output();
match svc {
Ok(out) if out.status.success() => println!(" service: active"),
Ok(_) => println!(" service: inactive"),
Err(_) => println!(" service: unknown"),
}
let target = match docker::discover_target(cfg) {
Ok(t) => t,
Err(e) => {
println!(" container: not available ({e})");
println!(" container_id: n/a");
println!(" pid: n/a");
println!(" netns: n/a");
println!(" interface: n/a");
print_degraded_sections(cfg)?;
return Ok(());
}
};
println!(" container: {}", target.container_name);
println!(" container_id: {}", target.container_id);
println!(" pid: {}", target.container_pid);
println!(" netns: {}", target.netns_path);
println!(" interface: {}", target.interface_name);
let docker_path_diag = docker::collect_port_path_diagnostics(
&target.container_name,
cfg.host_firewall.amnezia_port,
);
let peers = wg::discover_peers(
target.container_pid,
&target.container_name,
&cfg.amnezia.peer_ipv4_cidr,
)?;
println!("Peers:");
println!(" total: {}", peers.len());
for p in peers {
println!(" - {}", p.ipv4);
}
println!("WireGuard clients:");
println!(
" IP State Latest handshake RX TX Public key Endpoint"
);
match wg::discover_peer_runtime(
target.container_pid,
&target.container_name,
&cfg.amnezia.peer_ipv4_cidr,
) {
Ok(runtime_peers) => {
let mut by_ip: BTreeMap<String, &wg::PeerRuntime> = BTreeMap::new();
for rp in &runtime_peers {
for ip in &rp.allowed_ips {
by_ip.insert(ip.clone(), rp);
}
}
let peers_sorted = wg::discover_peers(
target.container_pid,
&target.container_name,
&cfg.amnezia.peer_ipv4_cidr,
)?;
for p in peers_sorted {
let ip = p.ipv4.to_string();
if let Some(rp) = by_ip.get(&ip) {
let state = handshake_is_online(
rp.latest_handshake_epoch,
ONLINE_HANDSHAKE_THRESHOLD_SECS,
);
let hs = format_handshake_age(rp.latest_handshake_epoch);
let rx = format_bytes(rp.transfer_rx_bytes.unwrap_or(0));
let tx = format_bytes(rp.transfer_tx_bytes.unwrap_or(0));
let ep = rp.endpoint.clone().unwrap_or_else(|| "-".to_string());
println!(
" {:<11} {:<8} {:<19} {:<9} {:<9} {:<44} {}",
ip, state, hs, rx, tx, rp.public_key, ep
);
} else {
println!(
" {:<11} {:<8} {:<19} {:<9} {:<9} {:<44} -",
ip, "offline", "never", "0B", "0B", "-"
);
}
}
}
Err(e) => println!(" unavailable ({e})"),
}
println!("Traffic desired:");
println!(" peer_limit: {}", cfg.traffic.peer_limit);
println!(" root_rate: {}", cfg.traffic.root_rate);
println!(" ifb_name: {}", cfg.traffic.ifb_name);
println!(" class_start_id: {}", cfg.traffic.class_start_id);
println!("Traffic actual:");
match traffic::inspect_drift_summary(cfg) {
Ok(None) => println!(" drift: no"),
Ok(Some(reason)) => {
println!(" drift: yes");
println!(" reason: {}", reason);
}
Err(e) => println!(" drift: unknown ({e})"),
}
println!(" topology:");
let transport = build_transport(target.container_pid, &target.container_name);
let egress_qdisc = transport
.run(&["tc", "qdisc", "show", "dev", &target.interface_name])
.unwrap_or_default();
let ingress_qdisc = egress_qdisc.clone();
let ifb_qdisc = transport
.run(&["tc", "qdisc", "show", "dev", &cfg.traffic.ifb_name])
.unwrap_or_default();
println!(" egress_htb_present: {}", egress_qdisc.contains("htb"));
println!(" ingress_present: {}", ingress_qdisc.contains("ingress"));
println!(" ifb_htb_present: {}", ifb_qdisc.contains("htb"));
println!(" peer_table:");
match traffic::peer_status_table(cfg) {
Ok(Some(rows)) if rows.is_empty() => println!(" no peers discovered"),
Ok(Some(rows)) => {
for r in rows {
println!(
" ip={} class_id={} egress_filter={} ingress_filter={} egress={}/{} ingress={}/{} drift={}",
r.ip,
r.class_id,
r.egress_filter_present,
r.ingress_filter_present,
r.egress_rate.unwrap_or_else(|| "n/a".to_string()),
r.egress_ceil.unwrap_or_else(|| "n/a".to_string()),
r.ingress_rate.unwrap_or_else(|| "n/a".to_string()),
r.ingress_ceil.unwrap_or_else(|| "n/a".to_string()),
r.drift_reason.unwrap_or_else(|| "no".to_string())
);
}
}
Ok(None) => println!(" traffic disabled"),
Err(e) => println!(" unavailable ({e})"),
}
println!("Peer traffic counters:");
println!(" IP Egress hits Ingress hits");
match collect_peer_filter_hits(
&target.interface_name,
&cfg.traffic.ifb_name,
target.container_pid,
&target.container_name,
) {
Ok(map) if map.is_empty() => println!(" filter statistics: unavailable"),
Ok(map) => {
for (ip, (eg, ing)) in map {
println!(" {:<11} {:<13} {}", ip, eg, ing);
}
}
Err(_) => println!(" filter statistics: unavailable"),
}
println!("Host firewall:");
println!(" enabled: {}", cfg.host_firewall.enabled);
println!(" ssh_port: {}/tcp", cfg.host_firewall.ssh_port);
println!(
" amnezia_port: {}/{}",
cfg.host_firewall.amnezia_port,
cfg.host_firewall.amnezia_protocols.join(",")
);
println!(
" input_policy: {}",
cfg.host_firewall
.default_input_policy
.trim()
.to_ascii_lowercase()
);
if cfg.host_firewall.enabled {
match firewall::host_firewall_drift_reason(cfg) {
Ok(None) => println!(" drift: no"),
Ok(Some(reason)) => {
println!(" drift: yes");
println!(" reason: {}", reason);
}
Err(e) => println!(" drift: unknown ({e})"),
}
}
print_docker_amnezia_port_path(&docker_path_diag);
println!("Anti-torrent:");
println!(" enabled: {}", cfg.anti_torrent.enabled);
println!(
" max_connections_per_peer: {}",
cfg.anti_torrent.max_connections_per_peer
);
if cfg.anti_torrent.enabled {
match firewall::antitorrent_drift_reason(cfg) {
Ok(None) => println!(" drift: no"),
Ok(Some(reason)) => {
println!(" drift: yes");
println!(" reason: {}", reason);
}
Err(e) => println!(" drift: unknown ({e})"),
}
match firewall::antitorrent_drop_counter(cfg) {
Ok(Some((packets, bytes))) => {
println!(" dropped_packets: {}", packets);
println!(" dropped_bytes: {}", bytes);
}
Ok(None) => println!(" dropped_packets: n/a"),
Err(e) => println!(" dropped_packets: inspect error ({e})"),
}
}
print_swap_status(cfg);
let ct_count = Command::new("sysctl")
.args(["-n", "net.netfilter.nf_conntrack_count"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.unwrap_or_else(|| "n/a".to_string());
let ct_max = Command::new("sysctl")
.args(["-n", "net.netfilter.nf_conntrack_max"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.unwrap_or_else(|| "n/a".to_string());
println!("Conntrack:");
println!(" nf_conntrack_count: {}", ct_count);
println!(" nf_conntrack_max: {}", ct_max);
println!("Last actions:");
let last = fs::read_to_string("/run/awg2-harness/last_action.state").unwrap_or_default();
if last.trim().is_empty() {
println!(" no persisted state");
} else {
for line in last.lines() {
println!(" {}", line);
}
}
println!("Recent events:");
let events = fs::read_to_string("/run/awg2-harness/events.log").unwrap_or_default();
let lines: Vec<&str> = events.lines().collect();
if lines.is_empty() {
println!(" no recent events");
} else {
for l in lines.iter().rev().take(10).rev() {
println!(" {l}");
}
}
print_reboot_section(cfg);
Ok(())
}
fn print_degraded_sections(cfg: &Config) -> anyhow::Result<()> {
println!("Peers:");
println!(" unavailable (container/netns not discovered)");
println!("WireGuard clients:");
println!(" unavailable (container/netns not discovered)");
println!("Traffic desired:");
println!(" peer_limit: {}", cfg.traffic.peer_limit);
println!(" root_rate: {}", cfg.traffic.root_rate);
println!(" ifb_name: {}", cfg.traffic.ifb_name);
println!(" class_start_id: {}", cfg.traffic.class_start_id);
println!("Traffic actual:");
println!(" drift: unknown (container/netns not discovered)");
println!("Host firewall:");
println!(" enabled: {}", cfg.host_firewall.enabled);
println!(" ssh_port: {}/tcp", cfg.host_firewall.ssh_port);
println!(
" amnezia_port: {}/{}",
cfg.host_firewall.amnezia_port,
cfg.host_firewall.amnezia_protocols.join(",")
);
println!(
" input_policy: {}",
cfg.host_firewall
.default_input_policy
.trim()
.to_ascii_lowercase()
);
if cfg.host_firewall.enabled {
match firewall::host_firewall_drift_reason(cfg) {
Ok(None) => println!(" drift: no"),
Ok(Some(reason)) => {
println!(" drift: yes");
println!(" reason: {}", reason);
}
Err(e) => println!(" drift: unknown ({e})"),
}
}
println!("Docker/Amnezia port path:");
println!(" docker_userland_proxy: unknown");
println!(" published_udp_port: unknown");
println!(" host_forward_policy_drop: unknown");
println!(" docker_forward_rules_present: unknown");
println!(" note: FORWARD is Docker-managed; awg2-harness does not modify host FORWARD");
println!("Anti-torrent:");
println!(" enabled: {}", cfg.anti_torrent.enabled);
println!(
" max_connections_per_peer: {}",
cfg.anti_torrent.max_connections_per_peer
);
if cfg.anti_torrent.enabled {
match firewall::antitorrent_drift_reason(cfg) {
Ok(None) => println!(" drift: no"),
Ok(Some(reason)) => {
println!(" drift: yes");
println!(" reason: {}", reason);
}
Err(e) => println!(" drift: unknown ({e})"),
}
match firewall::antitorrent_drop_counter(cfg) {
Ok(Some((packets, bytes))) => {
println!(" dropped_packets: {}", packets);
println!(" dropped_bytes: {}", bytes);
}
Ok(None) => println!(" dropped_packets: n/a"),
Err(e) => println!(" dropped_packets: inspect error ({e})"),
}
}
print_swap_status(cfg);
println!("Conntrack:");
let ct_count = Command::new("sysctl")
.args(["-n", "net.netfilter.nf_conntrack_count"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.unwrap_or_else(|| "n/a".to_string());
let ct_max = Command::new("sysctl")
.args(["-n", "net.netfilter.nf_conntrack_max"])
.output()
.ok()
.map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
.unwrap_or_else(|| "n/a".to_string());
println!(" nf_conntrack_count: {}", ct_count);
println!(" nf_conntrack_max: {}", ct_max);
println!("Last actions:");
let last = fs::read_to_string("/run/awg2-harness/last_action.state").unwrap_or_default();
if last.trim().is_empty() {
println!(" no persisted state");
} else {
for line in last.lines() {
println!(" {}", line);
}
}
println!("Recent events:");
let events = fs::read_to_string("/run/awg2-harness/events.log").unwrap_or_default();
let lines: Vec<&str> = events.lines().collect();
if lines.is_empty() {
println!(" no recent events");
} else {
for l in lines.iter().rev().take(10).rev() {
println!(" {l}");
}
}
print_reboot_section(cfg);
Ok(())
}
fn print_reboot_section(cfg: &Config) {
println!("Reboot scheduler:");
println!(" enabled: {}", cfg.reboot.enabled);
println!(" window_start_utc: {}", cfg.reboot.window_start_utc);
println!(" window_end_utc: {}", cfg.reboot.window_end_utc);
let svc = Command::new("systemctl")
.args(["is-active", "awg2-reboot-scheduler"])
.output();
match svc {
Ok(out) if out.status.success() => println!(" service: active"),
Ok(_) => println!(" service: inactive"),
Err(_) => println!(" service: unknown"),
}
let plan = fs::read_to_string("/run/awg2-harness/reboot.plan").unwrap_or_default();
if plan.trim().is_empty() {
println!(" planned: n/a");
} else {
for line in plan.lines() {
println!(" {}", line);
}
}
}
fn print_swap_status(cfg: &Config) {
println!("Swap:");
match swap::inspect_swap(cfg) {
Ok(s) => {
println!(" enabled: {}", s.enabled);
println!(" desired_path: {}", s.desired_path.display());
println!(" desired_size_mb: {}", cfg.swap.size_mb);
println!(" desired_priority: {}", cfg.swap.priority);
println!(" active_entries: {}", s.active_entries.len());
println!(
" desired_active: {}",
if s.desired_active { "yes" } else { "no" }
);
println!(" exclusive: {}", if s.exclusive { "yes" } else { "no" });
println!(
" fstab_exclusive: {}",
if s.fstab_exclusive { "yes" } else { "no" }
);
println!(" fstab_swap_entries: {}", s.fstab_swap_entries.len());
println!(
" swappiness_current: {}",
s.current_swappiness
.map(|v| v.to_string())
.unwrap_or_else(|| "n/a".to_string())
);
println!(
" swappiness_persisted: {}",
s.persisted_swappiness
.map(|v| v.to_string())
.unwrap_or_else(|| "n/a".to_string())
);
if let Some(reason) = s.drift_reason {
println!(" drift: yes");
println!(" drift_reason: {}", reason);
} else {
println!(" drift: no");
}
}
Err(e) => {
println!(" enabled: {}", cfg.swap.enabled);
println!(" desired_path: {}", cfg.swap.path);
println!(" drift: unknown ({e})");
}
}
}
fn print_docker_amnezia_port_path(diag: &docker::PortPathDiagnostics) {
println!("Docker/Amnezia port path:");
println!(
" docker_userland_proxy: {}",
diag.docker_userland_proxy.as_str()
);
match &diag.published_udp_port {
Some(p) => println!(
" published_udp_port: {} -> {}:{}",
p.host_port, p.container_ip, p.container_port
),
None => println!(" published_udp_port: unknown"),
}
println!(
" host_forward_policy_drop: {}",
diag.host_forward_policy_drop.as_str()
);
println!(
" docker_forward_rules_present: {}",
diag.docker_forward_rules_present.as_str()
);
println!(" note: FORWARD is Docker-managed; awg2-harness does not modify host FORWARD");
if let Some(w) = diag.warning() {
println!(" warning: {w}");
}
}
fn handshake_is_online(epoch: Option<u64>, threshold_secs: u64) -> &'static str {
if let Some(ts) = epoch {
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
if now.saturating_sub(ts) <= threshold_secs {
return "online";
}
}
"offline"
}
fn format_handshake_age(epoch: Option<u64>) -> String {
let Some(ts) = epoch else {
return "never".to_string();
};
let now = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let age = now.saturating_sub(ts);
format!("{}s ago", age)
}
fn format_bytes(v: u64) -> String {
const KB: f64 = 1024.0;
const MB: f64 = KB * 1024.0;
const GB: f64 = MB * 1024.0;
let b = v as f64;
if b >= GB {
format!("{:.1}GB", b / GB)
} else if b >= MB {
format!("{:.1}MB", b / MB)
} else if b >= KB {
format!("{:.1}KB", b / KB)
} else {
format!("{}B", v)
}
}
fn collect_peer_filter_hits(
iface: &str,
ifb: &str,
pid: u32,
container_name: &str,
) -> anyhow::Result<BTreeMap<String, (u64, u64)>> {
let transport = build_transport(pid, container_name);
let egress = transport.run(&["tc", "-s", "filter", "show", "dev", iface, "parent", "1:"])?;
let ingress = transport.run(&["tc", "-s", "filter", "show", "dev", ifb, "parent", "2:"])?;
let egress_hits = parse_tc_filter_hits(&egress);
let ingress_hits = parse_tc_filter_hits(&ingress);
let mut out: BTreeMap<String, (u64, u64)> = BTreeMap::new();
for (ip, c) in egress_hits {
out.entry(ip).or_insert((0, 0)).0 = c;
}
for (ip, c) in ingress_hits {
out.entry(ip).or_insert((0, 0)).1 = c;
}
Ok(out)
}
fn parse_tc_filter_hits(raw: &str) -> BTreeMap<String, u64> {
let mut out = BTreeMap::new();
let mut current_ip: Option<String> = None;
for line in raw.lines() {
let t = line.trim();
if t.contains("match ") {
current_ip = extract_tc_match_ip(t);
}
if t.starts_with("Sent ") {
if let Some(ip) = current_ip.clone() {
let hits = parse_sent_packets(t).unwrap_or(0);
out.insert(ip, hits);
}
} else if (t.contains(" success ") || t.starts_with("success "))
&& let Some(ip) = current_ip.clone()
{
let hits = parse_success_packets(t).unwrap_or(0);
out.insert(ip, hits);
}
}
out
}
fn parse_sent_packets(line: &str) -> Option<u64> {
let toks: Vec<&str> = line.split_whitespace().collect();
if toks.len() >= 5 && toks[0] == "Sent" && toks[2] == "bytes" && toks[4] == "pkt" {
return toks[3].parse::<u64>().ok();
}
None
}
fn parse_success_packets(line: &str) -> Option<u64> {
let toks: Vec<&str> = line.split_whitespace().collect();
let pos = toks.iter().position(|v| *v == "success")?;
toks.get(pos + 1)?.parse::<u64>().ok()
}
fn extract_tc_match_ip(line: &str) -> Option<String> {
let toks: Vec<&str> = line.split_whitespace().collect();
let pos = toks.iter().position(|v| *v == "match")?;
let match_val = *toks.get(pos + 1)?;
let (hex, mask) = match_val.split_once('/')?;
if !mask.eq_ignore_ascii_case("ffffffff") || hex.len() != 8 {
return None;
}
let b1 = u8::from_str_radix(&hex[0..2], 16).ok()?;
let b2 = u8::from_str_radix(&hex[2..4], 16).ok()?;
let b3 = u8::from_str_radix(&hex[4..6], 16).ok()?;
let b4 = u8::from_str_radix(&hex[6..8], 16).ok()?;
Some(format!("{b1}.{b2}.{b3}.{b4}"))
}
#[cfg(test)]
mod tests {
use super::{parse_sent_packets, parse_success_packets, parse_tc_filter_hits};
#[test]
fn parse_sent_pkt_counter() {
let line = "Sent 12345 bytes 67 pkt (dropped 0, overlimits 0 requeues 0)";
assert_eq!(parse_sent_packets(line), Some(67));
}
#[test]
fn parse_success_counter() {
let line = "match 0a080102/ffffffff at 16 success 42";
assert_eq!(parse_success_packets(line), Some(42));
}
#[test]
fn parse_tc_filter_hits_supports_sent_and_success() {
let raw = r#"
filter parent 1: protocol ip pref 1 u32 chain 0
filter parent 1: protocol ip pref 1 u32 chain 0 fh 800: ht divisor 1
filter parent 1: protocol ip pref 1 u32 chain 0 fh 800::800 order 2048 key ht 800 bkt 0 flowid 1:10 not_in_hw
match 0a080102/ffffffff at 16
Sent 100 bytes 7 pkt (dropped 0, overlimits 0 requeues 0)
filter parent 1: protocol ip pref 1 u32 chain 0 fh 800::801 order 2049 key ht 800 bkt 0 flowid 1:11 not_in_hw
match 0a080103/ffffffff at 16 success 9
"#;
let map = parse_tc_filter_hits(raw);
assert_eq!(map.get("10.8.1.2"), Some(&7));
assert_eq!(map.get("10.8.1.3"), Some(&9));
}
}