diff --git a/traffic_core/src/traffic.rs b/traffic_core/src/traffic.rs deleted file mode 100644 index a00ba3c..0000000 --- a/traffic_core/src/traffic.rs +++ /dev/null @@ -1,1159 +0,0 @@ -use crate::{ - config::Config, - docker, - lock::FileLock, - netns::{NsTransport, build_transport}, - wg, -}; - -use std::{ - collections::{BTreeMap, BTreeSet}, - net::Ipv4Addr, - time::Duration, -}; - -#[derive(Debug, Clone)] -struct TrafficDrift { - has_drift: bool, - reason: String, -} - -#[derive(Debug, Clone)] -struct FilterInspect { - map: BTreeMap, - duplicate_ips: BTreeSet, - duplicate_classes: BTreeSet, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -struct ClassRateCeil { - rate: String, - ceil: String, -} - -#[derive(Debug, Clone)] -pub struct PeerTrafficStatus { - pub ip: String, - pub class_id: u16, - pub egress_filter_present: bool, - pub ingress_filter_present: bool, - pub egress_rate: Option, - pub egress_ceil: Option, - pub ingress_rate: Option, - pub ingress_ceil: Option, - pub drift_reason: Option, -} - -pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result> { - if !cfg.traffic.enabled { - return Ok(None); - } - - let target = docker::discover_target(cfg)?; - let transport = build_transport(target.container_pid, &target.container_name); - let peers = wg::discover_peers(cfg, &target)?; - - if peers.is_empty() { - return Ok(Some("no peers discovered".to_string())); - } - - let drift = inspect_and_diff( - transport.as_ref(), - &target.interface_name, - &cfg.traffic.ifb_name, - cfg.traffic.class_start_id, - &cfg.traffic.root_rate, - &cfg.traffic.peer_limit, - &peers, - )?; - - if drift.has_drift { - Ok(Some(drift.reason)) - } else { - Ok(None) - } -} - -pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { - if !cfg.traffic.enabled { - return Ok(()); - } - - let _lock = - FileLock::acquire_timeout("/run/awg2-harness/traffic.lock", Duration::from_secs(10))?; - let target = docker::discover_target(cfg)?; - let transport = build_transport(target.container_pid, &target.container_name); - let peers = wg::discover_peers(cfg, &target)?; - - if peers.is_empty() { - println!("no peers found, skip tc apply"); - return Ok(()); - } - - let drift = inspect_and_diff( - transport.as_ref(), - &target.interface_name, - &cfg.traffic.ifb_name, - cfg.traffic.class_start_id, - &cfg.traffic.root_rate, - &cfg.traffic.peer_limit, - &peers, - )?; - - if !drift.has_drift { - println!("traffic rules up-to-date, skip apply"); - return Ok(()); - } - - println!("traffic drift detected: {}", drift.reason); - - teardown_owned_topology( - transport.as_ref(), - &target.interface_name, - &cfg.traffic.ifb_name, - )?; - apply_full_topology( - cfg, - transport.as_ref(), - &target.interface_name, - &cfg.traffic.ifb_name, - &peers, - )?; - - Ok(()) -} - -pub fn peer_status_table(cfg: &Config) -> anyhow::Result>> { - if !cfg.traffic.enabled { - return Ok(None); - } - - let target = docker::discover_target(cfg)?; - let transport = build_transport(target.container_pid, &target.container_name); - let peers = wg::discover_peers(cfg, &target)?; - if peers.is_empty() { - return Ok(Some(Vec::new())); - } - - let filters_egress = transport - .run(&[ - "tc", - "filter", - "show", - "dev", - &target.interface_name, - "parent", - "1:", - ]) - .unwrap_or_default(); - let filters_ingress = transport - .run(&[ - "tc", - "filter", - "show", - "dev", - &cfg.traffic.ifb_name, - "parent", - "2:", - ]) - .unwrap_or_default(); - let classes_egress_full = transport - .run(&["tc", "class", "show", "dev", &target.interface_name]) - .unwrap_or_default(); - let classes_ingress_full = transport - .run(&["tc", "class", "show", "dev", &cfg.traffic.ifb_name]) - .unwrap_or_default(); - - let inspect_egress = extract_filter_ip_class_map(&filters_egress); - let inspect_ingress = extract_filter_ip_class_map(&filters_ingress); - let actual_map_egress = inspect_egress.map; - let actual_map_ingress = inspect_ingress.map; - let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:"); - let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:"); - let expected_map = expected_peer_class_map(&peers, cfg.traffic.class_start_id)?; - let expected_limit = normalize_tc_rate(&cfg.traffic.peer_limit); - - let mut out = Vec::with_capacity(peers.len()); - for p in peers { - let ip = p.ipv4.to_string(); - let class_id = *expected_map.get(&ip).unwrap_or(&cfg.traffic.class_start_id); - let class_egress = format!("1:{class_id}"); - let class_ingress = format!("2:{class_id}"); - - let egress_filter_present = actual_map_egress.get(&ip).copied() == Some(class_id); - let ingress_filter_present = actual_map_ingress.get(&ip).copied() == Some(class_id); - let e = egress_rates.get(&class_egress); - let i = ingress_rates.get(&class_ingress); - - let mut reasons = Vec::new(); - if !egress_filter_present { - reasons.push("missing/mismatched egress filter".to_string()); - } - if !ingress_filter_present { - reasons.push("missing/mismatched ingress filter".to_string()); - } - if e.is_none() { - reasons.push("missing egress class".to_string()); - } - if i.is_none() { - reasons.push("missing ingress class".to_string()); - } - if let Some(v) = e - && !rate_ceil_matches_limit(v, &expected_limit) - { - reasons.push("egress rate/ceil mismatch".to_string()); - } - if let Some(v) = i - && !rate_ceil_matches_limit(v, &expected_limit) - { - reasons.push("ingress rate/ceil mismatch".to_string()); - } - if inspect_egress.duplicate_ips.contains(&ip) { - reasons.push("duplicate egress filters".to_string()); - } - if inspect_ingress.duplicate_ips.contains(&ip) { - reasons.push("duplicate ingress filters".to_string()); - } - - out.push(PeerTrafficStatus { - ip, - class_id, - egress_filter_present, - ingress_filter_present, - egress_rate: e.map(|v| v.rate.clone()), - egress_ceil: e.map(|v| v.ceil.clone()), - ingress_rate: i.map(|v| v.rate.clone()), - ingress_ceil: i.map(|v| v.ceil.clone()), - drift_reason: if reasons.is_empty() { - None - } else { - Some(reasons.join(", ")) - }, - }); - } - - Ok(Some(out)) -} - -fn inspect_and_diff( - transport: &dyn NsTransport, - iface: &str, - ifb: &str, - class_start_id: u16, - root_rate: &str, - peer_limit: &str, - peers: &[wg::Peer], -) -> anyhow::Result { - let root_qdisc = inspect_cmd( - transport, - &["tc", "qdisc", "show", "dev", iface], - "egress qdisc", - )?; - if !root_qdisc.contains("htb") { - return Ok(TrafficDrift { - has_drift: true, - reason: "missing egress root htb qdisc".to_string(), - }); - } - - let ingress_qdisc = inspect_cmd( - transport, - &["tc", "qdisc", "show", "dev", iface], - "ingress qdisc", - )?; - if !ingress_qdisc.contains("ingress") { - return Ok(TrafficDrift { - has_drift: true, - reason: "missing ingress qdisc".to_string(), - }); - } - - let ingress_redirect = inspect_cmd( - transport, - &["tc", "filter", "show", "dev", iface, "parent", "ffff:"], - "ingress redirect filter", - )?; - if !has_ingress_redirect_to_ifb(&ingress_redirect, ifb) { - return Ok(TrafficDrift { - has_drift: true, - reason: format!("missing ingress redirect to {ifb}"), - }); - } - - let ifb_link = match transport.run(&["ip", "link", "show", ifb]) { - Ok(out) => out, - Err(e) if is_not_found_error(&e.to_string()) => { - return Ok(TrafficDrift { - has_drift: true, - reason: "missing ifb interface".to_string(), - }); - } - Err(e) => return Err(anyhow::anyhow!("ifb link inspect failed: {e}")), - }; - if ifb_link.trim().is_empty() { - return Ok(TrafficDrift { - has_drift: true, - reason: "missing ifb interface".to_string(), - }); - } - if !link_is_up(&ifb_link) { - return Ok(TrafficDrift { - has_drift: true, - reason: format!("{ifb} exists but is not UP"), - }); - } - - let filters_egress = inspect_cmd( - transport, - &["tc", "filter", "show", "dev", iface, "parent", "1:"], - "egress filters", - )?; - let filters_ingress = inspect_cmd( - transport, - &["tc", "filter", "show", "dev", ifb, "parent", "2:"], - "ingress filters", - )?; - - let inspect_egress = extract_filter_ip_class_map(&filters_egress); - let inspect_ingress = extract_filter_ip_class_map(&filters_ingress); - if !inspect_egress.duplicate_ips.is_empty() { - return Ok(TrafficDrift { - has_drift: true, - reason: format!( - "duplicate egress filters detected for {} peer(s)", - inspect_egress.duplicate_ips.len() - ), - }); - } - if !inspect_egress.duplicate_classes.is_empty() { - return Ok(TrafficDrift { - has_drift: true, - reason: "duplicate egress class flowid mapping detected".to_string(), - }); - } - if !inspect_ingress.duplicate_ips.is_empty() { - return Ok(TrafficDrift { - has_drift: true, - reason: format!( - "duplicate ingress filters detected for {} peer(s)", - inspect_ingress.duplicate_ips.len() - ), - }); - } - if !inspect_ingress.duplicate_classes.is_empty() { - return Ok(TrafficDrift { - has_drift: true, - reason: "duplicate ingress class flowid mapping detected".to_string(), - }); - } - let actual_map_egress = inspect_egress.map; - let actual_map_ingress = inspect_ingress.map; - let actual_ips_egress: BTreeSet = actual_map_egress.keys().cloned().collect(); - let actual_ips_ingress: BTreeSet = actual_map_ingress.keys().cloned().collect(); - let expected_ips: BTreeSet = peers.iter().map(|p| p.ipv4.to_string()).collect(); - - if actual_ips_egress != expected_ips { - return Ok(TrafficDrift { - has_drift: true, - reason: format!( - "egress peer filter mismatch (actual={}, expected={})", - actual_ips_egress.len(), - expected_ips.len() - ), - }); - } - - if actual_ips_ingress != expected_ips { - return Ok(TrafficDrift { - has_drift: true, - reason: format!( - "ingress peer filter mismatch (actual={}, expected={})", - actual_ips_ingress.len(), - expected_ips.len() - ), - }); - } - - let expected_map = expected_peer_class_map(peers, class_start_id)?; - if actual_map_egress != expected_map { - return Ok(TrafficDrift { - has_drift: true, - reason: format!( - "egress class mapping mismatch (actual={}, expected={})", - actual_map_egress.len(), - expected_map.len() - ), - }); - } - if actual_map_ingress != expected_map { - return Ok(TrafficDrift { - has_drift: true, - reason: format!( - "ingress class mapping mismatch (actual={}, expected={})", - actual_map_ingress.len(), - expected_map.len() - ), - }); - } - - let expected_limit = normalize_tc_rate(peer_limit); - let classes_egress_full = inspect_cmd( - transport, - &["tc", "class", "show", "dev", iface], - "egress classes", - )?; - let classes_ingress_full = inspect_cmd( - transport, - &["tc", "class", "show", "dev", ifb], - "ingress classes", - )?; - let expected_root = normalize_tc_rate(root_rate); - let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:"); - let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:"); - - let root_egress = parse_root_class_rate_ceil(&classes_egress_full, "1:1"); - let root_ingress = parse_root_class_rate_ceil(&classes_ingress_full, "2:1"); - if root_egress - .as_ref() - .map(|v| !rate_ceil_matches_limit(v, &expected_root)) - .unwrap_or(true) - { - return Ok(TrafficDrift { - has_drift: true, - reason: "egress root class 1:1 rate/ceil mismatch".to_string(), - }); - } - if root_ingress - .as_ref() - .map(|v| !rate_ceil_matches_limit(v, &expected_root)) - .unwrap_or(true) - { - return Ok(TrafficDrift { - has_drift: true, - reason: "ingress root class 2:1 rate/ceil mismatch".to_string(), - }); - } - - for class_id in expected_map.values() { - let class_egress = format!("1:{class_id}"); - let class_ingress = format!("2:{class_id}"); - let e = egress_rates.get(&class_egress); - let i = ingress_rates.get(&class_ingress); - if e.is_none() || i.is_none() { - return Ok(TrafficDrift { - has_drift: true, - reason: format!("missing rate/ceil class details for peer class_id={class_id}"), - }); - } - if e.map(|v| !rate_ceil_matches_limit(v, &expected_limit)) - .unwrap_or(true) - || i.map(|v| !rate_ceil_matches_limit(v, &expected_limit)) - .unwrap_or(true) - { - return Ok(TrafficDrift { - has_drift: true, - reason: format!("rate/ceil mismatch for peer class_id={class_id}"), - }); - } - } - - let classes_egress = inspect_cmd( - transport, - &["tc", "class", "show", "dev", iface, "parent", "1:1"], - "egress leaf classes", - )?; - let classes_ingress = inspect_cmd( - transport, - &["tc", "class", "show", "dev", ifb, "parent", "2:1"], - "ingress leaf classes", - )?; - let expected_set: BTreeSet = expected_map.values().copied().collect(); - if collect_leaf_class_ids(&classes_egress) != expected_set { - return Ok(TrafficDrift { - has_drift: true, - reason: "egress per-peer class set mismatch".to_string(), - }); - } - if collect_leaf_class_ids(&classes_ingress) != expected_set { - return Ok(TrafficDrift { - has_drift: true, - reason: "ingress per-peer class set mismatch".to_string(), - }); - } - - let qdisc_egress = inspect_cmd( - transport, - &["tc", "qdisc", "show", "dev", iface], - "egress leaf qdisc", - )?; - let qdisc_ingress = inspect_cmd( - transport, - &["tc", "qdisc", "show", "dev", ifb], - "ingress leaf qdisc", - )?; - for class_id in expected_map.values() { - let class_egress = format!("1:{class_id}"); - let class_ingress = format!("2:{class_id}"); - if !has_fq_codel_for_parent(&qdisc_egress, &class_egress) { - return Ok(TrafficDrift { - has_drift: true, - reason: format!("missing fq_codel for egress class {class_egress}"), - }); - } - if !has_fq_codel_for_parent(&qdisc_ingress, &class_ingress) { - return Ok(TrafficDrift { - has_drift: true, - reason: format!("missing fq_codel for ingress class {class_ingress}"), - }); - } - } - - Ok(TrafficDrift { - has_drift: false, - reason: "no drift".to_string(), - }) -} - -fn inspect_cmd(transport: &dyn NsTransport, args: &[&str], what: &str) -> anyhow::Result { - transport - .run(args) - .map_err(|e| anyhow::anyhow!("{what} inspect failed: {e}")) -} - -fn extract_filter_ip_class_map(filters: &str) -> FilterInspect { - let mut out = BTreeMap::new(); - let mut duplicate_ips = BTreeSet::new(); - let mut class_to_ip = BTreeMap::new(); - let mut duplicate_classes = BTreeSet::new(); - - for block in split_filter_blocks(filters) { - let mut ip: Option = None; - let mut class_id: Option = None; - - for line in block.lines() { - let t = line.trim(); - if t.contains("match") && ip.is_none() { - ip = extract_u32_match_hex(t).and_then(|hex| hex_to_ipv4_string(&hex)); - } - if t.contains("flowid") && class_id.is_none() { - class_id = extract_flowid_class_id(t); - } - } - - if let (Some(ip), Some(class_id)) = (ip, class_id) { - if out.contains_key(&ip) { - duplicate_ips.insert(ip.clone()); - } - if class_to_ip.insert(class_id, ip.clone()).is_some() { - duplicate_classes.insert(class_id); - } - out.insert(ip, class_id); - } - } - - FilterInspect { - map: out, - duplicate_ips, - duplicate_classes, - } -} - -fn has_fq_codel_for_parent(raw_qdisc: &str, parent_class: &str) -> bool { - raw_qdisc.lines().map(str::trim).any(|line| { - line.contains("fq_codel") && line.contains("parent") && line.contains(parent_class) - }) -} - -fn has_ingress_redirect_to_ifb(output: &str, ifb_name: &str) -> bool { - let ifb_lower = ifb_name.to_ascii_lowercase(); - let needle_dev = format!("dev {ifb_lower}"); - let needle_device = format!("device {ifb_lower}"); - - output.lines().any(|line| { - let l = line.to_ascii_lowercase(); - l.contains("mirred") - && l.contains("redirect") - && (l.contains(&needle_dev) || l.contains(&needle_device)) - }) -} - -fn link_is_up(output: &str) -> bool { - output.contains("state UP") || (output.contains('<') && output.contains("UP")) -} - -fn is_not_found_error(stderr_or_msg: &str) -> bool { - let s = stderr_or_msg.to_ascii_lowercase(); - s.contains("does not exist") - || s.contains("not found") - || s.contains("cannot find device") - || s.contains("no such device") - || s.contains("no such file") - || s.contains("handle of zero") -} - -fn parse_root_class_rate_ceil(raw: &str, class_id: &str) -> Option { - for line in raw.lines() { - let t = line.trim(); - if !t.contains("class htb") || !t.contains(" rate ") || !t.contains(" ceil ") { - continue; - } - let toks: Vec<&str> = t.split_whitespace().collect(); - let class = toks - .iter() - .position(|v| *v == "htb") - .and_then(|i| toks.get(i + 1))?; - if *class != class_id { - continue; - } - let rate = toks - .iter() - .position(|v| *v == "rate") - .and_then(|i| toks.get(i + 1)) - .map(|s| s.to_string())?; - let ceil = toks - .iter() - .position(|v| *v == "ceil") - .and_then(|i| toks.get(i + 1)) - .map(|s| s.to_string())?; - return Some(ClassRateCeil { rate, ceil }); - } - None -} - -fn split_filter_blocks(raw: &str) -> Vec { - let mut blocks = Vec::new(); - let mut cur = Vec::new(); - - for line in raw.lines() { - let t = line.trim(); - if t.starts_with("filter") && !cur.is_empty() { - blocks.push(cur.join("\n")); - cur.clear(); - } - if !t.is_empty() { - cur.push(t.to_string()); - } - } - if !cur.is_empty() { - blocks.push(cur.join("\n")); - } - - blocks -} - -fn extract_flowid_class_id(line: &str) -> Option { - let toks: Vec<&str> = line.split_whitespace().collect(); - let flowid = toks - .iter() - .position(|v| *v == "flowid") - .and_then(|i| toks.get(i + 1))?; - let class_tail = flowid.split(':').nth(1)?; - class_tail.parse::().ok() -} - -fn expected_peer_class_map( - peers: &[wg::Peer], - class_start_id: u16, -) -> anyhow::Result> { - let mut map = BTreeMap::new(); - for (idx, p) in peers.iter().enumerate() { - let idx_u16 = u16::try_from(idx) - .map_err(|_| anyhow::anyhow!("too many peers for class id allocation"))?; - let class_id = class_start_id - .checked_add(idx_u16) - .ok_or_else(|| anyhow::anyhow!("class id overflow: too many peers"))?; - map.insert(p.ipv4.to_string(), class_id); - } - Ok(map) -} - -fn extract_u32_match_hex(line: &str) -> Option { - 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 Some(hex.to_ascii_lowercase()); - } - None -} - -fn hex_to_ipv4_string(hex: &str) -> Option { - if 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(Ipv4Addr::new(b1, b2, b3, b4).to_string()) -} - -fn collect_leaf_class_ids(raw: &str) -> BTreeSet { - let mut out = BTreeSet::new(); - for l in raw.lines() { - let line = l.trim(); - let Some(pos) = line.find("class htb ") else { - continue; - }; - let class = &line[(pos + "class htb ".len())..]; - let Some(head) = class.split_whitespace().next() else { - continue; - }; - if let Some(class_id) = head.split(':').nth(1).and_then(|v| v.parse::().ok()) - && class_id != 1 - { - out.insert(class_id); - } - } - out -} - -fn teardown_owned_topology( - transport: &dyn NsTransport, - iface: &str, - ifb: &str, -) -> anyhow::Result<()> { - run_allow_notfound(transport, &["tc", "qdisc", "del", "dev", iface, "root"])?; - run_allow_notfound(transport, &["tc", "qdisc", "del", "dev", iface, "ingress"])?; - run_allow_notfound(transport, &["ip", "link", "del", ifb])?; - Ok(()) -} - -fn apply_full_topology( - cfg: &Config, - transport: &dyn NsTransport, - iface: &str, - ifb: &str, - peers: &[wg::Peer], -) -> anyhow::Result<()> { - let root = cfg.traffic.root_rate.as_str(); - run_allow_exists(transport, &["modprobe", "ifb"])?; - run_allow_exists( - transport, - &["ip", "link", "add", "name", ifb, "type", "ifb"], - )?; - run_step(transport, &["ip", "link", "set", "dev", ifb, "up"])?; - run_allow_exists( - transport, - &[ - "tc", "qdisc", "add", "dev", iface, "root", "handle", "1:", "htb", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "class", "add", "dev", iface, "parent", "1:", "classid", "1:1", "htb", "rate", - root, "ceil", root, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "qdisc", "add", "dev", iface, "handle", "ffff:", "ingress", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "filter", "add", "dev", iface, "parent", "ffff:", "protocol", "all", "u32", - "match", "u32", "0", "0", "action", "mirred", "egress", "redirect", "dev", ifb, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "qdisc", "add", "dev", ifb, "root", "handle", "2:", "htb", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", "class", "add", "dev", ifb, "parent", "2:", "classid", "2:1", "htb", "rate", - root, "ceil", root, - ], - )?; - - for (idx, p) in peers.iter().enumerate() { - let idx_u16 = u16::try_from(idx) - .map_err(|_| anyhow::anyhow!("too many peers for class id allocation"))?; - let class_id = cfg - .traffic - .class_start_id - .checked_add(idx_u16) - .ok_or_else(|| anyhow::anyhow!("class id overflow: too many peers"))?; - let ip = p.ipv4.to_string(); - let class_egress = format!("1:{class_id}"); - let class_ingress = format!("2:{class_id}"); - let limit = cfg.traffic.peer_limit.as_str(); - - run_allow_exists( - transport, - &[ - "tc", - "class", - "add", - "dev", - iface, - "parent", - "1:1", - "classid", - class_egress.as_str(), - "htb", - "rate", - limit, - "ceil", - limit, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "qdisc", - "add", - "dev", - iface, - "parent", - class_egress.as_str(), - "fq_codel", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "filter", - "add", - "dev", - iface, - "parent", - "1:", - "protocol", - "ip", - "u32", - "match", - "ip", - "dst", - ip.as_str(), - "flowid", - class_egress.as_str(), - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "class", - "add", - "dev", - ifb, - "parent", - "2:1", - "classid", - class_ingress.as_str(), - "htb", - "rate", - limit, - "ceil", - limit, - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "qdisc", - "add", - "dev", - ifb, - "parent", - class_ingress.as_str(), - "fq_codel", - ], - )?; - run_allow_exists( - transport, - &[ - "tc", - "filter", - "add", - "dev", - ifb, - "parent", - "2:", - "protocol", - "ip", - "u32", - "match", - "ip", - "src", - ip.as_str(), - "flowid", - class_ingress.as_str(), - ], - )?; - } - - let verify = inspect_and_diff( - transport, - iface, - ifb, - cfg.traffic.class_start_id, - &cfg.traffic.root_rate, - &cfg.traffic.peer_limit, - peers, - )?; - if verify.has_drift { - anyhow::bail!("verify-after-apply failed: {}", verify.reason); - } - - Ok(()) -} - -fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { - let _ = transport.run(args)?; - Ok(()) -} - -fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { - match transport.run(args) { - Ok(_) => Ok(()), - Err(e) => { - let msg = e.to_string().to_ascii_lowercase(); - if msg.contains("file exists") - || msg.contains("eexist") - || msg.contains("object already exists") - || msg.contains("exclusive flag on") - { - return Ok(()); - } - Err(e) - } - } -} - -fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { - match transport.run(args) { - Ok(_) => Ok(()), - Err(e) => { - let msg = e.to_string().to_ascii_lowercase(); - if msg.contains("no such file") - || msg.contains("no such") - || msg.contains("not found") - || msg.contains("cannot find") - || msg.contains("cannot find device") - || msg.contains("no such device") - || msg.contains("does not exist") - || msg.contains("handle of zero") - { - return Ok(()); - } - Err(e) - } - } -} - -fn parse_class_rate_ceil_map(raw: &str, prefix: &str) -> BTreeMap { - let mut out = BTreeMap::new(); - for line in raw.lines() { - let t = line.trim(); - if !t.contains("class htb") || !t.contains(" rate ") || !t.contains(" ceil ") { - continue; - } - let toks: Vec<&str> = t.split_whitespace().collect(); - let class = toks - .iter() - .position(|v| *v == "htb") - .and_then(|i| toks.get(i + 1)) - .map(|s| s.to_string()); - let rate = toks - .iter() - .position(|v| *v == "rate") - .and_then(|i| toks.get(i + 1)) - .map(|s| s.to_string()); - let ceil = toks - .iter() - .position(|v| *v == "ceil") - .and_then(|i| toks.get(i + 1)) - .map(|s| s.to_string()); - if let (Some(class), Some(rate), Some(ceil)) = (class, rate, ceil) - && class.starts_with(prefix) - && !class.ends_with(":1") - { - out.insert(class, ClassRateCeil { rate, ceil }); - } - } - out -} - -fn rate_ceil_matches_limit(v: &ClassRateCeil, limit: &str) -> bool { - let Ok(limit_bps) = parse_tc_rate_bps(limit) else { - return false; - }; - let Ok(rate_bps) = parse_tc_rate_bps(&v.rate) else { - return false; - }; - let Ok(ceil_bps) = parse_tc_rate_bps(&v.ceil) else { - return false; - }; - rate_bps == limit_bps && ceil_bps == limit_bps -} - -fn normalize_tc_rate(v: &str) -> String { - v.trim().to_ascii_lowercase().replace(' ', "") -} - -fn parse_tc_rate_bps(v: &str) -> anyhow::Result { - let n = normalize_tc_rate(v); - let (num, mul) = if let Some(x) = n.strip_suffix("gbit") { - (x, 1_000_000_000u64) - } else if let Some(x) = n.strip_suffix("mbit") { - (x, 1_000_000u64) - } else if let Some(x) = n.strip_suffix("kbit") { - (x, 1_000u64) - } else if let Some(x) = n.strip_suffix("bit") { - (x, 1u64) - } else { - anyhow::bail!("unknown tc rate unit: {v}") - }; - let base: u64 = num - .parse() - .map_err(|_| anyhow::anyhow!("invalid tc rate: {v}"))?; - base.checked_mul(mul) - .ok_or_else(|| anyhow::anyhow!("tc rate overflow: {v}")) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn parse_tc_u32_hex_ipv4() { - let raw = " match 0a080105/ffffffff at 16\n flowid 1:42"; - let inspect = extract_filter_ip_class_map(raw); - assert!(inspect.duplicate_ips.is_empty()); - assert!(inspect.duplicate_classes.is_empty()); - assert_eq!(inspect.map.get("10.8.1.5").copied(), Some(42)); - } - - #[test] - fn parse_block_based_filters() { - let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:11"; - let inspect = extract_filter_ip_class_map(raw); - assert!(inspect.duplicate_ips.is_empty()); - assert!(inspect.duplicate_classes.is_empty()); - assert_eq!(inspect.map.get("10.8.1.2").copied(), Some(10)); - assert_eq!(inspect.map.get("10.8.1.3").copied(), Some(11)); - } - - #[test] - fn detect_duplicate_filter_ip() { - let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080102/ffffffff at 16\n flowid 1:10"; - let inspect = extract_filter_ip_class_map(raw); - assert!(inspect.duplicate_ips.contains("10.8.1.2")); - } - - #[test] - fn detect_duplicate_filter_class_flowid() { - let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:10"; - let inspect = extract_filter_ip_class_map(raw); - assert!(inspect.duplicate_classes.contains(&10)); - } - - #[test] - fn detect_fq_codel_parent() { - let raw = "qdisc fq_codel 8001: dev wg0 parent 1:10 limit 10240p"; - assert!(has_fq_codel_for_parent(raw, "1:10")); - assert!(!has_fq_codel_for_parent(raw, "1:11")); - } - - #[test] - fn detects_missing_ingress_redirect() { - let raw = "filter parent ffff: protocol all pref 1 u32"; - assert!(!has_ingress_redirect_to_ifb(raw, "ifb0")); - } - - #[test] - fn detects_ingress_redirect_to_ifb() { - let raw = "filter parent ffff: protocol all pref 1 u32\n action order 1: mirred (Egress Redirect to device ifb0) stolen"; - assert!(has_ingress_redirect_to_ifb(raw, "ifb0")); - } - - #[test] - fn detects_ingress_redirect_to_ifb_dev_format_lowercase() { - let raw = "filter parent ffff: protocol all pref 1 u32\n action order 1: mirred egress redirect dev ifb0"; - assert!(has_ingress_redirect_to_ifb(raw, "ifb0")); - } - - #[test] - fn does_not_match_redirect_to_other_ifb() { - let raw = "filter parent ffff: protocol all pref 1 u32\n action order 1: mirred (Egress Redirect to device ifb1) stolen"; - assert!(!has_ingress_redirect_to_ifb(raw, "ifb0")); - } - - #[test] - fn detects_ifb_up_state() { - assert!(link_is_up( - "2: ifb0: mtu 1500 qdisc fq_codel state UNKNOWN" - )); - assert!(link_is_up("2: ifb0: mtu 1500 state UP")); - assert!(!link_is_up( - "2: ifb0: mtu 1500 state DOWN" - )); - } - - #[test] - fn detects_not_found_errors() { - assert!(is_not_found_error("RTNETLINK answers: No such device")); - assert!(is_not_found_error("Cannot find device \"ifb0\"")); - assert!(is_not_found_error("does not exist")); - assert!(is_not_found_error( - "Cannot delete qdisc with handle of zero" - )); - assert!(!is_not_found_error("permission denied")); - } - - #[test] - fn parse_root_rate_ceil() { - let raw = "class htb 1:10 parent 1:1 leaf 800f: prio 0 rate 50Mbit ceil 50Mbit burst 1600b cburst 1600b\nclass htb 1:1 root rate 500Mbit ceil 500Mbit burst 1600b cburst 1600b"; - let rc = parse_root_class_rate_ceil(raw, "1:1").expect("root class exists"); - assert_eq!(normalize_tc_rate(&rc.rate), "500mbit"); - assert_eq!(normalize_tc_rate(&rc.ceil), "500mbit"); - } - - #[test] - fn parse_root_rate_ceil_returns_none_when_only_parent_matches() { - let raw = "class htb 1:10 parent 1:1 leaf 800f: prio 0 rate 50Mbit ceil 50Mbit burst 1600b cburst 1600b"; - assert!(parse_root_class_rate_ceil(raw, "1:1").is_none()); - } - - #[test] - fn parse_class_rate_ceil() { - let raw = "class htb 1:10 parent 1:1 rate 70Mbit ceil 70Mbit burst 1599b cburst 1599b\nclass htb 1:1 root rate 1000Mbit ceil 1000Mbit"; - let m = parse_class_rate_ceil_map(raw, "1:"); - let rc = m.get("1:10").expect("class exists"); - assert_eq!(normalize_tc_rate(&rc.rate), "70mbit"); - assert_eq!(normalize_tc_rate(&rc.ceil), "70mbit"); - } - - #[test] - fn rate_limit_match_normalized() { - let rc = ClassRateCeil { - rate: "70Mbit".to_string(), - ceil: "70mbit".to_string(), - }; - assert!(rate_ceil_matches_limit(&rc, "70mbit")); - assert!(!rate_ceil_matches_limit(&rc, "50mbit")); - } - - #[test] - fn tc_rate_numeric_equivalence() { - let rc = ClassRateCeil { - rate: "1Gbit".to_string(), - ceil: "1000Mbit".to_string(), - }; - assert!(rate_ceil_matches_limit(&rc, "1000mbit")); - assert!(rate_ceil_matches_limit(&rc, "1gbit")); - } -} diff --git a/traffic_core/src/traffic/apply.rs b/traffic_core/src/traffic/apply.rs new file mode 100644 index 0000000..d3b943c --- /dev/null +++ b/traffic_core/src/traffic/apply.rs @@ -0,0 +1,258 @@ +//! Tears down and rebuilds the owned tc/IFB topology. +//! +//! Only owned objects are touched: the root qdisc on the AWG/WG interface, its ingress +//! qdisc, and the IFB device. Apply always ends with a verify pass, so a partially +//! applied topology surfaces as an error instead of silently becoming the new state. + +use crate::{config::Config, netns::NsTransport, wg}; + +use super::inspect::inspect_and_diff; + +pub(super) fn teardown_owned_topology( + transport: &dyn NsTransport, + iface: &str, + ifb: &str, +) -> anyhow::Result<()> { + run_allow_notfound(transport, &["tc", "qdisc", "del", "dev", iface, "root"])?; + run_allow_notfound(transport, &["tc", "qdisc", "del", "dev", iface, "ingress"])?; + run_allow_notfound(transport, &["ip", "link", "del", ifb])?; + Ok(()) +} + +pub(super) fn apply_full_topology( + cfg: &Config, + transport: &dyn NsTransport, + iface: &str, + ifb: &str, + peers: &[wg::Peer], +) -> anyhow::Result<()> { + let root = cfg.traffic.root_rate.as_str(); + run_allow_exists(transport, &["modprobe", "ifb"])?; + run_allow_exists( + transport, + &["ip", "link", "add", "name", ifb, "type", "ifb"], + )?; + run_step(transport, &["ip", "link", "set", "dev", ifb, "up"])?; + run_allow_exists( + transport, + &[ + "tc", "qdisc", "add", "dev", iface, "root", "handle", "1:", "htb", + ], + )?; + run_allow_exists( + transport, + &[ + "tc", "class", "add", "dev", iface, "parent", "1:", "classid", "1:1", "htb", "rate", + root, "ceil", root, + ], + )?; + run_allow_exists( + transport, + &[ + "tc", "qdisc", "add", "dev", iface, "handle", "ffff:", "ingress", + ], + )?; + run_allow_exists( + transport, + &[ + "tc", "filter", "add", "dev", iface, "parent", "ffff:", "protocol", "all", "u32", + "match", "u32", "0", "0", "action", "mirred", "egress", "redirect", "dev", ifb, + ], + )?; + run_allow_exists( + transport, + &[ + "tc", "qdisc", "add", "dev", ifb, "root", "handle", "2:", "htb", + ], + )?; + run_allow_exists( + transport, + &[ + "tc", "class", "add", "dev", ifb, "parent", "2:", "classid", "2:1", "htb", "rate", + root, "ceil", root, + ], + )?; + + for (idx, p) in peers.iter().enumerate() { + let idx_u16 = u16::try_from(idx) + .map_err(|_| anyhow::anyhow!("too many peers for class id allocation"))?; + let class_id = cfg + .traffic + .class_start_id + .checked_add(idx_u16) + .ok_or_else(|| anyhow::anyhow!("class id overflow: too many peers"))?; + let ip = p.ipv4.to_string(); + let class_egress = format!("1:{class_id}"); + let class_ingress = format!("2:{class_id}"); + let limit = cfg.traffic.peer_limit.as_str(); + + run_allow_exists( + transport, + &[ + "tc", + "class", + "add", + "dev", + iface, + "parent", + "1:1", + "classid", + class_egress.as_str(), + "htb", + "rate", + limit, + "ceil", + limit, + ], + )?; + run_allow_exists( + transport, + &[ + "tc", + "qdisc", + "add", + "dev", + iface, + "parent", + class_egress.as_str(), + "fq_codel", + ], + )?; + run_allow_exists( + transport, + &[ + "tc", + "filter", + "add", + "dev", + iface, + "parent", + "1:", + "protocol", + "ip", + "u32", + "match", + "ip", + "dst", + ip.as_str(), + "flowid", + class_egress.as_str(), + ], + )?; + run_allow_exists( + transport, + &[ + "tc", + "class", + "add", + "dev", + ifb, + "parent", + "2:1", + "classid", + class_ingress.as_str(), + "htb", + "rate", + limit, + "ceil", + limit, + ], + )?; + run_allow_exists( + transport, + &[ + "tc", + "qdisc", + "add", + "dev", + ifb, + "parent", + class_ingress.as_str(), + "fq_codel", + ], + )?; + run_allow_exists( + transport, + &[ + "tc", + "filter", + "add", + "dev", + ifb, + "parent", + "2:", + "protocol", + "ip", + "u32", + "match", + "ip", + "src", + ip.as_str(), + "flowid", + class_ingress.as_str(), + ], + )?; + } + + let verify = inspect_and_diff( + transport, + iface, + ifb, + cfg.traffic.class_start_id, + &cfg.traffic.root_rate, + &cfg.traffic.peer_limit, + peers, + )?; + if verify.has_drift { + anyhow::bail!("verify-after-apply failed: {}", verify.reason); + } + + Ok(()) +} + +fn run_step(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { + let _ = transport.run(args)?; + Ok(()) +} + +/// Tolerates only the idempotency conflicts that a rebuild legitimately produces. +/// Anything else is a real apply failure and must propagate. +fn run_allow_exists(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { + match transport.run(args) { + Ok(_) => Ok(()), + Err(e) => { + let msg = e.to_string().to_ascii_lowercase(); + if msg.contains("file exists") + || msg.contains("eexist") + || msg.contains("object already exists") + || msg.contains("exclusive flag on") + { + return Ok(()); + } + Err(e) + } + } +} + +/// Deleting an object that is already gone is the expected outcome of teardown, not an +/// error. Every other failure propagates. +fn run_allow_notfound(transport: &dyn NsTransport, args: &[&str]) -> anyhow::Result<()> { + match transport.run(args) { + Ok(_) => Ok(()), + Err(e) => { + let msg = e.to_string().to_ascii_lowercase(); + if msg.contains("no such file") + || msg.contains("no such") + || msg.contains("not found") + || msg.contains("cannot find") + || msg.contains("cannot find device") + || msg.contains("no such device") + || msg.contains("does not exist") + || msg.contains("handle of zero") + { + return Ok(()); + } + Err(e) + } + } +} diff --git a/traffic_core/src/traffic/inspect.rs b/traffic_core/src/traffic/inspect.rs new file mode 100644 index 0000000..61d0244 --- /dev/null +++ b/traffic_core/src/traffic/inspect.rs @@ -0,0 +1,325 @@ +//! Reads the actual tc/IFB topology and diffs it against the desired one. +//! +//! The first mismatch wins: every check returns immediately with a reason, because the +//! only supported repair is a full controlled rebuild, so there is nothing to gain from +//! collecting a complete list of defects. + +use std::collections::{BTreeMap, BTreeSet}; + +use crate::{netns::NsTransport, wg}; + +use super::{ + TrafficDrift, + parse_tc::{ + collect_leaf_class_ids, extract_filter_ip_class_map, has_fq_codel_for_parent, + has_ingress_redirect_to_ifb, is_not_found_error, link_is_up, normalize_tc_rate, + parse_class_rate_ceil_map, parse_root_class_rate_ceil, rate_ceil_matches_limit, + }, +}; + +#[allow(clippy::too_many_arguments)] +pub(super) fn inspect_and_diff( + transport: &dyn NsTransport, + iface: &str, + ifb: &str, + class_start_id: u16, + root_rate: &str, + peer_limit: &str, + peers: &[wg::Peer], +) -> anyhow::Result { + let root_qdisc = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", iface], + "egress qdisc", + )?; + if !root_qdisc.contains("htb") { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing egress root htb qdisc".to_string(), + }); + } + + let ingress_qdisc = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", iface], + "ingress qdisc", + )?; + if !ingress_qdisc.contains("ingress") { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing ingress qdisc".to_string(), + }); + } + + let ingress_redirect = inspect_cmd( + transport, + &["tc", "filter", "show", "dev", iface, "parent", "ffff:"], + "ingress redirect filter", + )?; + if !has_ingress_redirect_to_ifb(&ingress_redirect, ifb) { + return Ok(TrafficDrift { + has_drift: true, + reason: format!("missing ingress redirect to {ifb}"), + }); + } + + let ifb_link = match transport.run(&["ip", "link", "show", ifb]) { + Ok(out) => out, + Err(e) if is_not_found_error(&e.to_string()) => { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing ifb interface".to_string(), + }); + } + Err(e) => return Err(anyhow::anyhow!("ifb link inspect failed: {e}")), + }; + if ifb_link.trim().is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: "missing ifb interface".to_string(), + }); + } + if !link_is_up(&ifb_link) { + return Ok(TrafficDrift { + has_drift: true, + reason: format!("{ifb} exists but is not UP"), + }); + } + + let filters_egress = inspect_cmd( + transport, + &["tc", "filter", "show", "dev", iface, "parent", "1:"], + "egress filters", + )?; + let filters_ingress = inspect_cmd( + transport, + &["tc", "filter", "show", "dev", ifb, "parent", "2:"], + "ingress filters", + )?; + + let inspect_egress = extract_filter_ip_class_map(&filters_egress); + let inspect_ingress = extract_filter_ip_class_map(&filters_ingress); + if !inspect_egress.duplicate_ips.is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: format!( + "duplicate egress filters detected for {} peer(s)", + inspect_egress.duplicate_ips.len() + ), + }); + } + if !inspect_egress.duplicate_classes.is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: "duplicate egress class flowid mapping detected".to_string(), + }); + } + if !inspect_ingress.duplicate_ips.is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: format!( + "duplicate ingress filters detected for {} peer(s)", + inspect_ingress.duplicate_ips.len() + ), + }); + } + if !inspect_ingress.duplicate_classes.is_empty() { + return Ok(TrafficDrift { + has_drift: true, + reason: "duplicate ingress class flowid mapping detected".to_string(), + }); + } + let actual_map_egress = inspect_egress.map; + let actual_map_ingress = inspect_ingress.map; + let actual_ips_egress: BTreeSet = actual_map_egress.keys().cloned().collect(); + let actual_ips_ingress: BTreeSet = actual_map_ingress.keys().cloned().collect(); + let expected_ips: BTreeSet = peers.iter().map(|p| p.ipv4.to_string()).collect(); + + if actual_ips_egress != expected_ips { + return Ok(TrafficDrift { + has_drift: true, + reason: format!( + "egress peer filter mismatch (actual={}, expected={})", + actual_ips_egress.len(), + expected_ips.len() + ), + }); + } + + if actual_ips_ingress != expected_ips { + return Ok(TrafficDrift { + has_drift: true, + reason: format!( + "ingress peer filter mismatch (actual={}, expected={})", + actual_ips_ingress.len(), + expected_ips.len() + ), + }); + } + + let expected_map = expected_peer_class_map(peers, class_start_id)?; + if actual_map_egress != expected_map { + return Ok(TrafficDrift { + has_drift: true, + reason: format!( + "egress class mapping mismatch (actual={}, expected={})", + actual_map_egress.len(), + expected_map.len() + ), + }); + } + if actual_map_ingress != expected_map { + return Ok(TrafficDrift { + has_drift: true, + reason: format!( + "ingress class mapping mismatch (actual={}, expected={})", + actual_map_ingress.len(), + expected_map.len() + ), + }); + } + + let expected_limit = normalize_tc_rate(peer_limit); + let classes_egress_full = inspect_cmd( + transport, + &["tc", "class", "show", "dev", iface], + "egress classes", + )?; + let classes_ingress_full = inspect_cmd( + transport, + &["tc", "class", "show", "dev", ifb], + "ingress classes", + )?; + let expected_root = normalize_tc_rate(root_rate); + let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:"); + let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:"); + + let root_egress = parse_root_class_rate_ceil(&classes_egress_full, "1:1"); + let root_ingress = parse_root_class_rate_ceil(&classes_ingress_full, "2:1"); + if root_egress + .as_ref() + .map(|v| !rate_ceil_matches_limit(v, &expected_root)) + .unwrap_or(true) + { + return Ok(TrafficDrift { + has_drift: true, + reason: "egress root class 1:1 rate/ceil mismatch".to_string(), + }); + } + if root_ingress + .as_ref() + .map(|v| !rate_ceil_matches_limit(v, &expected_root)) + .unwrap_or(true) + { + return Ok(TrafficDrift { + has_drift: true, + reason: "ingress root class 2:1 rate/ceil mismatch".to_string(), + }); + } + + for class_id in expected_map.values() { + let class_egress = format!("1:{class_id}"); + let class_ingress = format!("2:{class_id}"); + let e = egress_rates.get(&class_egress); + let i = ingress_rates.get(&class_ingress); + if e.is_none() || i.is_none() { + return Ok(TrafficDrift { + has_drift: true, + reason: format!("missing rate/ceil class details for peer class_id={class_id}"), + }); + } + if e.map(|v| !rate_ceil_matches_limit(v, &expected_limit)) + .unwrap_or(true) + || i.map(|v| !rate_ceil_matches_limit(v, &expected_limit)) + .unwrap_or(true) + { + return Ok(TrafficDrift { + has_drift: true, + reason: format!("rate/ceil mismatch for peer class_id={class_id}"), + }); + } + } + + let classes_egress = inspect_cmd( + transport, + &["tc", "class", "show", "dev", iface, "parent", "1:1"], + "egress leaf classes", + )?; + let classes_ingress = inspect_cmd( + transport, + &["tc", "class", "show", "dev", ifb, "parent", "2:1"], + "ingress leaf classes", + )?; + let expected_set: BTreeSet = expected_map.values().copied().collect(); + if collect_leaf_class_ids(&classes_egress) != expected_set { + return Ok(TrafficDrift { + has_drift: true, + reason: "egress per-peer class set mismatch".to_string(), + }); + } + if collect_leaf_class_ids(&classes_ingress) != expected_set { + return Ok(TrafficDrift { + has_drift: true, + reason: "ingress per-peer class set mismatch".to_string(), + }); + } + + let qdisc_egress = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", iface], + "egress leaf qdisc", + )?; + let qdisc_ingress = inspect_cmd( + transport, + &["tc", "qdisc", "show", "dev", ifb], + "ingress leaf qdisc", + )?; + for class_id in expected_map.values() { + let class_egress = format!("1:{class_id}"); + let class_ingress = format!("2:{class_id}"); + if !has_fq_codel_for_parent(&qdisc_egress, &class_egress) { + return Ok(TrafficDrift { + has_drift: true, + reason: format!("missing fq_codel for egress class {class_egress}"), + }); + } + if !has_fq_codel_for_parent(&qdisc_ingress, &class_ingress) { + return Ok(TrafficDrift { + has_drift: true, + reason: format!("missing fq_codel for ingress class {class_ingress}"), + }); + } + } + + Ok(TrafficDrift { + has_drift: false, + reason: "no drift".to_string(), + }) +} + +/// The class-id contract: peer IP strings sorted lexicographically, then +/// `class_start_id + index`. +/// +/// Sorting is textual on purpose, so `10.8.1.10` comes before `10.8.1.2`. Switching to +/// numeric IPv4 ordering would silently renumber every existing class. +pub(super) fn expected_peer_class_map( + peers: &[wg::Peer], + class_start_id: u16, +) -> anyhow::Result> { + let mut map = BTreeMap::new(); + for (idx, p) in peers.iter().enumerate() { + let idx_u16 = u16::try_from(idx) + .map_err(|_| anyhow::anyhow!("too many peers for class id allocation"))?; + let class_id = class_start_id + .checked_add(idx_u16) + .ok_or_else(|| anyhow::anyhow!("class id overflow: too many peers"))?; + map.insert(p.ipv4.to_string(), class_id); + } + Ok(map) +} + +fn inspect_cmd(transport: &dyn NsTransport, args: &[&str], what: &str) -> anyhow::Result { + transport + .run(args) + .map_err(|e| anyhow::anyhow!("{what} inspect failed: {e}")) +} diff --git a/traffic_core/src/traffic/mod.rs b/traffic_core/src/traffic/mod.rs new file mode 100644 index 0000000..f0e4f15 --- /dev/null +++ b/traffic_core/src/traffic/mod.rs @@ -0,0 +1,247 @@ +//! Per-peer traffic shaping: owned tc topology on the AWG/WG interface and its IFB +//! counterpart for ingress. +//! +//! The module is split by responsibility: [`parse_tc`] turns command output into models, +//! [`inspect`] diffs actual against desired, [`apply`] rebuilds the owned topology. This +//! file keeps the shared models and the entry points that orchestrate the three. + +mod apply; +mod inspect; +mod parse_tc; + +use crate::{config::Config, docker, lock::FileLock, netns::build_transport, wg}; + +use std::{ + collections::{BTreeMap, BTreeSet}, + time::Duration, +}; + +use apply::{apply_full_topology, teardown_owned_topology}; +use inspect::{expected_peer_class_map, inspect_and_diff}; +use parse_tc::{ + extract_filter_ip_class_map, normalize_tc_rate, parse_class_rate_ceil_map, + rate_ceil_matches_limit, +}; + +#[derive(Debug, Clone)] +struct TrafficDrift { + has_drift: bool, + reason: String, +} + +#[derive(Debug, Clone)] +struct FilterInspect { + map: BTreeMap, + duplicate_ips: BTreeSet, + duplicate_classes: BTreeSet, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ClassRateCeil { + rate: String, + ceil: String, +} + +#[derive(Debug, Clone)] +pub struct PeerTrafficStatus { + pub ip: String, + pub class_id: u16, + pub egress_filter_present: bool, + pub ingress_filter_present: bool, + pub egress_rate: Option, + pub egress_ceil: Option, + pub ingress_rate: Option, + pub ingress_ceil: Option, + pub drift_reason: Option, +} + +pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result> { + if !cfg.traffic.enabled { + return Ok(None); + } + + let target = docker::discover_target(cfg)?; + let transport = build_transport(target.container_pid, &target.container_name); + let peers = wg::discover_peers(cfg, &target)?; + + if peers.is_empty() { + return Ok(Some("no peers discovered".to_string())); + } + + let drift = inspect_and_diff( + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + cfg.traffic.class_start_id, + &cfg.traffic.root_rate, + &cfg.traffic.peer_limit, + &peers, + )?; + + if drift.has_drift { + Ok(Some(drift.reason)) + } else { + Ok(None) + } +} + +pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> { + if !cfg.traffic.enabled { + return Ok(()); + } + + let _lock = + FileLock::acquire_timeout("/run/awg2-harness/traffic.lock", Duration::from_secs(10))?; + let target = docker::discover_target(cfg)?; + let transport = build_transport(target.container_pid, &target.container_name); + let peers = wg::discover_peers(cfg, &target)?; + + if peers.is_empty() { + println!("no peers found, skip tc apply"); + return Ok(()); + } + + let drift = inspect_and_diff( + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + cfg.traffic.class_start_id, + &cfg.traffic.root_rate, + &cfg.traffic.peer_limit, + &peers, + )?; + + if !drift.has_drift { + println!("traffic rules up-to-date, skip apply"); + return Ok(()); + } + + println!("traffic drift detected: {}", drift.reason); + + teardown_owned_topology( + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + )?; + apply_full_topology( + cfg, + transport.as_ref(), + &target.interface_name, + &cfg.traffic.ifb_name, + &peers, + )?; + + Ok(()) +} + +pub fn peer_status_table(cfg: &Config) -> anyhow::Result>> { + if !cfg.traffic.enabled { + return Ok(None); + } + + let target = docker::discover_target(cfg)?; + let transport = build_transport(target.container_pid, &target.container_name); + let peers = wg::discover_peers(cfg, &target)?; + if peers.is_empty() { + return Ok(Some(Vec::new())); + } + + let filters_egress = transport + .run(&[ + "tc", + "filter", + "show", + "dev", + &target.interface_name, + "parent", + "1:", + ]) + .unwrap_or_default(); + let filters_ingress = transport + .run(&[ + "tc", + "filter", + "show", + "dev", + &cfg.traffic.ifb_name, + "parent", + "2:", + ]) + .unwrap_or_default(); + let classes_egress_full = transport + .run(&["tc", "class", "show", "dev", &target.interface_name]) + .unwrap_or_default(); + let classes_ingress_full = transport + .run(&["tc", "class", "show", "dev", &cfg.traffic.ifb_name]) + .unwrap_or_default(); + + let inspect_egress = extract_filter_ip_class_map(&filters_egress); + let inspect_ingress = extract_filter_ip_class_map(&filters_ingress); + let actual_map_egress = inspect_egress.map; + let actual_map_ingress = inspect_ingress.map; + let egress_rates = parse_class_rate_ceil_map(&classes_egress_full, "1:"); + let ingress_rates = parse_class_rate_ceil_map(&classes_ingress_full, "2:"); + let expected_map = expected_peer_class_map(&peers, cfg.traffic.class_start_id)?; + let expected_limit = normalize_tc_rate(&cfg.traffic.peer_limit); + + let mut out = Vec::with_capacity(peers.len()); + for p in peers { + let ip = p.ipv4.to_string(); + let class_id = *expected_map.get(&ip).unwrap_or(&cfg.traffic.class_start_id); + let class_egress = format!("1:{class_id}"); + let class_ingress = format!("2:{class_id}"); + + let egress_filter_present = actual_map_egress.get(&ip).copied() == Some(class_id); + let ingress_filter_present = actual_map_ingress.get(&ip).copied() == Some(class_id); + let e = egress_rates.get(&class_egress); + let i = ingress_rates.get(&class_ingress); + + let mut reasons = Vec::new(); + if !egress_filter_present { + reasons.push("missing/mismatched egress filter".to_string()); + } + if !ingress_filter_present { + reasons.push("missing/mismatched ingress filter".to_string()); + } + if e.is_none() { + reasons.push("missing egress class".to_string()); + } + if i.is_none() { + reasons.push("missing ingress class".to_string()); + } + if let Some(v) = e + && !rate_ceil_matches_limit(v, &expected_limit) + { + reasons.push("egress rate/ceil mismatch".to_string()); + } + if let Some(v) = i + && !rate_ceil_matches_limit(v, &expected_limit) + { + reasons.push("ingress rate/ceil mismatch".to_string()); + } + if inspect_egress.duplicate_ips.contains(&ip) { + reasons.push("duplicate egress filters".to_string()); + } + if inspect_ingress.duplicate_ips.contains(&ip) { + reasons.push("duplicate ingress filters".to_string()); + } + + out.push(PeerTrafficStatus { + ip, + class_id, + egress_filter_present, + ingress_filter_present, + egress_rate: e.map(|v| v.rate.clone()), + egress_ceil: e.map(|v| v.ceil.clone()), + ingress_rate: i.map(|v| v.rate.clone()), + ingress_ceil: i.map(|v| v.ceil.clone()), + drift_reason: if reasons.is_empty() { + None + } else { + Some(reasons.join(", ")) + }, + }); + } + + Ok(Some(out)) +} diff --git a/traffic_core/src/traffic/parse_tc.rs b/traffic_core/src/traffic/parse_tc.rs new file mode 100644 index 0000000..f77af71 --- /dev/null +++ b/traffic_core/src/traffic/parse_tc.rs @@ -0,0 +1,397 @@ +//! Parsers for `tc` and `ip` command output. +//! +//! Everything here is pure: text in, model out, no transport and no side effects. That is +//! what makes this the tested part of the traffic module. + +use std::{ + collections::{BTreeMap, BTreeSet}, + net::Ipv4Addr, +}; + +use super::{ClassRateCeil, FilterInspect}; + +/// Maps peer IP to class id from `tc filter show`, and reports both kinds of duplicate. +/// +/// Parsing is block-based: a new block starts at every line beginning with `filter`, and +/// the IP and flowid are taken from within one block. Scanning tokens across unrelated +/// lines would happily pair an IP from one filter with a flowid from the next. +pub(super) fn extract_filter_ip_class_map(filters: &str) -> FilterInspect { + let mut out = BTreeMap::new(); + let mut duplicate_ips = BTreeSet::new(); + let mut class_to_ip = BTreeMap::new(); + let mut duplicate_classes = BTreeSet::new(); + + for block in split_filter_blocks(filters) { + let mut ip: Option = None; + let mut class_id: Option = None; + + for line in block.lines() { + let t = line.trim(); + if t.contains("match") && ip.is_none() { + ip = extract_u32_match_hex(t).and_then(|hex| hex_to_ipv4_string(&hex)); + } + if t.contains("flowid") && class_id.is_none() { + class_id = extract_flowid_class_id(t); + } + } + + if let (Some(ip), Some(class_id)) = (ip, class_id) { + if out.contains_key(&ip) { + duplicate_ips.insert(ip.clone()); + } + if class_to_ip.insert(class_id, ip.clone()).is_some() { + duplicate_classes.insert(class_id); + } + out.insert(ip, class_id); + } + } + + FilterInspect { + map: out, + duplicate_ips, + duplicate_classes, + } +} + +pub(super) fn has_fq_codel_for_parent(raw_qdisc: &str, parent_class: &str) -> bool { + raw_qdisc.lines().map(str::trim).any(|line| { + line.contains("fq_codel") && line.contains("parent") && line.contains(parent_class) + }) +} + +pub(super) fn has_ingress_redirect_to_ifb(output: &str, ifb_name: &str) -> bool { + let ifb_lower = ifb_name.to_ascii_lowercase(); + let needle_dev = format!("dev {ifb_lower}"); + let needle_device = format!("device {ifb_lower}"); + + output.lines().any(|line| { + let l = line.to_ascii_lowercase(); + l.contains("mirred") + && l.contains("redirect") + && (l.contains(&needle_dev) || l.contains(&needle_device)) + }) +} + +pub(super) fn link_is_up(output: &str) -> bool { + output.contains("state UP") || (output.contains('<') && output.contains("UP")) +} + +pub(super) fn is_not_found_error(stderr_or_msg: &str) -> bool { + let s = stderr_or_msg.to_ascii_lowercase(); + s.contains("does not exist") + || s.contains("not found") + || s.contains("cannot find device") + || s.contains("no such device") + || s.contains("no such file") + || s.contains("handle of zero") +} + +pub(super) fn parse_root_class_rate_ceil(raw: &str, class_id: &str) -> Option { + for line in raw.lines() { + let t = line.trim(); + if !t.contains("class htb") || !t.contains(" rate ") || !t.contains(" ceil ") { + continue; + } + let toks: Vec<&str> = t.split_whitespace().collect(); + let class = toks + .iter() + .position(|v| *v == "htb") + .and_then(|i| toks.get(i + 1))?; + if *class != class_id { + continue; + } + let rate = toks + .iter() + .position(|v| *v == "rate") + .and_then(|i| toks.get(i + 1)) + .map(|s| s.to_string())?; + let ceil = toks + .iter() + .position(|v| *v == "ceil") + .and_then(|i| toks.get(i + 1)) + .map(|s| s.to_string())?; + return Some(ClassRateCeil { rate, ceil }); + } + None +} + +pub(super) fn parse_class_rate_ceil_map( + raw: &str, + prefix: &str, +) -> BTreeMap { + let mut out = BTreeMap::new(); + for line in raw.lines() { + let t = line.trim(); + if !t.contains("class htb") || !t.contains(" rate ") || !t.contains(" ceil ") { + continue; + } + let toks: Vec<&str> = t.split_whitespace().collect(); + let class = toks + .iter() + .position(|v| *v == "htb") + .and_then(|i| toks.get(i + 1)) + .map(|s| s.to_string()); + let rate = toks + .iter() + .position(|v| *v == "rate") + .and_then(|i| toks.get(i + 1)) + .map(|s| s.to_string()); + let ceil = toks + .iter() + .position(|v| *v == "ceil") + .and_then(|i| toks.get(i + 1)) + .map(|s| s.to_string()); + if let (Some(class), Some(rate), Some(ceil)) = (class, rate, ceil) + && class.starts_with(prefix) + && !class.ends_with(":1") + { + out.insert(class, ClassRateCeil { rate, ceil }); + } + } + out +} + +pub(super) fn collect_leaf_class_ids(raw: &str) -> BTreeSet { + let mut out = BTreeSet::new(); + for l in raw.lines() { + let line = l.trim(); + let Some(pos) = line.find("class htb ") else { + continue; + }; + let class = &line[(pos + "class htb ".len())..]; + let Some(head) = class.split_whitespace().next() else { + continue; + }; + if let Some(class_id) = head.split(':').nth(1).and_then(|v| v.parse::().ok()) + && class_id != 1 + { + out.insert(class_id); + } + } + out +} + +/// Compares rate and ceil numerically, so `1Gbit` and `1000Mbit` are the same limit. +pub(super) fn rate_ceil_matches_limit(v: &ClassRateCeil, limit: &str) -> bool { + let Ok(limit_bps) = parse_tc_rate_bps(limit) else { + return false; + }; + let Ok(rate_bps) = parse_tc_rate_bps(&v.rate) else { + return false; + }; + let Ok(ceil_bps) = parse_tc_rate_bps(&v.ceil) else { + return false; + }; + rate_bps == limit_bps && ceil_bps == limit_bps +} + +pub(super) fn normalize_tc_rate(v: &str) -> String { + v.trim().to_ascii_lowercase().replace(' ', "") +} + +fn split_filter_blocks(raw: &str) -> Vec { + let mut blocks = Vec::new(); + let mut cur = Vec::new(); + + for line in raw.lines() { + let t = line.trim(); + if t.starts_with("filter") && !cur.is_empty() { + blocks.push(cur.join("\n")); + cur.clear(); + } + if !t.is_empty() { + cur.push(t.to_string()); + } + } + if !cur.is_empty() { + blocks.push(cur.join("\n")); + } + + blocks +} + +fn extract_flowid_class_id(line: &str) -> Option { + let toks: Vec<&str> = line.split_whitespace().collect(); + let flowid = toks + .iter() + .position(|v| *v == "flowid") + .and_then(|i| toks.get(i + 1))?; + let class_tail = flowid.split(':').nth(1)?; + class_tail.parse::().ok() +} + +fn extract_u32_match_hex(line: &str) -> Option { + 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 Some(hex.to_ascii_lowercase()); + } + None +} + +fn hex_to_ipv4_string(hex: &str) -> Option { + if 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(Ipv4Addr::new(b1, b2, b3, b4).to_string()) +} + +fn parse_tc_rate_bps(v: &str) -> anyhow::Result { + let n = normalize_tc_rate(v); + let (num, mul) = if let Some(x) = n.strip_suffix("gbit") { + (x, 1_000_000_000u64) + } else if let Some(x) = n.strip_suffix("mbit") { + (x, 1_000_000u64) + } else if let Some(x) = n.strip_suffix("kbit") { + (x, 1_000u64) + } else if let Some(x) = n.strip_suffix("bit") { + (x, 1u64) + } else { + anyhow::bail!("unknown tc rate unit: {v}") + }; + let base: u64 = num + .parse() + .map_err(|_| anyhow::anyhow!("invalid tc rate: {v}"))?; + base.checked_mul(mul) + .ok_or_else(|| anyhow::anyhow!("tc rate overflow: {v}")) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_tc_u32_hex_ipv4() { + let raw = " match 0a080105/ffffffff at 16\n flowid 1:42"; + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_ips.is_empty()); + assert!(inspect.duplicate_classes.is_empty()); + assert_eq!(inspect.map.get("10.8.1.5").copied(), Some(42)); + } + + #[test] + fn parse_block_based_filters() { + let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:11"; + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_ips.is_empty()); + assert!(inspect.duplicate_classes.is_empty()); + assert_eq!(inspect.map.get("10.8.1.2").copied(), Some(10)); + assert_eq!(inspect.map.get("10.8.1.3").copied(), Some(11)); + } + + #[test] + fn detect_duplicate_filter_ip() { + let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080102/ffffffff at 16\n flowid 1:10"; + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_ips.contains("10.8.1.2")); + } + + #[test] + fn detect_duplicate_filter_class_flowid() { + let raw = "filter parent 1: protocol ip pref 1 u32\n match 0a080102/ffffffff at 16\n flowid 1:10\nfilter parent 1: protocol ip pref 2 u32\n match 0a080103/ffffffff at 16\n flowid 1:10"; + let inspect = extract_filter_ip_class_map(raw); + assert!(inspect.duplicate_classes.contains(&10)); + } + + #[test] + fn detect_fq_codel_parent() { + let raw = "qdisc fq_codel 8001: dev wg0 parent 1:10 limit 10240p"; + assert!(has_fq_codel_for_parent(raw, "1:10")); + assert!(!has_fq_codel_for_parent(raw, "1:11")); + } + + #[test] + fn detects_missing_ingress_redirect() { + let raw = "filter parent ffff: protocol all pref 1 u32"; + assert!(!has_ingress_redirect_to_ifb(raw, "ifb0")); + } + + #[test] + fn detects_ingress_redirect_to_ifb() { + let raw = "filter parent ffff: protocol all pref 1 u32\n action order 1: mirred (Egress Redirect to device ifb0) stolen"; + assert!(has_ingress_redirect_to_ifb(raw, "ifb0")); + } + + #[test] + fn detects_ingress_redirect_to_ifb_dev_format_lowercase() { + let raw = "filter parent ffff: protocol all pref 1 u32\n action order 1: mirred egress redirect dev ifb0"; + assert!(has_ingress_redirect_to_ifb(raw, "ifb0")); + } + + #[test] + fn does_not_match_redirect_to_other_ifb() { + let raw = "filter parent ffff: protocol all pref 1 u32\n action order 1: mirred (Egress Redirect to device ifb1) stolen"; + assert!(!has_ingress_redirect_to_ifb(raw, "ifb0")); + } + + #[test] + fn detects_ifb_up_state() { + assert!(link_is_up( + "2: ifb0: mtu 1500 qdisc fq_codel state UNKNOWN" + )); + assert!(link_is_up("2: ifb0: mtu 1500 state UP")); + assert!(!link_is_up( + "2: ifb0: mtu 1500 state DOWN" + )); + } + + #[test] + fn detects_not_found_errors() { + assert!(is_not_found_error("RTNETLINK answers: No such device")); + assert!(is_not_found_error("Cannot find device \"ifb0\"")); + assert!(is_not_found_error("does not exist")); + assert!(is_not_found_error( + "Cannot delete qdisc with handle of zero" + )); + assert!(!is_not_found_error("permission denied")); + } + + #[test] + fn parse_root_rate_ceil() { + let raw = "class htb 1:10 parent 1:1 leaf 800f: prio 0 rate 50Mbit ceil 50Mbit burst 1600b cburst 1600b\nclass htb 1:1 root rate 500Mbit ceil 500Mbit burst 1600b cburst 1600b"; + let rc = parse_root_class_rate_ceil(raw, "1:1").expect("root class exists"); + assert_eq!(normalize_tc_rate(&rc.rate), "500mbit"); + assert_eq!(normalize_tc_rate(&rc.ceil), "500mbit"); + } + + #[test] + fn parse_root_rate_ceil_returns_none_when_only_parent_matches() { + let raw = "class htb 1:10 parent 1:1 leaf 800f: prio 0 rate 50Mbit ceil 50Mbit burst 1600b cburst 1600b"; + assert!(parse_root_class_rate_ceil(raw, "1:1").is_none()); + } + + #[test] + fn parse_class_rate_ceil() { + let raw = "class htb 1:10 parent 1:1 rate 70Mbit ceil 70Mbit burst 1599b cburst 1599b\nclass htb 1:1 root rate 1000Mbit ceil 1000Mbit"; + let m = parse_class_rate_ceil_map(raw, "1:"); + let rc = m.get("1:10").expect("class exists"); + assert_eq!(normalize_tc_rate(&rc.rate), "70mbit"); + assert_eq!(normalize_tc_rate(&rc.ceil), "70mbit"); + } + + #[test] + fn rate_limit_match_normalized() { + let rc = ClassRateCeil { + rate: "70Mbit".to_string(), + ceil: "70mbit".to_string(), + }; + assert!(rate_ceil_matches_limit(&rc, "70mbit")); + assert!(!rate_ceil_matches_limit(&rc, "50mbit")); + } + + #[test] + fn tc_rate_numeric_equivalence() { + let rc = ClassRateCeil { + rate: "1Gbit".to_string(), + ceil: "1000Mbit".to_string(), + }; + assert!(rate_ceil_matches_limit(&rc, "1000mbit")); + assert!(rate_ceil_matches_limit(&rc, "1gbit")); + } +}