feat: traffic intel hot path, SYN RST-challenge, XDP environment diagnostics

Traffic Intel (was dead code, now wired):
- TrafficHook in listener accept path: cps/pps windows -> AttackDetector
- auto-ban IPs below reputation threshold under attack ([detect.autoban])
- AlertDispatcher: webhook on attack state transition only (dedup), metrics
  AUTO_BANS_TOTAL / INTEL_* ; [detect.alert].webhook_url

XDP SYN RST-challenge (Oubliette pattern, off by default):
- G_SYN_CHALLENGE_ENABLED=0: kernel replies bad-ACK SYN-ACK via XDP_TX,
  spoofed sources stay silent, live clients answer RST with secret echo ->
  challenge_verified (LRU, sliding TTL); brute-force of marker impossible
- maps challenge_verified/challenge_pending, STAT_CHALLENGE_*, all logic
  in xdp/core/syn_challenge.h (221 lines)

XDP diagnostics (src/xdp/diagnostics.rs):
- EnvironmentReport: kernel version/BTF/driver->AttachMode verdict,
  fail-fast before load on unsupported kernels; wired into  CLI
- SystemProbe trait for kernel-less testing

fix: .gitignore 'bin/' matched src/bin/ — rampart.rs was never committed

cargo build/clippy(-D warnings, --features xdp)/test green: 107 tests

.gitignore

@@ -41,7 +41,7 @@ Thumbs.db
 .project
 .settings/
 .factorypath
-bin/
+/bin/
 
 # Reference projects (cloned for research)
 ref/

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<String>,
+        #[arg(required = false)]
+        value: Option<String>,
+    },
+    /// 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<String> },
+    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,
+    }
+}

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::<u64>()
+        .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::<std::net::SocketAddr>();
+    axum::serve(listener, app).await?;
+    Ok(())
+}

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<Arc<tokio::sync::Mutex<ClickHouseWriter>>> = 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<Arc<tokio::sync::Mutex<ClickHouseWriter>>>, 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<Config>,
+    shutdown_rx: &watch::Receiver<bool>,
+) -> anyhow::Result<Option<Arc<Mutex<XdpFilter>>>> {
+    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<Config>,
+    _shutdown_rx: &watch::Receiver<bool>,
+) -> anyhow::Result<Option<Arc<Mutex<XdpFilter>>>> {
+    Ok(None)
+}
+
+/// Юзерспейс-агрегатор префиксов: активен, когда detect.prefix включён,
+/// а XDP-путь не работает (не собран или выключен в конфиге).
+fn start_subnet_tracker(config: &Config) -> Option<Arc<SubnetTracker>> {
+    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<Arc<HashSet<IpAddr>>> {
+    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() => {}
+    }
+}

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<String> {
+    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.");

src/config/mod.rs

@@ -2,8 +2,8 @@
 mod sections;
 
 pub use sections::{
-    BackendConfig, BanConfig, BindConfig, DetectConfig, DetectPrefixConfig, LimitsConfig, LoggingConfig, MetricsConfig,
+    BackendConfig, BanConfig, BindConfig, DetectAlertConfig, DetectAutobanConfig, DetectConfig, DetectPrefixConfig,
-    PowConfig, StoreConfig, WorkerConfig, XdpConfig,
+    LimitsConfig, LoggingConfig, MetricsConfig, PowConfig, StoreConfig, WorkerConfig, XdpConfig,
 };
 
 use serde::Deserialize;

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<String>,
 }
 
 /// Секция `[detect.prefix]`: subnet-level детектор распределённых атак.

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<Config>, gateway: Arc<Gateway>, shutdown: watch::Receiver<bool>) -> anyhow::Result<()> {
+pub async fn run(
-    let _ = &config;
+    config: Arc<Config>,
+    gateway: Arc<Gateway>,
+    hook: Arc<TrafficHook>,
+    shutdown: watch::Receiver<bool>,
+) -> 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<Config>, gateway: Arc<Gateway>, 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<TcpListener> {
     Ok(TcpListener::from_std(socket.into())?)
 }
 
-async fn accept_loop(listener: TcpListener, gateway: Arc<Gateway>, mut shutdown: watch::Receiver<bool>) {
+async fn accept_loop(
+    listener: TcpListener,
+    gateway: Arc<Gateway>,
+    hook: Arc<TrafficHook>,
+    mut shutdown: watch::Receiver<bool>,
+) {
     loop {
         tokio::select! {
             biased;
@@ -62,10 +73,14 @@ async fn accept_loop(listener: TcpListener, gateway: Arc<Gateway>, 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());
                 });
             }
         }

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<IntCounterVec> = LazyLock::new(|| {
     .expect("SUBNET_VERDICTS")
 });
 
+pub static AUTO_BANS_TOTAL: LazyLock<IntCounterVec> = 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<IntCounter> = 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<IntGauge> =
+    LazyLock::new(|| register_int_gauge!("rampart_intel_cps", "New connections per second window").expect("INTEL_CPS"));
+
+pub static INTEL_ALERTS_TOTAL: LazyLock<IntCounter> = 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 {

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<Option<AttackStatus>>,
+}
+
+impl AlertDispatcher {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Возвращает `Some(Alert)` только при переходе через границу атаки:
+    /// вход в `UnderAttack` («attack started») или выход из него
+    /// («attack ended»). Первое наблюдение алерта не порождает.
+    pub fn on_status(&self, status: AttackStatus) -> Option<Alert> {
+        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);
+        }
+    }
 }

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, 100.0);
-        d.analyze(10_000.0);
+        d.analyze(10_000.0, 100.0);
-        assert_eq!(d.analyze(10_000.0), AttackStatus::UnderAttack);
+        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, 10.0);
-        d.analyze(100.0);
+        d.analyze(100.0, 10.0);
-        d.analyze(1.0);
+        d.analyze(1.0, 1.0);
-        assert_eq!(d.analyze(1.0), AttackStatus::Normal);
+        assert_eq!(d.analyze(1.0, 1.0), AttackStatus::Normal);
     }
 }

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<IpReputation>,
+    blacklist: Arc<Blacklist>,
+    autoban: DetectAutobanConfig,
+    ban_ttl_secs: u64,
+    conn_window: AtomicU64,
+}
+
+impl TrafficHook {
+    pub fn new(
+        reputation: Arc<IpReputation>,
+        blacklist: Arc<Blacklist>,
+        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<IpReputation>, bl: &Arc<Blacklist>) -> 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");
+    }
+}

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;

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<IpAddr> {
+        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());
+    }
 }

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<Self> {
+        let mut nums = release
+            .split(|c: char| !c.is_ascii_digit())
+            .filter_map(|part| part.parse::<u16>().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<String>;
+
+    #[must_use]
+    fn path_exists(&self, path: &str) -> bool;
+
+    #[must_use]
+    fn symlink_target(&self, path: &str) -> Option<String>;
+
+    #[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<String> {
+        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<String> {
+        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<u32> {
+    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<String> {
+    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<KernelVersion>,
+    pub btf_available: bool,
+    pub privileged: bool,
+    pub driver_name: Option<String>,
+    pub warnings: Vec<String>,
+}
+
+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()
+}

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)

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")]

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")
+    );
+}

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"));
+}

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<String, String>,
+    symlinks: HashMap<String, String>,
+    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<String> {
+        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<String> {
+        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}");
+}

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 */

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);

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); }
 

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 */

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