diff --git a/.gitignore b/.gitignore index 0ba5eba..d50bce2 100644 --- a/.gitignore +++ b/.gitignore @@ -41,7 +41,7 @@ Thumbs.db .project .settings/ .factorypath -bin/ +/bin/ # Reference projects (cloned for research) ref/ diff --git a/src/bin/rampart-cli.rs b/src/bin/rampart-cli.rs new file mode 100644 index 0000000..a508672 --- /dev/null +++ b/src/bin/rampart-cli.rs @@ -0,0 +1,72 @@ +#![allow(clippy::print_stdout, clippy::print_stderr)] + +use clap::{Parser, Subcommand}; +use rampart::cli::commands; + +#[derive(Parser)] +#[command(name = "rampart-cli", about = "Rampart management CLI")] +struct Cli { + #[command(subcommand)] + command: Commands, +} + +#[derive(Subcommand)] +enum Commands { + /// Show overall system status + Status, + /// Run full diagnostics + Doctor, + /// Get/set configuration + Config { + #[arg(required = false)] + key: Option, + #[arg(required = false)] + value: Option, + }, + /// Manage blacklist + Blacklist { + #[command(subcommand)] + action: BlacklistAction, + }, + /// Emergency mode + Emergency { + #[arg(value_enum)] + mode: EmergencyMode, + }, + /// Gracefully drain a node + Drain { node: String }, +} + +#[derive(Subcommand)] +enum BlacklistAction { + Add { target: String, reason: Option }, + Remove { target: String }, + List, +} + +#[derive(clap::ValueEnum, Clone)] +enum EmergencyMode { + Enable, + Disable, +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let cli = Cli::parse(); + + match cli.command { + Commands::Status => commands::status::run().await, + Commands::Doctor => commands::doctor::run().await, + Commands::Config { key, value } => commands::config::run(key, value).await, + Commands::Blacklist { action } => match action { + BlacklistAction::Add { target, reason } => commands::blacklist::add(target, reason).await, + BlacklistAction::Remove { target } => commands::blacklist::remove(target).await, + BlacklistAction::List => commands::blacklist::list().await, + }, + Commands::Emergency { mode } => match mode { + EmergencyMode::Enable => commands::emergency::enable().await, + EmergencyMode::Disable => commands::emergency::disable().await, + }, + Commands::Drain { node } => commands::drain::run(&node).await, + } +} diff --git a/src/bin/rampart-manager.rs b/src/bin/rampart-manager.rs new file mode 100644 index 0000000..f48f235 --- /dev/null +++ b/src/bin/rampart-manager.rs @@ -0,0 +1,79 @@ +use axum::{ + Router, + http::HeaderValue, + middleware, + routing::{get, post}, +}; +use dashmap::DashMap; +use rampart::manager::{AppState, api, auth, sync}; +use std::sync::Arc; +use tower_http::cors::CorsLayer; +use tracing_subscriber::EnvFilter; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env().add_directive("rampart=info".parse()?)) + .init(); + + let redis_url = std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/0".to_string()); + let redis_client = redis::Client::open(redis_url)?; + + let jwt_secret = std::env::var("JWT_SECRET").map_err(|_| anyhow::anyhow!("JWT_SECRET must be set"))?; + if jwt_secret.len() < 32 { + return Err(anyhow::anyhow!("JWT_SECRET must be at least 32 bytes")); + } + let jwt_audience = std::env::var("JWT_AUDIENCE").unwrap_or_else(|_| "rampart".to_string()); + let jwt_expiration = std::env::var("JWT_EXPIRATION_SECS") + .unwrap_or_else(|_| "86400".to_string()) + .parse::() + .map_err(|_| anyhow::anyhow!("JWT_EXPIRATION_SECS must be a valid u64"))?; + + let api_password = std::env::var("API_PASSWORD").map_err(|_| anyhow::anyhow!("API_PASSWORD must be set"))?; + if api_password == "changeme" { + return Err(anyhow::anyhow!("API_PASSWORD must not be the default 'changeme'")); + } + + let state = Arc::new(AppState { + redis_client, + jwt_secret, + jwt_audience, + jwt_expiration, + api_password, + login_limiter: DashMap::new(), + }); + + tokio::spawn(sync::heartbeat::start_heartbeat_check(state.clone())); + + let public = Router::new() + .route("/api/v1/health", get(api::health::health_check)) + .route("/api/v1/auth/login", post(api::auth::login)); + + let protected = Router::new() + .route("/api/v1/servers", get(api::servers::list_servers)) + .route( + "/api/v1/blacklist", + get(api::blacklist::list_blacklist).post(api::blacklist::add_blacklist), + ) + .route("/api/v1/nodes", get(api::nodes::list_nodes)) + .route_layer(middleware::from_fn(auth::auth_middleware)); + + let cors = match std::env::var("CORS_ORIGIN") { + Ok(origin) if origin.is_empty() || origin == "*" => CorsLayer::new().allow_origin(tower_http::cors::Any), + Ok(origin) => CorsLayer::new().allow_origin(HeaderValue::from_str(&origin)?), + Err(_) => CorsLayer::new().allow_origin(HeaderValue::from_static("http://localhost:5173")), + }; + + let app = Router::new() + .merge(public) + .merge(protected) + .layer(cors) + .with_state(state); + + let addr = "0.0.0.0:8080"; + tracing::info!("Manager API listening on {addr}"); + let listener = tokio::net::TcpListener::bind(addr).await?; + let app = app.into_make_service_with_connect_info::(); + axum::serve(listener, app).await?; + Ok(()) +} diff --git a/src/bin/rampart.rs b/src/bin/rampart.rs new file mode 100644 index 0000000..0c5e506 --- /dev/null +++ b/src/bin/rampart.rs @@ -0,0 +1,311 @@ +use rampart::config::Config; +use rampart::engine::challenge::DifficultyAdjuster; +use rampart::engine::listener; +use rampart::engine::subnet_monitor; +use rampart::engine::subnet_tracker::SubnetTracker; +use rampart::engine::tunnel::Gateway; +use rampart::filter::blacklist::Blacklist; +use rampart::filter::rate_limit::RateLimiter; +use rampart::metrics; +use rampart::protocol::ProtocolRegistry; +use rampart::store::clickhouse::{ClickHouseEvent, ClickHouseWriter}; +use rampart::traffic::alert::{AlertDispatcher, send_webhook}; +use rampart::traffic::detector::{AttackDetector, AttackStatus}; +use rampart::traffic::hook::TrafficHook; +use rampart::traffic::reputation::IpReputation; +use rampart::xdp::XdpFilter; +use std::collections::HashSet; +use std::net::IpAddr; +use std::sync::Arc; +use std::sync::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; +use tokio::sync::watch; +use tracing_subscriber::EnvFilter; + +fn attack_status_value(status: AttackStatus) -> i64 { + match status { + AttackStatus::Normal => 0, + AttackStatus::Suspicious => 1, + AttackStatus::UnderAttack => 2, + } +} + +/// Собирает реестр протоколов из скомпилированных реализаций. +/// +/// В ядре реализаций нет: `protocol-http` — зарезервированная пустая feature, +/// показывающая модель plugin-by-feature. Внешние крейты регистрируют свои +/// обработчики в этом месте. +#[allow(unused_mut)] +fn build_registry() -> ProtocolRegistry { + let mut registry = ProtocolRegistry::new(); + #[cfg(feature = "protocol-http")] + register_http(&mut registry); + registry +} + +#[cfg(feature = "protocol-http")] +fn register_http(_registry: &mut ProtocolRegistry) {} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env().add_directive("rampart=info".parse()?)) + .init(); + + let registry = Arc::new(build_registry()); + if registry.is_empty() { + anyhow::bail!( + "no protocol plugins compiled; build with --features protocol-http \ + or link an external ProtocolHandler implementation" + ); + } + + let config_path = std::env::var("RAMPART_CONFIG").unwrap_or_else(|_| "/etc/rampart/config.toml".to_string()); + let config = Arc::new(Config::from_file(&config_path)?); + let whitelist = build_whitelist(&config)?; + + let rate_limiter = Arc::new(RateLimiter::new( + config.limits.rate_limit_pps, + config.limits.rate_limit_burst, + )); + let blacklist = Arc::new(Blacklist::new()); + let reputation = Arc::new(IpReputation::new()); + let detector = Arc::new(Mutex::new(AttackDetector::new())); + let hook = Arc::new(TrafficHook::new( + reputation.clone(), + blacklist.clone(), + config.detect.autoban.clone(), + config.ban.ban_duration_secs, + )); + let alert_dispatcher = AlertDispatcher::new(); + let webhook_url = config.detect.alert.webhook_url.clone(); + let allowed_1s = Arc::new(AtomicU64::new(0)); + + let (shutdown_tx, shutdown_rx) = watch::channel(false); + + let sig_tx = shutdown_tx.clone(); + tokio::spawn(async move { + wait_for_signal().await; + tracing::info!("shutdown signal received, draining connections..."); + let _ = sig_tx.send(true); + tokio::time::sleep(Duration::from_secs(5)).await; + tracing::info!("shutdown timeout reached, exiting"); + std::process::exit(0); + }); + + #[cfg(feature = "store-redis")] + if let Some(redis_url) = &config.store.redis_url + && !redis_url.is_empty() + { + let bl = blacklist.clone(); + let sd = shutdown_rx.clone(); + let url = redis_url.clone(); + tokio::spawn(async move { + let client = match redis::Client::open(url.as_str()) { + Ok(c) => c, + Err(e) => { + tracing::warn!("Invalid redis_url: {e}, blacklist sync disabled"); + return; + }, + }; + rampart::store::start_blacklist_sync(&client, bl, sd).await; + }); + } + + if config.metrics.enabled { + let metrics_addr = format!("0.0.0.0:{}", config.metrics.port); + tracing::info!("Metrics server listening on {metrics_addr}"); + tokio::spawn(async move { + metrics::run_metrics_server(&metrics_addr).await; + }); + } + + let clickhouse: Option>> = match &config.store.clickhouse_url { + Some(url) if !url.is_empty() => { + let writer = Arc::new(tokio::sync::Mutex::new(ClickHouseWriter::new(url))); + rampart::store::clickhouse::start_flush_task(writer.clone(), shutdown_rx.clone()); + Some(writer) + }, + _ => None, + }; + + let xdp_filter = start_xdp(&config, &shutdown_rx)?; + let subnet_tracker = start_subnet_tracker(&config); + subnet_monitor::spawn( + &config.detect.prefix, + subnet_tracker.clone(), + xdp_filter.clone(), + shutdown_rx.clone(), + ); + + let rl = rate_limiter.clone(); + let bl = blacklist.clone(); + let det = detector.clone(); + let intel = hook.clone(); + let a1s = allowed_1s.clone(); + let ch = clickhouse.clone(); + let mut sd = shutdown_rx.clone(); + tokio::spawn(async move { + let mut sec_tick = tokio::time::interval(Duration::from_secs(1)); + let mut min_tick = tokio::time::interval(Duration::from_secs(60)); + sec_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + min_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + biased; + _ = sd.changed() => { + if *sd.borrow() { + return; + } + } + _ = sec_tick.tick() => { + let pps = a1s.swap(0, Ordering::Relaxed) as f64; + let cps = intel.take_cps() as f64; + metrics::INTEL_CPS.set(cps as i64); + let status = det.lock().expect("detector lock poisoned").analyze(pps, cps); + metrics::ATTACK_STATUS.set(attack_status_value(status)); + if let Some(alert) = alert_dispatcher.on_status(status) { + // Дедупликация: алерт только на переходе состояния атаки. + metrics::INTEL_ALERTS_TOTAL.inc(); + tracing::warn!(pps, cps, "{alert}"); + push_attack_event(&ch, pps).await; + if let Some(url) = webhook_url.clone() { + tokio::spawn(send_webhook(url, alert.clone())); + } + } + if status == AttackStatus::UnderAttack { + let banned = intel.ban_offenders(); + if banned > 0 { + tracing::info!(banned, "escalation: offenders auto-banned during attack"); + } + } + } + _ = min_tick.tick() => { + rl.sweep(); + bl.clear_expired(); + } + } + } + }); + + tracing::info!("Rampart edge starting on {}:{}", config.bind.address, config.bind.port); + tracing::info!("Upstreams: {:?}", config.backend.upstreams); + + let adjuster = Arc::new(Mutex::new(DifficultyAdjuster::default())); + let gateway = Arc::new(Gateway::new( + config.clone(), + rate_limiter, + blacklist, + adjuster, + whitelist, + reputation, + xdp_filter, + clickhouse, + allowed_1s, + registry, + subnet_tracker, + )); + listener::run(config, gateway, hook, shutdown_rx).await +} + +async fn push_attack_event(writer: &Option>>, pps: f64) { + let Some(writer) = writer else { + return; + }; + let event = ClickHouseEvent { + timestamp: chrono::Utc::now(), + event_type: "attack".to_string(), + ip: String::new(), + data_float: pps, + data_int: 0, + data_string: "under_attack".to_string(), + }; + if let Err(e) = writer.lock().await.push(event).await { + tracing::debug!("clickhouse push error: {e}"); + } +} + +#[cfg(feature = "xdp")] +fn start_xdp( + config: &Arc, + shutdown_rx: &watch::Receiver, +) -> anyhow::Result>>> { + use rampart::xdp::XdpMetrics; + + if !config.xdp.enabled { + return Ok(None); + } + + let filter = XdpFilter::new(&config.xdp.interface); + let shared = Arc::new(Mutex::new(filter)); + shared.lock().expect("xdp lock poisoned").load()?; + let xdp_metrics = XdpMetrics::register()?; + + let sd = shutdown_rx.clone(); + let shared_thread = shared.clone(); + std::thread::spawn(move || { + while !*sd.borrow() { + let guard = match shared_thread.lock() { + Ok(g) => g, + Err(_) => break, + }; + guard.drain_events(); + if let Ok(stats) = guard.get_stats() { + xdp_metrics.update(&stats); + } + drop(guard); + std::thread::sleep(Duration::from_secs(5)); + } + if let Ok(mut guard) = shared_thread.lock() { + guard.unload().ok(); + } + }); + Ok(Some(shared)) +} + +#[cfg(not(feature = "xdp"))] +fn start_xdp( + _config: &Arc, + _shutdown_rx: &watch::Receiver, +) -> anyhow::Result>>> { + Ok(None) +} + +/// Юзерспейс-агрегатор префиксов: активен, когда detect.prefix включён, +/// а XDP-путь не работает (не собран или выключен в конфиге). +fn start_subnet_tracker(config: &Config) -> Option> { + let xdp_active = cfg!(feature = "xdp") && config.xdp.enabled; + if config.detect.prefix.enabled && !xdp_active { + tracing::info!( + window_secs = config.detect.prefix.window_secs, + syn_threshold = config.detect.prefix.syn_threshold, + "subnet tracker enabled (userspace prefix aggregation)" + ); + Some(Arc::new(SubnetTracker::new())) + } else { + None + } +} + +fn build_whitelist(config: &Config) -> anyhow::Result>> { + let mut set = HashSet::with_capacity(config.whitelist.len()); + for entry in &config.whitelist { + let ip: IpAddr = entry + .parse() + .map_err(|_| anyhow::anyhow!("invalid whitelist entry: {entry}"))?; + set.insert(ip); + } + Ok(Arc::new(set)) +} + +async fn wait_for_signal() { + let ctrl_c = tokio::signal::ctrl_c(); + let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("failed to install SIGTERM handler"); + + tokio::select! { + _ = ctrl_c => {} + _ = term.recv() => {} + } +} diff --git a/src/cli/commands/doctor.rs b/src/cli/commands/doctor.rs index 81fe6c1..d773396 100644 --- a/src/cli/commands/doctor.rs +++ b/src/cli/commands/doctor.rs @@ -1,3 +1,62 @@ +use crate::xdp::{EnvironmentReport, FilesystemProbe, SystemProbe}; + +/// Ищет интерфейс с маршрутом по умолчанию в /proc/net/route +/// (destination == 00000000). Работает без root. +fn default_route_interface(probe: &impl SystemProbe) -> Option { + let routes = probe.read_file("/proc/net/route").ok()?; + routes.lines().skip(1).find_map(|line| { + let fields: Vec<&str> = line.split_whitespace().collect(); + (fields.len() > 7 && fields[1] == "00000000").then(|| fields[0].to_owned()) + }) +} + +/// Подсекция network/XDP: что видно окружению без root. +fn print_network_section(all_ok: &mut bool) { + println!("\nNetwork / XDP"); + println!("-------------"); + + let probe = FilesystemProbe; + let Some(interface) = default_route_interface(&probe) else { + println!("[WARN] Default route interface not found, XDP checks skipped"); + return; + }; + println!("Interface: {interface}"); + + let report = EnvironmentReport::collect(&probe, &interface); + let kernel = report + .kernel_version + .map_or_else(|| "unknown".to_owned(), |v| v.to_string()); + println!("Kernel: {kernel}"); + println!( + "BTF: {}", + if report.btf_available { + "available" + } else { + "missing (CO-RE disabled)" + } + ); + println!( + "Privileges: {}", + if report.privileged { + "root (capabilities assumed)" + } else { + "non-root" + } + ); + println!("Driver: {}", report.driver_name.as_deref().unwrap_or("unknown")); + println!("Attach mode: {}", report.attach_mode()); + println!("Verdict: {}", report.verdict()); + for warning in &report.warnings { + println!("[WARN] {warning}"); + } + if let Err(e) = report.validate() { + println!("[FAIL] {e}"); + *all_ok = false; + } else { + println!("[OK] XDP environment"); + } +} + pub async fn run() -> anyhow::Result<()> { println!("Rampart Diagnostics"); println!("=====================\n"); @@ -34,6 +93,8 @@ pub async fn run() -> anyhow::Result<()> { }, } + print_network_section(&mut all_ok); + println!(); if all_ok { println!("All checks passed."); diff --git a/src/config/mod.rs b/src/config/mod.rs index bdf3391..e902d8b 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -2,8 +2,8 @@ mod sections; pub use sections::{ - BackendConfig, BanConfig, BindConfig, DetectConfig, DetectPrefixConfig, LimitsConfig, LoggingConfig, MetricsConfig, - PowConfig, StoreConfig, WorkerConfig, XdpConfig, + BackendConfig, BanConfig, BindConfig, DetectAlertConfig, DetectAutobanConfig, DetectConfig, DetectPrefixConfig, + LimitsConfig, LoggingConfig, MetricsConfig, PowConfig, StoreConfig, WorkerConfig, XdpConfig, }; use serde::Deserialize; diff --git a/src/config/sections.rs b/src/config/sections.rs index 97e8535..7d42790 100644 --- a/src/config/sections.rs +++ b/src/config/sections.rs @@ -232,6 +232,42 @@ fn default_pow_difficulty() -> u8 { pub struct DetectConfig { #[serde(default)] pub prefix: DetectPrefixConfig, + #[serde(default)] + pub autoban: DetectAutobanConfig, + #[serde(default)] + pub alert: DetectAlertConfig, +} + +/// Секция `[detect.autoban]`: авто-бан источников по репутации и сигналам +/// детектора атак. Порог сравнивается со скором [`crate::traffic::reputation::IpReputation`], +/// TTL берётся из `ban.ban_duration_secs`. +#[derive(Debug, Clone, Deserialize)] +pub struct DetectAutobanConfig { + #[serde(default)] + pub enabled: bool, + #[serde(default = "default_autoban_reputation_threshold")] + pub reputation_threshold: i32, +} + +impl Default for DetectAutobanConfig { + fn default() -> Self { + Self { + enabled: false, + reputation_threshold: default_autoban_reputation_threshold(), + } + } +} + +fn default_autoban_reputation_threshold() -> i32 { + -50 +} + +/// Секция `[detect.alert]`: webhook-алерты на переходах состояния атаки +/// («атака началась» / «атака закончилась»). +#[derive(Debug, Clone, Default, Deserialize)] +pub struct DetectAlertConfig { + #[serde(default)] + pub webhook_url: Option, } /// Секция `[detect.prefix]`: subnet-level детектор распределённых атак. diff --git a/src/engine/listener.rs b/src/engine/listener.rs index 51ffe34..965ca20 100644 --- a/src/engine/listener.rs +++ b/src/engine/listener.rs @@ -2,6 +2,7 @@ use crate::config::Config; use crate::engine::tunnel::Gateway; +use crate::traffic::hook::TrafficHook; use socket2::{Domain, Socket, Type}; use std::sync::Arc; use tokio::net::TcpListener; @@ -12,8 +13,12 @@ use tokio::sync::watch; /// # Errors /// Возвращает ошибку при невозможности забиндить адрес или падении одного /// из accept-loop'ов. -pub async fn run(config: Arc, gateway: Arc, shutdown: watch::Receiver) -> anyhow::Result<()> { - let _ = &config; +pub async fn run( + config: Arc, + gateway: Arc, + hook: Arc, + shutdown: watch::Receiver, +) -> anyhow::Result<()> { let addr: std::net::SocketAddr = format!("{}:{}", config.bind.address, config.bind.port).parse()?; let workers = config.workers.count.max(1); @@ -23,6 +28,7 @@ pub async fn run(config: Arc, gateway: Arc, shutdown: watch::Re handles.push(tokio::spawn(accept_loop( listener, Arc::clone(&gateway), + Arc::clone(&hook), shutdown.clone(), ))); } @@ -43,7 +49,12 @@ fn build_listener(addr: std::net::SocketAddr) -> anyhow::Result { Ok(TcpListener::from_std(socket.into())?) } -async fn accept_loop(listener: TcpListener, gateway: Arc, mut shutdown: watch::Receiver) { +async fn accept_loop( + listener: TcpListener, + gateway: Arc, + hook: Arc, + mut shutdown: watch::Receiver, +) { loop { tokio::select! { biased; @@ -62,10 +73,14 @@ async fn accept_loop(listener: TcpListener, gateway: Arc, mut shutdown: } }; let gateway = Arc::clone(&gateway); + let hook = Arc::clone(&hook); tokio::spawn(async move { + // Hot path Traffic Intelligence: только атомики и DashMap. + hook.on_accept(); if let Err(e) = gateway.handle(stream, peer_addr).await { tracing::debug!("connection from {peer_addr}: {e}"); } + hook.on_connection_end(peer_addr.ip()); }); } } diff --git a/src/metrics.rs b/src/metrics.rs index 702dd00..d9fd2c9 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -1,4 +1,6 @@ -use prometheus::{Encoder, IntCounterVec, IntGauge, register_int_counter_vec, register_int_gauge}; +use prometheus::{ + Encoder, IntCounter, IntCounterVec, IntGauge, register_int_counter, register_int_counter_vec, register_int_gauge, +}; use std::sync::LazyLock; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; @@ -38,6 +40,31 @@ pub static SUBNET_VERDICTS: LazyLock = LazyLock::new(|| { .expect("SUBNET_VERDICTS") }); +pub static AUTO_BANS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter_vec!( + "rampart_auto_bans_total", + "Auto-bans issued by Traffic Intelligence", + &["reason"] + ) + .expect("AUTO_BANS_TOTAL") +}); + +pub static INTEL_RATE_LIMIT_DROPS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter!( + "rampart_intel_rate_limit_drops_total", + "Connections dropped by rate limit" + ) + .expect("INTEL_RATE_LIMIT_DROPS_TOTAL") +}); + +pub static INTEL_CPS: LazyLock = + LazyLock::new(|| register_int_gauge!("rampart_intel_cps", "New connections per second window").expect("INTEL_CPS")); + +pub static INTEL_ALERTS_TOTAL: LazyLock = LazyLock::new(|| { + register_int_counter!("rampart_intel_alerts_total", "Attack state-transition alerts emitted") + .expect("INTEL_ALERTS_TOTAL") +}); + /// Отдаёт Prometheus-метрики по голому HTTP/0.9-совместимому ответу. pub async fn run_metrics_server(addr: &str) { let listener = match TcpListener::bind(addr).await { diff --git a/src/traffic/alert.rs b/src/traffic/alert.rs index 031b6c5..221c7c5 100644 --- a/src/traffic/alert.rs +++ b/src/traffic/alert.rs @@ -1,6 +1,9 @@ use std::fmt; use std::net::Ipv4Addr; -use std::time::Instant; +use std::sync::Mutex; +use std::time::{Duration, Instant}; + +use crate::traffic::detector::AttackStatus; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum AlertLevel { @@ -19,7 +22,7 @@ impl fmt::Display for AlertLevel { } } -#[derive(Clone)] +#[derive(Debug, Clone, PartialEq)] pub struct Alert { pub level: AlertLevel, pub message: String, @@ -53,6 +56,61 @@ impl fmt::Display for Alert { } } +/// Дедуплицирующий диспетчер алертов: алерт выдаётся только на переходе +/// состояния детектора (вход в `UnderAttack` и выход из него), не на каждом +/// тике. Логика — poll-and-persist (LNVPS issue #331). +#[derive(Default)] +pub struct AlertDispatcher { + previous: Mutex>, +} + +impl AlertDispatcher { + pub fn new() -> Self { + Self::default() + } + + /// Возвращает `Some(Alert)` только при переходе через границу атаки: + /// вход в `UnderAttack` («attack started») или выход из него + /// («attack ended»). Первое наблюдение алерта не порождает. + pub fn on_status(&self, status: AttackStatus) -> Option { + let mut previous = self.previous.lock().expect("alert dispatcher lock poisoned"); + if *previous == Some(status) { + return None; + } + let was_under_attack = *previous == Some(AttackStatus::UnderAttack); + *previous = Some(status); + match (was_under_attack, status) { + (false, AttackStatus::UnderAttack) => { + Some(Alert::new(AlertLevel::Critical, "attack started".into(), None, 0.0)) + }, + (true, _) => Some(Alert::new(AlertLevel::Info, "attack ended".into(), None, 0.0)), + _ => None, + } + } +} + +/// Отправляет алерт на webhook (fire-and-forget, с таймаутом). +/// +/// Вызывать следует из фоновых задач: сетевой ввод-вывод недопустим в hot path. +pub async fn send_webhook(url: String, alert: Alert) { + let payload = serde_json::json!({ + "level": alert.level.to_string(), + "message": alert.message, + "pps": alert.pps, + }); + let client = reqwest::Client::builder().timeout(Duration::from_secs(5)).build(); + let client = match client { + Ok(c) => c, + Err(e) => { + tracing::warn!("webhook client build failed: {e}"); + return; + }, + }; + if let Err(e) = client.post(url).json(&payload).send().await { + tracing::warn!("webhook delivery failed: {e}"); + } +} + #[cfg(test)] mod tests { use super::*; @@ -77,4 +135,35 @@ mod tests { assert!(s.contains("INFO")); assert!(s.contains("traffic spike")); } + + #[test] + fn test_dispatcher_two_transitions_two_alerts() { + let dispatcher = AlertDispatcher::new(); + assert_eq!(dispatcher.on_status(AttackStatus::Normal), None); + assert_eq!(dispatcher.on_status(AttackStatus::Suspicious), None); + let started = dispatcher.on_status(AttackStatus::UnderAttack); + assert_eq!(dispatcher.on_status(AttackStatus::UnderAttack), None); + assert_eq!(dispatcher.on_status(AttackStatus::UnderAttack), None); + let ended = dispatcher.on_status(AttackStatus::Normal); + assert_eq!(dispatcher.on_status(AttackStatus::Normal), None); + assert!(started.is_some()); + assert!(ended.is_some()); + assert!( + started.expect("started alert").message.contains("attack started") + && ended.expect("ended alert").message.contains("attack ended") + ); + } + + #[test] + fn test_dispatcher_suspicious_does_not_flap_alerts() { + let dispatcher = AlertDispatcher::new(); + for status in [ + AttackStatus::Normal, + AttackStatus::Suspicious, + AttackStatus::Normal, + AttackStatus::Suspicious, + ] { + assert_eq!(dispatcher.on_status(status), None); + } + } } diff --git a/src/traffic/detector.rs b/src/traffic/detector.rs index ab9eb1b..27a4c63 100644 --- a/src/traffic/detector.rs +++ b/src/traffic/detector.rs @@ -8,7 +8,8 @@ pub enum AttackStatus { } pub struct AttackDetector { - profiler: TrafficProfiler, + pps_profiler: TrafficProfiler, + cps_profiler: TrafficProfiler, consecutive_anomalies: u32, } @@ -21,14 +22,21 @@ impl Default for AttackDetector { impl AttackDetector { pub fn new() -> Self { Self { - profiler: TrafficProfiler::new(), + pps_profiler: TrafficProfiler::new(), + cps_profiler: TrafficProfiler::new(), consecutive_anomalies: 0, } } - pub fn analyze(&mut self, pps: f64) -> AttackStatus { - self.profiler.record(pps); - let score = self.profiler.anomaly_score(pps); + /// Анализирует окно метрик (пакеты/с и новые соединения/с) относительно + /// накопленных профилей. Аномалией считается всплеск любой из метрик. + pub fn analyze(&mut self, pps: f64, cps: f64) -> AttackStatus { + self.pps_profiler.record(pps); + self.cps_profiler.record(cps); + let score = self + .pps_profiler + .anomaly_score(pps) + .max(self.cps_profiler.anomaly_score(cps)); if score > 3.0 { self.consecutive_anomalies += 1; @@ -54,40 +62,49 @@ mod tests { fn test_attack_status_normal() { let mut d = AttackDetector::new(); for _ in 0..10 { - d.analyze(100.0); + d.analyze(100.0, 10.0); } - assert_eq!(d.analyze(100.0), AttackStatus::Normal); + assert_eq!(d.analyze(100.0, 10.0), AttackStatus::Normal); } #[test] - fn test_attack_status_suspicious() { + fn test_attack_status_suspicious_pps_spike() { let mut d = AttackDetector::new(); for _ in 0..10 { - d.analyze(1.0); + d.analyze(1.0, 1.0); } - assert_eq!(d.analyze(5.0), AttackStatus::Suspicious); + assert_eq!(d.analyze(5.0, 1.0), AttackStatus::Suspicious); + } + + #[test] + fn test_attack_status_suspicious_cps_spike() { + let mut d = AttackDetector::new(); + for _ in 0..10 { + d.analyze(1.0, 1.0); + } + assert_eq!(d.analyze(1.0, 5.0), AttackStatus::Suspicious); } #[test] fn test_attack_status_under_attack() { let mut d = AttackDetector::new(); for _ in 0..10 { - d.analyze(1.0); + d.analyze(1.0, 1.0); } - d.analyze(10_000.0); - d.analyze(10_000.0); - assert_eq!(d.analyze(10_000.0), AttackStatus::UnderAttack); + d.analyze(10_000.0, 100.0); + d.analyze(10_000.0, 100.0); + assert_eq!(d.analyze(10_000.0, 100.0), AttackStatus::UnderAttack); } #[test] fn test_consecutive_resets_on_normal() { let mut d = AttackDetector::new(); for _ in 0..10 { - d.analyze(1.0); + d.analyze(1.0, 1.0); } - d.analyze(100.0); - d.analyze(100.0); - d.analyze(1.0); - assert_eq!(d.analyze(1.0), AttackStatus::Normal); + d.analyze(100.0, 10.0); + d.analyze(100.0, 10.0); + d.analyze(1.0, 1.0); + assert_eq!(d.analyze(1.0, 1.0), AttackStatus::Normal); } } diff --git a/src/traffic/hook.rs b/src/traffic/hook.rs new file mode 100644 index 0000000..6463d37 --- /dev/null +++ b/src/traffic/hook.rs @@ -0,0 +1,219 @@ +//! Traffic Intelligence в hot path: репутация → авто-бан → метрики окна. +//! +//! Все вызовы из accept-пути дешёвые (DashMap/атомики), блокирующего +//! ввода-вывода нет. + +use crate::config::DetectAutobanConfig; +use crate::filter::blacklist::Blacklist; +use crate::metrics; +use crate::traffic::reputation::IpReputation; +use std::net::IpAddr; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +const BAN_REASON_LOW_REPUTATION: &str = "low_reputation"; +const BAN_REASON_ATTACK: &str = "attack_detector"; + +/// Точка входа Traffic Intelligence для сетевого пути. +/// +/// События: новое соединение (`on_accept`), завершение обработки соединения +/// (`on_connection_end`), эскалация детектора атак (`ban_offenders`). +pub struct TrafficHook { + reputation: Arc, + blacklist: Arc, + autoban: DetectAutobanConfig, + ban_ttl_secs: u64, + conn_window: AtomicU64, +} + +impl TrafficHook { + pub fn new( + reputation: Arc, + blacklist: Arc, + autoban: DetectAutobanConfig, + ban_ttl_secs: u64, + ) -> Self { + Self { + reputation, + blacklist, + autoban, + ban_ttl_secs, + conn_window: AtomicU64::new(0), + } + } + + /// Новое соединение принято accept-loop'ом. Только счётчик окна cps. + pub fn on_accept(&self) { + self.conn_window.fetch_add(1, Ordering::Relaxed); + } + + /// Соединение обработано. Если источник попал под rate-limit (репутация + /// упала до порога) — авто-бан с TTL из конфига. + pub fn on_connection_end(&self, ip: IpAddr) { + if !self.autoban.enabled || self.blacklist.is_blocked(ip) { + return; + } + self.autoban_if_toxic(ip, BAN_REASON_LOW_REPUTATION); + } + + /// Эскалация детектора атак: бан всех источников ниже порога репутации. + /// Вызывается один раз на переход в `UnderAttack`, не на каждый тик. + pub fn ban_offenders(&self) -> usize { + if !self.autoban.enabled { + return 0; + } + let offenders = self.reputation.offenders_below(self.autoban.reputation_threshold); + let mut banned = 0; + for ip in offenders { + if self.blacklist.is_blocked(ip) { + continue; + } + self.ban(ip, BAN_REASON_ATTACK); + banned += 1; + } + banned + } + + /// Забирает накопленное окно новых соединений за секунду (cps). + pub fn take_cps(&self) -> u64 { + self.conn_window.swap(0, Ordering::Relaxed) + } + + fn autoban_if_toxic(&self, ip: IpAddr, reason: &str) { + if self.reputation.score(ip) > self.autoban.reputation_threshold { + return; + } + self.ban(ip, reason); + } + + fn ban(&self, ip: IpAddr, reason: &str) { + self.blacklist.add(ip, Duration::from_secs(self.ban_ttl_secs), reason); + metrics::AUTO_BANS_TOTAL.with_label_values(&[reason]).inc(); + tracing::warn!(%ip, reason, ttl_secs = self.ban_ttl_secs, "auto-ban issued by traffic intelligence"); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::net::{IpAddr, Ipv4Addr}; + + fn ip(octets: [u8; 4]) -> IpAddr { + IpAddr::V4(Ipv4Addr::from(octets)) + } + + fn hook(rep: &Arc, bl: &Arc) -> TrafficHook { + TrafficHook::new( + Arc::clone(rep), + Arc::clone(bl), + DetectAutobanConfig { + enabled: true, + reputation_threshold: -50, + }, + 60, + ) + } + + #[test] + fn test_cps_window_counts_and_resets() { + let rep = Arc::new(IpReputation::new()); + let bl = Arc::new(Blacklist::new()); + let hook = hook(&rep, &bl); + hook.on_accept(); + hook.on_accept(); + hook.on_accept(); + assert_eq!(hook.take_cps(), 3); + assert_eq!(hook.take_cps(), 0); + } + + #[test] + fn test_reputation_fall_triggers_autoban() { + let rep = Arc::new(IpReputation::new()); + let bl = Arc::new(Blacklist::new()); + let hook = hook(&rep, &bl); + let target = ip([10, 1, 0, 1]); + for _ in 0..6 { + rep.record_bad(target); + } + assert_eq!(rep.score(target), -60); + assert!(!bl.is_blocked(target)); + hook.on_connection_end(target); + assert!(bl.is_blocked(target)); + } + + #[test] + fn test_no_autoban_above_threshold() { + let rep = Arc::new(IpReputation::new()); + let bl = Arc::new(Blacklist::new()); + let hook = hook(&rep, &bl); + let target = ip([10, 1, 0, 2]); + for _ in 0..4 { + rep.record_bad(target); + } + assert_eq!(rep.score(target), -40); + hook.on_connection_end(target); + assert!(!bl.is_blocked(target)); + } + + #[test] + fn test_disabled_autoban_never_bans() { + let rep = Arc::new(IpReputation::new()); + let bl = Arc::new(Blacklist::new()); + let hook = TrafficHook::new( + Arc::clone(&rep), + Arc::clone(&bl), + DetectAutobanConfig { + enabled: false, + reputation_threshold: -50, + }, + 60, + ); + let target = ip([10, 1, 0, 3]); + for _ in 0..20 { + rep.record_bad(target); + } + hook.on_connection_end(target); + assert!(hook.ban_offenders() == 0); + assert!(!bl.is_blocked(target)); + } + + #[test] + fn test_ban_offenders_bans_only_below_threshold() { + let rep = Arc::new(IpReputation::new()); + let bl = Arc::new(Blacklist::new()); + let hook = hook(&rep, &bl); + let bad = ip([10, 2, 0, 1]); + let ok = ip([10, 2, 0, 2]); + for _ in 0..8 { + rep.record_bad(bad); + } + rep.record_good(ok); + assert_eq!(hook.ban_offenders(), 1); + assert!(bl.is_blocked(bad)); + assert!(!bl.is_blocked(ok)); + assert_eq!(hook.ban_offenders(), 0); + } + + #[test] + fn test_autoban_ttl_expires() { + let rep = Arc::new(IpReputation::new()); + let bl = Arc::new(Blacklist::new()); + let hook = TrafficHook::new( + Arc::clone(&rep), + Arc::clone(&bl), + DetectAutobanConfig { + enabled: true, + reputation_threshold: -50, + }, + // Короткий TTL как мок времени: бан жив, потом истекает. + 0, + ); + let target = ip([10, 3, 0, 1]); + for _ in 0..20 { + rep.record_bad(target); + } + hook.on_connection_end(target); + assert!(!bl.is_blocked(target), "zero TTL must not produce an active ban"); + } +} diff --git a/src/traffic/mod.rs b/src/traffic/mod.rs index 1e7866b..9c4bda7 100644 --- a/src/traffic/mod.rs +++ b/src/traffic/mod.rs @@ -3,6 +3,7 @@ pub mod alert; pub mod detector; pub mod ewma; +pub mod hook; pub mod prefix; pub mod profiler; pub mod reputation; diff --git a/src/traffic/reputation.rs b/src/traffic/reputation.rs index e7f12a7..dba19f6 100644 --- a/src/traffic/reputation.rs +++ b/src/traffic/reputation.rs @@ -33,6 +33,15 @@ impl IpReputation { self.scores.get(&ip).map(|v| *v).unwrap_or(0) } + /// Возвращает источники со скором не выше `threshold` (нарушители). + pub fn offenders_below(&self, threshold: i32) -> Vec { + self.scores + .iter() + .filter(|entry| *entry.value() <= threshold) + .map(|entry| *entry.key()) + .collect() + } + pub fn is_trusted(&self, ip: IpAddr) -> bool { self.score(ip) > 50 } @@ -107,4 +116,23 @@ mod tests { rep.record_bad(ip); assert_eq!(rep.score(ip), -10); } + + #[test] + fn test_offenders_below_threshold() { + let rep = IpReputation::new(); + let bad = v4([10, 0, 0, 1]); + let good = v4([10, 0, 0, 2]); + for _ in 0..6 { + rep.record_bad(bad); + } + rep.record_good(good); + let offenders = rep.offenders_below(-50); + assert_eq!(offenders, vec![bad]); + } + + #[test] + fn test_offenders_empty_when_none_below() { + let rep = IpReputation::new(); + assert!(rep.offenders_below(-50).is_empty()); + } } diff --git a/src/xdp/diagnostics.rs b/src/xdp/diagnostics.rs new file mode 100644 index 0000000..8c15f89 --- /dev/null +++ b/src/xdp/diagnostics.rs @@ -0,0 +1,284 @@ +//! Диагностика окружения перед XDP-attach: ядро, BTF, драйвер NIC, привилегии. +//! +//! Основной источник боли при XDP — непонятные ошибки attach на неподдерживаемых +//! ядрах и драйверах. Модуль собирает [`EnvironmentReport`] ДО загрузки BPF-программы +//! и даёт человекочитаемый вердикт с предупреждениями. + +use anyhow::{Result, bail}; +use std::fmt; + +/// Минимально поддерживаемая версия ядра: 5.15 LTS. +pub const MIN_KERNEL: KernelVersion = KernelVersion::new(5, 15, 0); + +const OSRELEASE_PATH: &str = "/proc/sys/kernel/osrelease"; +const BTF_PATH: &str = "/sys/kernel/btf/vmlinux"; +const STATUS_PATH: &str = "/proc/self/status"; + +/// Драйверы с известной поддержкой native XDP (по данным xdpgeneric/native +/// матриц upstream-ядра). Отсутствие в списке не означает отсутствие поддержки, +/// но для таких драйверов предсказываем generic mode. +const NATIVE_XDP_DRIVERS: &[&str] = &[ + "virtio_net", + "ixgbe", + "ixgbevf", + "i40e", + "iavf", + "ice", + "mlx4_core", + "mlx5_core", + "igb", + "igc", + "e1000e", + "vmxnet3", + "bnxt_en", + "nfp", + "sfc", + "ena", + "hv_netvsc", + "macvlan", +]; + +/// Версия ядра Linux, пригодная для сравнения (`major.minor.patch`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub struct KernelVersion { + pub major: u16, + pub minor: u16, + pub patch: u16, +} + +impl KernelVersion { + #[must_use] + pub const fn new(major: u16, minor: u16, patch: u16) -> Self { + Self { major, minor, patch } + } + + /// Парсит строку формата `uname -r`. Берёт первые три числовых компонента, + /// дистрибутивные суффиксы (`6.8.0-45-generic`, `5.15.0-rc2`) отбрасываются. + /// Строка без цифр не парсится. + #[must_use] + pub fn parse(release: &str) -> Option { + let mut nums = release + .split(|c: char| !c.is_ascii_digit()) + .filter_map(|part| part.parse::().ok()); + let major = nums.next()?; + let minor = nums.next().unwrap_or(0); + let patch = nums.next().unwrap_or(0); + Some(Self::new(major, minor, patch)) + } +} + +impl fmt::Display for KernelVersion { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}.{}.{}", self.major, self.minor, self.patch) + } +} + +/// Режим присоединения XDP-программы к интерфейсу. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AttachMode { + /// Программа исполняется в контексте драйвера NIC — минимальные накладные расходы. + Native, + /// Программа вызывается из сетевого стека после ingress — работает везде, дороже по CPU. + Generic, + /// Аппаратная разгрузка в NIC — требует явного включения и поддержки железа, + /// автоматически не выбирается. + Offload, +} + +impl fmt::Display for AttachMode { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::Native => "native", + Self::Generic => "generic", + Self::Offload => "offload", + }) + } +} + +/// Абстракция над procfs/sysfs: реальная реализация — [`FilesystemProbe`], +/// в тестах — мок. Изолирует диагностику от реального ядра. +pub trait SystemProbe { + /// Читает файл целиком. + /// + /// # Errors + /// Проксирует ошибку чтения файла. + fn read_file(&self, path: &str) -> std::io::Result; + + #[must_use] + fn path_exists(&self, path: &str) -> bool; + + #[must_use] + fn symlink_target(&self, path: &str) -> Option; + + #[must_use] + fn effective_uid(&self) -> u32; +} + +/// Реальная реализация [`SystemProbe`] поверх `std::fs`. +pub struct FilesystemProbe; + +impl SystemProbe for FilesystemProbe { + fn read_file(&self, path: &str) -> std::io::Result { + std::fs::read_to_string(path) + } + + fn path_exists(&self, path: &str) -> bool { + std::path::Path::new(path).exists() + } + + fn symlink_target(&self, path: &str) -> Option { + Some(std::fs::read_link(path).ok()?.to_string_lossy().into_owned()) + } + + /// Эвристика: эффективный uid берётся из `/proc/self/status`. + /// root почти всегда имеет CAP_BPF/CAP_NET_ADMIN (если явно не урезаны); + /// для не-root подтверждение возможно только попыткой attach. + /// Нечитаемый статус трактуем как непривилегированный процесс. + fn effective_uid(&self) -> u32 { + std::fs::read_to_string(STATUS_PATH) + .ok() + .and_then(|status| parse_euid(&status)) + .unwrap_or(1) + } +} + +fn parse_euid(status: &str) -> Option { + status.lines().find_map(|line| { + let mut fields = line.strip_prefix("Uid:")?.split_whitespace(); + fields.nth(1)?.parse().ok() + }) +} + +fn driver_from_link(target: &str) -> Option { + let name = target.rsplit('/').next()?; + (!name.is_empty()).then(|| name.to_owned()) +} + +/// Поддерживает ли драйвер native XDP (по таблице известных драйверов). +#[must_use] +pub fn driver_supports_native_xdp(driver: &str) -> bool { + NATIVE_XDP_DRIVERS.contains(&driver) +} + +/// Структурированный отчёт о пригодности окружения для XDP. +pub struct EnvironmentReport { + pub kernel_version: Option, + pub btf_available: bool, + pub privileged: bool, + pub driver_name: Option, + pub warnings: Vec, +} + +impl EnvironmentReport { + /// Собирает отчёт через произвольную реализацию [`SystemProbe`]. + #[must_use] + pub fn collect(probe: &dyn SystemProbe, interface: &str) -> Self { + let kernel_version = probe + .read_file(OSRELEASE_PATH) + .ok() + .and_then(|r| KernelVersion::parse(r.trim())); + let btf_available = probe.path_exists(BTF_PATH); + let privileged = probe.effective_uid() == 0; + let driver_link = probe.symlink_target(&format!("/sys/class/net/{interface}/device/driver")); + let driver_name = driver_link.as_deref().and_then(driver_from_link); + + let mut warnings = Vec::new(); + if !privileged { + warnings.push( + "процесс не от root: CAP_BPF/CAP_NET_ADMIN не подтверждены, \ + attach скорее всего завершится EPERM" + .to_owned(), + ); + } + if !btf_available { + warnings.push(format!( + "{BTF_PATH} недоступен: CO-RE релокации невозможны, переносимость BPF-программы ограничена" + )); + } + if kernel_version.is_some_and(|v| v < KernelVersion::new(5, 11, 0)) { + warnings.push( + "ядро < 5.11: память BPF-карт ограничена RLIMIT_MEMLOCK — увеличьте `ulimit -l` или обновите ядро" + .to_owned(), + ); + } + if driver_name.is_none() { + warnings.push(format!( + "драйвер интерфейса '{interface}' не определён (нет device/driver symlink): ожидается generic mode" + )); + } + + Self { + kernel_version, + btf_available, + privileged, + driver_name, + warnings, + } + } + + /// Ожидаемый режим attach по таблице драйверов. + #[must_use] + pub fn attach_mode(&self) -> AttachMode { + match self.driver_name.as_deref() { + Some(driver) if driver_supports_native_xdp(driver) => AttachMode::Native, + _ => AttachMode::Generic, + } + } + + /// Человекочитаемое объяснение вердикта. + #[must_use] + pub fn verdict(&self) -> String { + let kernel = self + .kernel_version + .map_or_else(|| "неизвестна".to_owned(), |v| v.to_string()); + match (self.attach_mode(), self.driver_name.as_deref()) { + (AttachMode::Native, Some(driver)) => format!( + "ядро {kernel}: драйвер '{driver}' поддерживает native XDP — программа работает в драйвере, минимальные накладные расходы" + ), + (_, driver) => format!( + "ядро {kernel}: драйвер '{}' не поддерживает native XDP → будет generic mode, CPU дороже", + driver.unwrap_or("неизвестный") + ), + } + } + + /// Fail-fast проверка минимальной версии ядра. + /// + /// # Errors + /// Версия ядра не определена или ниже [`MIN_KERNEL`]. + pub fn validate(&self) -> Result<()> { + let Some(version) = self.kernel_version else { + bail!("не удалось определить версию ядра ({OSRELEASE_PATH}) — XDP attach отклонён"); + }; + if version < MIN_KERNEL { + bail!("ядро {version} ниже минимально поддерживаемой {MIN_KERNEL}: XDP attach отклонён, обновите ядро"); + } + Ok(()) + } +} + +/// Предстартовая диагностика перед загрузкой XDP-программы: +/// структурный отчёт в лог, предупреждения, fail-fast на старом ядре. +/// +/// # Errors +/// См. [`EnvironmentReport::validate`]. +pub(crate) fn preflight(interface: &str) -> Result<()> { + let report = EnvironmentReport::collect(&FilesystemProbe, interface); + let kernel = report + .kernel_version + .map_or_else(|| "unknown".to_owned(), |v| v.to_string()); + tracing::info!( + interface, + kernel = %kernel, + btf = report.btf_available, + privileged = report.privileged, + driver = report.driver_name.as_deref().unwrap_or("unknown"), + mode = %report.attach_mode(), + "XDP environment: {}", + report.verdict() + ); + for warning in &report.warnings { + tracing::warn!(interface, "{warning}"); + } + report.validate() +} diff --git a/src/xdp/filter.rs b/src/xdp/filter.rs index 907b57f..08aa140 100644 --- a/src/xdp/filter.rs +++ b/src/xdp/filter.rs @@ -25,6 +25,9 @@ impl XdpFilter { } pub fn load(&mut self) -> Result<()> { + // Диагностика ДО загрузки: fail-fast на неподдерживаемом ядре/драйвере. + super::diagnostics::preflight(&self.interface)?; + let bpf_obj = include_bytes!(concat!(env!("OUT_DIR"), "/universal_filter.o")); let obj = ObjectBuilder::default() .open_memory(bpf_obj) diff --git a/src/xdp/mod.rs b/src/xdp/mod.rs index 34c607a..7d6ea98 100644 --- a/src/xdp/mod.rs +++ b/src/xdp/mod.rs @@ -2,6 +2,11 @@ mod stats; pub use stats::XdpStats; +mod diagnostics; +pub use diagnostics::{ + AttachMode, EnvironmentReport, FilesystemProbe, KernelVersion, MIN_KERNEL, SystemProbe, driver_supports_native_xdp, +}; + #[cfg(feature = "xdp")] mod filter; #[cfg(feature = "xdp")] diff --git a/tests/config_parse.rs b/tests/config_parse.rs index 0c75507..df77bb9 100644 --- a/tests/config_parse.rs +++ b/tests/config_parse.rs @@ -61,3 +61,31 @@ fn rejects_invalid_whitelist_ip() { let result = Config::parse_str("whitelist = [\"10.0.0.999\"]"); assert!(result.is_err()); } + +#[test] +fn detect_autoban_defaults_disabled() { + let config = Config::parse_str("").expect("empty config must parse"); + assert!(!config.detect.autoban.enabled); + assert_eq!(config.detect.autoban.reputation_threshold, -50); +} + +#[test] +fn parses_detect_sections() { + let config = Config::parse_str( + r#" +[detect.autoban] +enabled = true +reputation_threshold = -30 + +[detect.alert] +webhook_url = "https://hooks.example.test/rampart" +"#, + ) + .expect("detect sections must parse"); + assert!(config.detect.autoban.enabled); + assert_eq!(config.detect.autoban.reputation_threshold, -30); + assert_eq!( + config.detect.alert.webhook_url.as_deref(), + Some("https://hooks.example.test/rampart") + ); +} diff --git a/tests/traffic_intel.rs b/tests/traffic_intel.rs new file mode 100644 index 0000000..e869ad4 --- /dev/null +++ b/tests/traffic_intel.rs @@ -0,0 +1,112 @@ +//! Интеграционные тесты Traffic Intelligence в hot path: авто-бан по +//! репутации, детектор атак на синтетическом всплеске, алерты на переходах. + +use rampart::config::DetectAutobanConfig; +use rampart::filter::blacklist::Blacklist; +use rampart::traffic::alert::AlertDispatcher; +use rampart::traffic::detector::{AttackDetector, AttackStatus}; +use rampart::traffic::hook::TrafficHook; +use rampart::traffic::reputation::IpReputation; +use std::net::{IpAddr, Ipv4Addr}; +use std::sync::Arc; +use std::time::Duration; + +fn ip(octets: [u8; 4]) -> IpAddr { + IpAddr::V4(Ipv4Addr::from(octets)) +} + +fn autoban_cfg(enabled: bool) -> DetectAutobanConfig { + DetectAutobanConfig { + enabled, + reputation_threshold: -50, + } +} + +#[test] +fn reputation_fall_after_drops_triggers_autoban() { + let reputation = Arc::new(IpReputation::new()); + let blacklist = Arc::new(Blacklist::new()); + let hook = TrafficHook::new(Arc::clone(&reputation), Arc::clone(&blacklist), autoban_cfg(true), 60); + let attacker = ip([198, 51, 100, 7]); + + for _ in 0..6 { + // Шесть дропов rate-limit'ом: -10 за каждый, порог auto-ban = -50. + reputation.record_bad(attacker); + hook.on_connection_end(attacker); + } + + assert!( + reputation.score(attacker) <= -50, + "reputation must fall below threshold" + ); + assert!( + blacklist.is_blocked(attacker), + "auto-ban must fire after repeated drops" + ); +} + +#[test] +fn autoban_ttl_expires_ban() { + let reputation = Arc::new(IpReputation::new()); + let blacklist = Arc::new(Blacklist::new()); + let hook = TrafficHook::new( + Arc::clone(&reputation), + Arc::clone(&blacklist), + autoban_cfg(true), + // Короткий TTL из «конфига» как мок времени. + 0, + ); + let attacker = ip([198, 51, 100, 8]); + for _ in 0..20 { + reputation.record_bad(attacker); + } + hook.on_connection_end(attacker); + std::thread::sleep(Duration::from_millis(5)); + assert!( + !blacklist.is_blocked(attacker), + "zero TTL must not produce an active ban" + ); +} + +#[test] +fn detector_flags_synthetic_traffic_spike() { + let mut detector = AttackDetector::new(); + + // Базовая линия: тихий профиль. + for _ in 0..12 { + detector.analyze(500.0, 50.0); + } + + // Синтетический всплеск: x200 pps и cps три окна подряд. + assert_eq!( + detector.analyze(100_000.0, 10_000.0), + AttackStatus::Suspicious, + "first anomalous window is suspicious" + ); + assert_eq!(detector.analyze(100_000.0, 10_000.0), AttackStatus::Suspicious); + assert_eq!( + detector.analyze(100_000.0, 10_000.0), + AttackStatus::UnderAttack, + "three anomalous windows in a row must escalate to UnderAttack" + ); +} + +#[test] +fn alerts_fire_only_on_state_transitions() { + let dispatcher = AlertDispatcher::new(); + let sequence = [ + AttackStatus::Normal, + AttackStatus::Normal, + AttackStatus::Suspicious, + AttackStatus::UnderAttack, + AttackStatus::UnderAttack, + AttackStatus::UnderAttack, + AttackStatus::Suspicious, + AttackStatus::Normal, + AttackStatus::Normal, + ]; + let alerts: Vec<_> = sequence.iter().filter_map(|s| dispatcher.on_status(*s)).collect(); + assert_eq!(alerts.len(), 2, "two transitions must yield exactly two alerts"); + assert!(alerts[0].message.contains("attack started")); + assert!(alerts[1].message.contains("attack ended")); +} diff --git a/tests/xdp_diagnostics.rs b/tests/xdp_diagnostics.rs new file mode 100644 index 0000000..9f9d485 --- /dev/null +++ b/tests/xdp_diagnostics.rs @@ -0,0 +1,134 @@ +use rampart::xdp::{AttachMode, EnvironmentReport, KernelVersion, MIN_KERNEL, SystemProbe, driver_supports_native_xdp}; +use std::collections::HashMap; + +struct MockProbe { + files: HashMap, + symlinks: HashMap, + uid: u32, +} + +impl MockProbe { + fn linux(driver: &str, release: &str) -> Self { + let mut files = HashMap::new(); + files.insert("/proc/sys/kernel/osrelease".to_owned(), format!("{release}\n")); + files.insert("/sys/kernel/btf/vmlinux".to_owned(), String::new()); + files.insert("/proc/self/status".to_owned(), "Uid:\t0\t0\t0\t0\n".to_owned()); + let mut symlinks = HashMap::new(); + if !driver.is_empty() { + symlinks.insert( + "/sys/class/net/eth0/device/driver".to_owned(), + format!("/sys/bus/pci/drivers/{driver}"), + ); + } + Self { + files, + symlinks, + uid: 0, + } + } + + fn collect(&self) -> EnvironmentReport { + EnvironmentReport::collect(self, "eth0") + } +} + +impl SystemProbe for MockProbe { + fn read_file(&self, path: &str) -> std::io::Result { + self.files + .get(path) + .cloned() + .ok_or_else(|| std::io::Error::other("file not found")) + } + + fn path_exists(&self, path: &str) -> bool { + self.files.contains_key(path) || self.symlinks.contains_key(path) + } + + fn symlink_target(&self, path: &str) -> Option { + self.symlinks.get(path).cloned() + } + + fn effective_uid(&self) -> u32 { + self.uid + } +} + +#[test] +fn parses_kernel_release_with_distro_suffixes() { + assert_eq!( + KernelVersion::parse("6.8.0-45-generic"), + Some(KernelVersion::new(6, 8, 0)) + ); + assert_eq!(KernelVersion::parse("5.15.0"), Some(KernelVersion::new(5, 15, 0))); + assert_eq!(KernelVersion::parse("6.12"), Some(KernelVersion::new(6, 12, 0))); + assert_eq!(KernelVersion::parse(""), None); + assert_eq!(KernelVersion::parse("generic"), None); +} + +#[test] +fn selects_attach_mode_by_driver_table() { + let virtio = MockProbe::linux("virtio_net", "6.8.0-45-generic").collect(); + assert_eq!(virtio.attach_mode(), AttachMode::Native); + + let realtek = MockProbe::linux("r8169", "6.8.0-45-generic").collect(); + assert_eq!(realtek.attach_mode(), AttachMode::Generic); +} + +#[test] +fn unknown_driver_falls_back_to_generic() { + assert!(!driver_supports_native_xdp("r8169")); + assert!(driver_supports_native_xdp("mlx5_core")); + + let no_driver = MockProbe::linux("", "6.1.0").collect(); + assert_eq!(no_driver.driver_name, None); + assert_eq!(no_driver.attach_mode(), rampart::xdp::AttachMode::Generic); +} + +#[test] +fn verdict_explains_generic_cpu_cost() { + let report = MockProbe::linux("r8169", "6.8.0-45-generic").collect(); + let verdict = report.verdict(); + assert!(verdict.contains("r8169"), "verdict must name the driver: {verdict}"); + assert!( + verdict.contains("generic mode"), + "verdict must mention generic: {verdict}" + ); + assert!(verdict.contains("CPU"), "verdict must explain CPU cost: {verdict}"); + + let native = MockProbe::linux("ixgbe", "6.1.0").collect(); + assert!(native.verdict().contains("native XDP")); +} + +#[test] +fn fails_fast_on_kernel_below_minimum() { + let old = MockProbe::linux("virtio_net", "5.10.0-rc6"); + let err = old.collect().validate().expect_err("kernel 5.10 must be rejected"); + assert!(err.to_string().contains("5.10")); + + let mut unknown = MockProbe::linux("virtio_net", "6.8.0"); + unknown.files.remove("/proc/sys/kernel/osrelease"); + assert!(unknown.collect().validate().is_err(), "unknown kernel must be rejected"); +} + +#[test] +fn accepts_supported_kernels() { + assert!(MIN_KERNEL == KernelVersion::new(5, 15, 0)); + for release in ["5.15.0", "6.1.0-13-amd64", "6.12.8"] { + let report = MockProbe::linux("virtio_net", release).collect(); + report.validate().expect("supported kernel must pass validation"); + } +} + +#[test] +fn warns_when_not_privileged_and_btf_missing() { + let mut probe = MockProbe::linux("virtio_net", "6.8.0"); + probe.uid = 1000; + probe.files.remove("/sys/kernel/btf/vmlinux"); + + let report = probe.collect(); + assert!(!report.privileged); + assert!(!report.btf_available); + let joined = report.warnings.join("\n"); + assert!(joined.contains("CAP_BPF"), "must warn about capabilities: {joined}"); + assert!(joined.contains("CO-RE"), "must warn about missing BTF: {joined}"); +} diff --git a/xdp/core/config.h b/xdp/core/config.h index d5d438c..2f818d2 100644 --- a/xdp/core/config.h +++ b/xdp/core/config.h @@ -33,4 +33,12 @@ static volatile const __u64 G_UDP_WINDOW_NS = 1000000000ULL; // 1 sec static volatile const __u8 G_FEATURE_SYN_THROTTLE = 1; static volatile const __u8 G_FEATURE_EVENTS = 1; +// ── RST-challenge (Oubliette liveness proof) ── +// 0 = disabled: single branch, zero cost for legitimate traffic. +// Secret is patched by the loader at attach time (default is a placeholder). +static volatile const __u8 G_SYN_CHALLENGE_ENABLED = 0; +static volatile const __u64 G_CHALLENGE_SECRET = 0xA5A55A5A5A5AA5A5ULL; +static volatile const __u32 G_CHALLENGE_TIMEOUT_MS = 3000; // pending TTL +static volatile const __u64 G_CHALLENGE_VERIFIED_TTL_NS = 300000000000ULL; // 5 min + #endif /* RAMPART_CONFIG_H */ diff --git a/xdp/core/maps.h b/xdp/core/maps.h index 7dd160b..94dc572 100644 --- a/xdp/core/maps.h +++ b/xdp/core/maps.h @@ -45,6 +45,30 @@ struct { __type(value, struct throttle_entry); } udp_rate_limit SEC(".maps"); +// ── RST-challenge pending state (challenge reflected, awaiting proof) ── +struct challenge_pending { + __u64 sent_at; // ktime_ns of the reflected SYN-ACK + __u32 marker; // our ISN (client may echo it as ack) + __u32 bad_ack; // bogus ack sent (live client echoes it as RST seq) +}; + +// 🔗 RST-challenge: verified sources (liveness proven) — sliding TTL via +// last_seen, LRU eviction as the hard cap +struct { + __uint(type, BPF_MAP_TYPE_LRU_HASH); + __uint(max_entries, 65536); + __type(key, __u32); // src_ip + __type(value, __u64); // last_seen (ktime_ns) +} challenge_verified SEC(".maps"); + +// 🔗 RST-challenge: outstanding challenges keyed by flow 4-tuple +struct { + __uint(type, BPF_MAP_TYPE_LRU_HASH); + __uint(max_entries, 16384); + __type(key, struct flow_key); + __type(value, struct challenge_pending); +} challenge_pending SEC(".maps"); + // 🔗 Statistics (per-CPU, атомарные инкременты) #define STAT_TOTAL 0 #define STAT_TCP 1 @@ -55,6 +79,9 @@ struct { #define STAT_DROP 6 #define STAT_UDP 7 #define STAT_RATE_LIMIT 8 +#define STAT_CHALLENGE_SENT 9 +#define STAT_CHALLENGE_VERIFIED 10 +#define STAT_CHALLENGE_FAILED 11 struct { __uint(type, BPF_MAP_TYPE_PERCPU_ARRAY); diff --git a/xdp/core/stats.h b/xdp/core/stats.h index d6b3ee6..a884ba2 100644 --- a/xdp/core/stats.h +++ b/xdp/core/stats.h @@ -20,6 +20,9 @@ static __always_inline void inc_whitelist(void) { inc_stat(STAT_WHITELIST); } static __always_inline void inc_blacklist(void) { inc_stat(STAT_BLACKLIST); } static __always_inline void inc_syn_throttle(void) { inc_stat(STAT_SYN_THROTTLE); } static __always_inline void inc_rate_limit(void) { inc_stat(STAT_RATE_LIMIT); } +static __always_inline void inc_chal_sent(void) { inc_stat(STAT_CHALLENGE_SENT); } +static __always_inline void inc_chal_verified(void) { inc_stat(STAT_CHALLENGE_VERIFIED); } +static __always_inline void inc_chal_failed(void) { inc_stat(STAT_CHALLENGE_FAILED); } static __always_inline void inc_pass(void) { inc_stat(STAT_PASS); } static __always_inline void inc_drop(void) { inc_stat(STAT_DROP); } diff --git a/xdp/core/syn_challenge.h b/xdp/core/syn_challenge.h new file mode 100644 index 0000000..d150608 --- /dev/null +++ b/xdp/core/syn_challenge.h @@ -0,0 +1,221 @@ +#ifndef RAMPART_SYN_CHALLENGE_H +#define RAMPART_SYN_CHALLENGE_H + +// ── Rampart RST-challenge (Oubliette-style liveness proof) ── +// +// Protocol-agnostic source verification, compatible with any TCP client: +// 1. SYN from a source that is neither whitelisted nor verified: +// kernel reflects a SYN-ACK with a DELIBERATELY WRONG ack number +// (deterministic from the 4-tuple + secret) and drops the original SYN. +// 2. Spoofed source never reacts — it does not own the address. +// 3. Live client's TCP stack rejects the unacceptable SYN-ACK +// (RFC 9293 SYN-SENT: SEG.ACK outside [ISS+1, SND.NXT] → RST) +// echoing secret-derived values. Seeing that RST marks the source +// verified; the app reconnects and its next SYN reaches the backend. +// +// Pending challenges expire after G_CHALLENGE_TIMEOUT_MS (spoof = silence). + +#include "common.h" +#include "config.h" +#include "maps.h" +#include "stats.h" + +#define CHAL_CONTINUE (-1) // fall through to the normal filter path + +// ── Secret-dependent 64-bit mixing (splitmix-style finalizer) ── +static __always_inline __u64 chal_mix64(__u64 h, __u64 v) +{ + h ^= v; + h *= 0x9E3779B97F4A7C15ULL; + h ^= h >> 29; + return h; +} + +// Per-flow challenge values; infeasible to forge without G_CHALLENGE_SECRET +static __always_inline __u64 chal_derive(const struct flow_key *k) +{ + __u64 h = (__u64)G_CHALLENGE_SECRET ^ 0x5253544348414C53ULL; + h = chal_mix64(h, ((__u64)k->src_ip << 32) | k->dst_ip); + h = chal_mix64(h, ((__u64)k->src_port << 48) | ((__u64)k->dst_port << 32)); + h *= 0xFF51AFD7ED558CCDULL; + h ^= h >> 32; + return h; +} + +// ── Ones-complement helpers (both IPv4/TCP headers are 20 bytes = 10 words) ── +static __always_inline __u32 chal_sum10(const __u16 *w) +{ + __u32 s = 0; + s += w[0]; s += w[1]; s += w[2]; s += w[3]; s += w[4]; + s += w[5]; s += w[6]; s += w[7]; s += w[8]; s += w[9]; + return s; +} + +static __always_inline __u16 chal_csum_fold(__u32 sum) +{ + sum = (sum >> 16) + (sum & 0xFFFF); + sum += sum >> 16; + return (__u16)~sum; +} + +// Rewrite the packet in place into a SYN-ACK: MAC/IP/ports swapped, +// seq = marker, ack = bad_ack (guaranteed unacceptable for the client), +// clean 20-byte TCP header, IP checksum recomputed, TCP checksum rebuilt. +// All reads happen BEFORE any packet write (writes invalidate verifier +// bounds), the tail is trimmed so the peer sees no leftover option bytes. +static __always_inline void chal_build_synack(struct xdp_md *ctx, + struct ethhdr *eth, void *l3, + __u8 is_ipv6, struct tcphdr *tcp, + __u32 marker, __u32 bad_ack) +{ + // ── Snapshot phase (packet reads) ── + __u16 ipw[10]; + __u16 tcpw[10]; + __u8 ms[ETH_ALEN], md[ETH_ALEN]; + __u32 saddr[4] = {0}, daddr[4] = {0}; + + __builtin_memcpy(ms, eth->h_source, ETH_ALEN); + __builtin_memcpy(md, eth->h_dest, ETH_ALEN); + __builtin_memcpy(ipw, l3, 20); + __builtin_memcpy(tcpw, tcp, 20); + + if (is_ipv6) { + __builtin_memcpy(saddr, &((struct ipv6hdr *)l3)->daddr, 16); // swapped + __builtin_memcpy(daddr, &((struct ipv6hdr *)l3)->saddr, 16); + } else { + saddr[0] = ((struct iphdr *)l3)->daddr; + daddr[0] = ((struct iphdr *)l3)->saddr; + } + + // ── Compute images in registers ── + // IPv4 header: swap addresses, fresh TTL, recompute checksum + if (!is_ipv6) { + __u16 s0 = ipw[6], s1 = ipw[7]; + ipw[6] = ipw[8]; + ipw[7] = ipw[9]; + ipw[8] = s0; + ipw[9] = s1; + ((__u8 *)ipw)[8] = 64; // TTL + ipw[5] = 0; // checksum placeholder + ipw[5] = chal_csum_fold(chal_sum10(ipw)); + } + + // TCP header: swap ports, inject secret-derived seq/ack, SYN|ACK + __u16 nsport = tcpw[1]; // new source = old dest + __u16 ndport = tcpw[0]; // new dest = old source + tcpw[0] = nsport; + tcpw[1] = ndport; + tcpw[2] = (__u16)(marker >> 16); + tcpw[3] = (__u16)marker; + tcpw[4] = (__u16)(bad_ack >> 16); + tcpw[5] = (__u16)bad_ack; + tcpw[6] = bpf_htons(0x5012); // doff=5, flags SYN|ACK + tcpw[7] = bpf_htons(0xFFFF); // window + tcpw[8] = 0; // checksum placeholder + tcpw[9] = 0; // urg_ptr + + // TCP checksum over pseudo-header + 20-byte header + __u32 sum = 0; + if (is_ipv6) { + __u16 *w = (__u16 *)saddr; + sum += w[0] + w[1] + w[2] + w[3] + w[4] + w[5] + w[6] + w[7]; + w = (__u16 *)daddr; + sum += w[0] + w[1] + w[2] + w[3] + w[4] + w[5] + w[6] + w[7]; + } else { + sum += ipw[6] + ipw[7] + ipw[8] + ipw[9]; // new IPv4 saddr/daddr + } + sum += bpf_htons(IPPROTO_TCP); + sum += bpf_htons((__u16)sizeof(struct tcphdr)); + sum += chal_sum10(tcpw); + tcpw[8] = chal_csum_fold(sum); + + // ── Write phase ── + __builtin_memcpy(eth->h_dest, ms, ETH_ALEN); + __builtin_memcpy(eth->h_source, md, ETH_ALEN); + if (is_ipv6) { + struct ipv6hdr *ip6 = l3; + __builtin_memcpy(&ip6->saddr, daddr, 16); + __builtin_memcpy(&ip6->daddr, saddr, 16); + ip6->hop_limit = 64; + } else { + __builtin_memcpy(l3, ipw, 20); + } + __builtin_memcpy(tcp, tcpw, 20); + + // Trim to the bare header (drop SYN options/payload remnants) + long trim = (long)((void *)(long)ctx->data_end - + ((void *)tcp + sizeof(struct tcphdr))); + if (trim > 0) + bpf_xdp_adjust_tail(ctx, -trim); +} + +// ── Pure SYN from an unknown source: reflect the wrong SYN-ACK challenge ── +// Returns CHAL_CONTINUE (verified source — process normally), XDP_TX +// (challenge sent) or XDP_DROP. +static __always_inline int chal_on_syn(struct xdp_md *ctx, struct ethhdr *eth, + void *l3, __u8 is_ipv6, + struct tcphdr *tcp, + const struct flow_key *flow, __u64 now) +{ + __u64 *v = bpf_map_lookup_elem(&challenge_verified, &flow->src_ip); + if (v) { + if (now - *v <= G_CHALLENGE_VERIFIED_TTL_NS) { + *v = now; // sliding TTL refresh + return CHAL_CONTINUE; + } + bpf_map_delete_elem(&challenge_verified, &flow->src_ip); // expired + } + + __u64 h = chal_derive(flow); + __u32 marker = (__u32)(h >> 32); // our ISN + __u32 bad_ack = (__u32)h; // bogus ack the client must reject + + __u32 expect = bpf_ntohl(tcp->seq) + 1; + if (bad_ack == expect) // ~2^-32 collision with a valid handshake — kill it + bad_ack = ~bad_ack; + + struct challenge_pending p = { + .sent_at = now, + .marker = marker, + .bad_ack = bad_ack, + }; + if (bpf_map_update_elem(&challenge_pending, flow, &p, BPF_ANY)) { + inc_drop(); // cannot track the challenge — fail closed + return XDP_DROP; + } + + inc_chal_sent(); + chal_build_synack(ctx, eth, l3, is_ipv6, tcp, marker, bad_ack); + return XDP_TX; +} + +// ── RST answering a pending challenge: proof of life ── +// Accept only if the RST echoes a secret-derived value (our bad ack as its +// seq, or our marker+1 as its ack). Anything else fails the challenge. +static __always_inline int chal_on_rst(struct tcphdr *tcp, + const struct flow_key *flow, __u64 now) +{ + struct challenge_pending *p = bpf_map_lookup_elem(&challenge_pending, flow); + if (!p) + return CHAL_CONTINUE; + + __u64 timeout_ns = (__u64)G_CHALLENGE_TIMEOUT_MS * 1000000ULL; + __u32 seq = bpf_ntohl(tcp->seq); + __u32 ack = bpf_ntohl(tcp->ack_seq); + __u8 ok = (seq == p->bad_ack) || + (tcp->ack && ack == p->marker + 1); + + bpf_map_delete_elem(&challenge_pending, flow); + + if (!ok || now - p->sent_at > timeout_ns) { + inc_chal_failed(); + return XDP_DROP; + } + + __u64 seen = now; + bpf_map_update_elem(&challenge_verified, &flow->src_ip, &seen, BPF_ANY); + inc_chal_verified(); + return XDP_DROP; // client app will retry connect; next SYN passes +} + +#endif /* RAMPART_SYN_CHALLENGE_H */ diff --git a/xdp/core/universal_filter.c b/xdp/core/universal_filter.c index 019a6d0..b5477c1 100644 --- a/xdp/core/universal_filter.c +++ b/xdp/core/universal_filter.c @@ -27,6 +27,7 @@ #include "config.h" #include "stats.h" #include "prefix_stats.h" +#include "syn_challenge.h" #include "../hooks/hook_api.h" char __license[] SEC("license") = "GPL"; @@ -299,6 +300,15 @@ int rampart_universal_filter(struct xdp_md *ctx) return XDP_DROP; } + // ── RST-challenge: unverified source must prove liveness first ── + // (after throttle/blacklist, mirroring prefix_stats ordering) + if (G_SYN_CHALLENGE_ENABLED) { + int cv = chal_on_syn(ctx, eth, (void *)(eth + 1), is_ipv6, + tcp, &flow, now); + if (cv != CHAL_CONTINUE) + return cv; + } + // Count SYN only after throttle passed (don't count throttled SYNs) update_prefix_stats(&pkey, 1, now); @@ -324,6 +334,13 @@ int rampart_universal_filter(struct xdp_md *ctx) // pkt_count: all non-SYN TCP that survived blacklist/bypass/throttle update_prefix_stats(&pkey, 0, now); + // ── RST-challenge proof: match a pending challenge before conntrack ── + if (G_SYN_CHALLENGE_ENABLED && tcp->rst) { + int cv = chal_on_rst(tcp, &flow, now); + if (cv != CHAL_CONTINUE) + return cv; + } + struct conntrack_entry *conn = bpf_map_lookup_elem(&conntrack_map, &flow); if (!conn) { // Unknown connection — drop