prod/AWG2_harness
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Полный продакшн-фикс: hardened build/install, drift/firewall/wg проверки и актуализация документации

Browse Source
This commit is contained in:
Crimson
2026-05-10 15:41:44 +05:00
parent 535c1dbc33
commit b3dad5974c
11 changed files with 410 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
traffic_core/src/firewall.rs
+76 -15
View File
@@ -3,6 +3,7 @@ use crate::{
docker,
netns::{NsTransport, build_host_transport, build_transport},
};
use serde_json::Value;
#[derive(Debug, Clone)]
struct FirewallActual {
@@ -77,7 +78,14 @@ fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result
"ct count over {}",
cfg.anti_torrent.max_connections_per_peer
);
let rule_matches = table.contains(&needle_cidr) && table.contains(&needle_max);
let rule_matches = table.contains(&needle_cidr)
&& table.contains(&needle_max)
&& table.contains("ct state new")
&& table.contains("meter awg2_peer_connlimit_v4")
&& table.contains("counter name antitorrent_dropped_v4")
&& table.contains("drop")
&& table.contains("hook forward")
&& table.contains("policy accept");
Ok(FirewallActual {
table_exists,
@@ -90,7 +98,10 @@ fn inspect_firewall(transport: &dyn NsTransport, cfg: &Config) -> anyhow::Result
fn firewall_transport(cfg: &Config) -> anyhow::Result<Box<dyn NsTransport>> {
if cfg.anti_torrent.apply_in_amnezia_netns {
let target = docker::discover_target(cfg)?;
Ok(build_transport(target.container_pid, &target.container_name))
Ok(build_transport(
target.container_pid,
&target.container_name,
))
} else {
Ok(build_host_transport())
}
@@ -180,25 +191,75 @@ pub fn antitorrent_drop_counter(cfg: &Config) -> anyhow::Result<Option<(u64, u64
}
let transport = firewall_transport(cfg)?;
let json = transport.run(&[
"nft",
"-j",
"list",
"counter",
"inet",
"awg2_harness",
"antitorrent_dropped_v4",
]);
if let Ok(raw) = json {
if let Some(v) = parse_counter_json(&raw)? {
return Ok(Some(v));
}
}
let table = transport
.run(&["nft", "list", "table", "inet", "awg2_harness"])
.unwrap_or_default();
for line in table.lines() {
if line.contains("counter antitorrent_dropped_v4") && line.contains("packets") {
let toks: Vec<&str> = line.split_whitespace().collect();
if let Some(idx) = toks.iter().position(|t| *t == "packets") {
if let Some(v) = toks.get(idx + 1).and_then(|s| s.parse::<u64>().ok()) {
let bytes = toks
.iter()
.position(|t| *t == "bytes")
.and_then(|i| toks.get(i + 1))
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0);
return Ok(Some((v, bytes)));
}
let t = line.trim();
if !t.contains("packets") {
continue;
}
let toks: Vec<&str> = t.split_whitespace().collect();
let packets = toks
.iter()
.position(|x| *x == "packets")
.and_then(|i| toks.get(i + 1))
.and_then(|s| s.parse::<u64>().ok());
let bytes = toks
.iter()
.position(|x| *x == "bytes")
.and_then(|i| toks.get(i + 1))
.and_then(|s| s.parse::<u64>().ok())
.unwrap_or(0);
if let Some(p) = packets {
return Ok(Some((p, bytes)));
}
}
Ok(None)
}
fn parse_counter_json(raw: &str) -> anyhow::Result<Option<(u64, u64)>> {
let v: Value = serde_json::from_str(raw)?;
let Some(arr) = v.get("nftables").and_then(|x| x.as_array()) else {
return Ok(None);
};
for item in arr {
if let Some(counter) = item.get("counter") {
let packets = counter.get("packets").and_then(|x| x.as_u64());
let bytes = counter.get("bytes").and_then(|x| x.as_u64());
if let (Some(p), Some(b)) = (packets, bytes) {
return Ok(Some((p, b)));
}
}
}
Ok(None)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn parse_nft_counter_json() {
let raw = r#"{"nftables":[{"metainfo":{"json_schema_version":1}},{"counter":{"family":"inet","table":"awg2_harness","name":"antitorrent_dropped_v4","handle":2,"packets":7,"bytes":448}}]}"#;
let got = parse_counter_json(raw).expect("json parse");
assert_eq!(got, Some((7, 448)));
}
}
traffic_core/src/netns.rs
+8 -2
View File
@@ -47,11 +47,17 @@ impl NsTransport for DockerExecTransport {
impl NsTransport for HostTransport {
fn run(&self, args: &[&str]) -> anyhow::Result<String> {
let mut cmd = Command::new(args.first().ok_or_else(|| anyhow::anyhow!("empty command"))?);
let mut cmd = Command::new(
args.first()
.ok_or_else(|| anyhow::anyhow!("empty command"))?,
);
cmd.args(&args[1..]);
let out = cmd.output()?;
if !out.status.success() {
anyhow::bail!("host transport failed: {}", String::from_utf8_lossy(&out.stderr));
anyhow::bail!(
"host transport failed: {}",
String::from_utf8_lossy(&out.stderr)
);
}
Ok(String::from_utf8_lossy(&out.stdout).to_string())
}
traffic_core/src/reboot.rs
+1 -1
View File
@@ -1,4 +1,4 @@
use chrono::{Datelike, Duration, NaiveDateTime, NaiveTime, Timelike, Utc};
use chrono::{Duration, NaiveDateTime, NaiveTime, Timelike, Utc};
use rand::Rng;
use std::{process::Command, thread, time::Duration as StdDuration};
traffic_core/src/traffic.rs
+187 -29
View File
@@ -58,6 +58,7 @@ pub fn inspect_drift_summary(cfg: &Config) -> anyhow::Result<Option<String>> {
&target.interface_name,
&cfg.traffic.ifb_name,
cfg.traffic.class_start_id,
&cfg.traffic.root_rate,
&cfg.traffic.peer_limit,
&peers,
)?;
@@ -93,6 +94,7 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
&target.interface_name,
&cfg.traffic.ifb_name,
cfg.traffic.class_start_id,
&cfg.traffic.root_rate,
&cfg.traffic.peer_limit,
&peers,
)?;
@@ -104,21 +106,26 @@ pub fn reconcile_traffic(cfg: &Config) -> anyhow::Result<()> {
println!("traffic drift detected: {}", drift.reason);
if cfg.traffic.full_rebuild_on_drift {
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,
)?;
if !cfg.traffic.full_rebuild_on_drift {
anyhow::bail!(
"traffic drift detected but full_rebuild_on_drift=false: {}",
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(())
}
@@ -139,10 +146,26 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
}
let filters_egress = transport
.run(&["tc", "filter", "show", "dev", &target.interface_name, "parent", "1:"])
.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:"])
.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])
@@ -151,8 +174,8 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
.run(&["tc", "class", "show", "dev", &cfg.traffic.ifb_name])
.unwrap_or_default();
let actual_map_egress = extract_filter_ip_class_map(&filters_egress);
let actual_map_ingress = extract_filter_ip_class_map(&filters_ingress);
let (actual_map_egress, dup_egress) = extract_filter_ip_class_map(&filters_egress);
let (actual_map_ingress, dup_ingress) = extract_filter_ip_class_map(&filters_ingress);
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);
@@ -193,6 +216,12 @@ pub fn peer_status_table(cfg: &Config) -> anyhow::Result<Option<Vec<PeerTrafficS
{
reasons.push("ingress rate/ceil mismatch".to_string());
}
if dup_egress.contains(&ip) {
reasons.push("duplicate egress filters".to_string());
}
if dup_ingress.contains(&ip) {
reasons.push("duplicate ingress filters".to_string());
}
out.push(PeerTrafficStatus {
ip,
@@ -219,6 +248,7 @@ fn inspect_and_diff(
iface: &str,
ifb: &str,
class_start_id: u16,
root_rate: &str,
peer_limit: &str,
peers: &[wg::Peer],
) -> anyhow::Result<TrafficDrift> {
@@ -259,8 +289,26 @@ fn inspect_and_diff(
.run(&["tc", "filter", "show", "dev", ifb, "parent", "2:"])
.unwrap_or_default();
let actual_map_egress = extract_filter_ip_class_map(&filters_egress);
let actual_map_ingress = extract_filter_ip_class_map(&filters_ingress);
let (actual_map_egress, dup_egress) = extract_filter_ip_class_map(&filters_egress);
let (actual_map_ingress, dup_ingress) = extract_filter_ip_class_map(&filters_ingress);
if !dup_egress.is_empty() {
return Ok(TrafficDrift {
has_drift: true,
reason: format!(
"duplicate egress filters detected for {} peer(s)",
dup_egress.len()
),
});
}
if !dup_ingress.is_empty() {
return Ok(TrafficDrift {
has_drift: true,
reason: format!(
"duplicate ingress filters detected for {} peer(s)",
dup_ingress.len()
),
});
}
let actual_ips_egress: BTreeSet<String> = actual_map_egress.keys().cloned().collect();
let actual_ips_ingress: BTreeSet<String> = actual_map_ingress.keys().cloned().collect();
let expected_ips: BTreeSet<String> = peers.iter().map(|p| p.ipv4.to_string()).collect();
@@ -316,9 +364,33 @@ fn inspect_and_diff(
let classes_ingress_full = transport
.run(&["tc", "class", "show", "dev", ifb])
.unwrap_or_default();
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}");
@@ -330,8 +402,10 @@ fn inspect_and_diff(
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)
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,
@@ -347,27 +421,51 @@ fn inspect_and_diff(
.run(&["tc", "class", "show", "dev", ifb, "parent", "2:1"])
.unwrap_or_default();
let expected_class_count = peers.len();
if count_leaf_classes(&classes_egress) < expected_class_count {
if count_leaf_classes(&classes_egress) != expected_class_count {
return Ok(TrafficDrift {
has_drift: true,
reason: "missing egress per-peer classes".to_string(),
reason: "egress per-peer class count mismatch".to_string(),
});
}
if count_leaf_classes(&classes_ingress) < expected_class_count {
if count_leaf_classes(&classes_ingress) != expected_class_count {
return Ok(TrafficDrift {
has_drift: true,
reason: "missing ingress per-peer classes".to_string(),
reason: "ingress per-peer class count mismatch".to_string(),
});
}
let qdisc_egress = transport
.run(&["tc", "qdisc", "show", "dev", iface])
.unwrap_or_default();
let qdisc_ingress = transport
.run(&["tc", "qdisc", "show", "dev", ifb])
.unwrap_or_default();
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 extract_filter_ip_class_map(filters: &str) -> BTreeMap<String, u16> {
fn extract_filter_ip_class_map(filters: &str) -> (BTreeMap<String, u16>, BTreeSet<String>) {
let mut out = BTreeMap::new();
let mut duplicates = BTreeSet::new();
for block in split_filter_blocks(filters) {
let mut ip: Option<String> = None;
@@ -384,11 +482,46 @@ fn extract_filter_ip_class_map(filters: &str) -> BTreeMap<String, u16> {
}
if let (Some(ip), Some(class_id)) = (ip, class_id) {
if out.contains_key(&ip) {
duplicates.insert(ip.clone());
}
out.insert(ip, class_id);
}
}
out
(out, duplicates)
}
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 parse_root_class_rate_ceil(raw: &str, class_id: &str) -> Option<ClassRateCeil> {
for line in raw.lines() {
let t = line.trim();
if !t.contains("class htb")
|| !t.contains(class_id)
|| !t.contains(" rate ")
|| !t.contains(" ceil ")
{
continue;
}
let toks: Vec<&str> = t.split_whitespace().collect();
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<String> {
@@ -649,6 +782,7 @@ fn apply_full_topology(
iface,
ifb,
cfg.traffic.class_start_id,
&cfg.traffic.root_rate,
&cfg.traffic.peer_limit,
peers,
)?;
@@ -742,18 +876,42 @@ mod tests {
#[test]
fn parse_tc_u32_hex_ipv4() {
let raw = " match 0a080105/ffffffff at 16\n flowid 1:42";
let m = extract_filter_ip_class_map(raw);
let (m, dup) = extract_filter_ip_class_map(raw);
assert!(dup.is_empty());
assert_eq!(m.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 m = extract_filter_ip_class_map(raw);
let (m, dup) = extract_filter_ip_class_map(raw);
assert!(dup.is_empty());
assert_eq!(m.get("10.8.1.2").copied(), Some(10));
assert_eq!(m.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 (_m, dup) = extract_filter_ip_class_map(raw);
assert!(dup.contains("10.8.1.2"));
}
#[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 parse_root_rate_ceil() {
let raw = "class htb 1:1 root rate 1000Mbit ceil 1000Mbit";
let rc = parse_root_class_rate_ceil(raw, "1:1").expect("root class exists");
assert_eq!(normalize_tc_rate(&rc.rate), "1000mbit");
assert_eq!(normalize_tc_rate(&rc.ceil), "1000mbit");
}
#[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";
traffic_core/src/wg.rs
+4 -2
View File
@@ -28,11 +28,13 @@ pub fn discover_peers_with_transport(
) -> anyhow::Result<Vec<Peer>> {
let primary = transport
.run(&["wg", "show", "all", "allowed-ips"])
.unwrap_or_default();
.map_err(|e| anyhow::anyhow!("wg show all allowed-ips failed: {e}"))?;
let mut peers = parse_allowed_ips(&primary, peer_cidr)?;
if peers.is_empty() {
let fallback = transport.run(&["wg", "show"]).unwrap_or_default();
let fallback = transport
.run(&["wg", "show"])
.map_err(|e| anyhow::anyhow!("wg show fallback failed: {e}"))?;
peers = parse_fallback_wg_show(&fallback, peer_cidr)?;
}