v0.3: 6-layer architecture complete

Layers:
  Layer 1: XDP/eBPF — TCP state machine, SYN throttle, blacklist, ringbuf
  Layer 2: PoW Challenge — SHA-256 hashcash, dynamic difficulty, constant-time verify
  Layer 3: Rust Core — HMAC handshake, rate limit, death code (existing)
  Layer 4: Velocity — Physics check, CAPTCHA, protocol verification
  Layer 5: Paper — Heartbeat, auto-registration (existing)
  Layer 6: Traffic Intel — EWMA, 168h profiling, reputation, alerts

Infra: XDP→Prometheus metrics, ClickHouse + Grafana dashboard, Docker Compose
Testing: 100-IP DDoS simulation, MHDDoS ref analysis, load test report
Fixes: VarInt sign extension UB, pure ACK deadlock, RST/FIN cleanup
Ref: MHDDoS, Sonar, LimboFilter, AtomGuard, Infrarust, MC-XDP-eBPF, PowGo

crates/rampart-core/Cargo.toml

@@ -19,21 +19,25 @@ dashmap.workspace = true
 crossbeam.workspace = true
 hex.workspace = true
 sha2.workspace = true
+rand.workspace = true
 hmac.workspace = true
 subtle.workspace = true
 socket2 = { workspace = true, features = ["all"] }
 prometheus.workspace = true
 toml.workspace = true
 futures.workspace = true
+chrono = { workspace = true, features = ["serde"] }
+reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] }
 
 redis = { version = "0.27", optional = true, features = ["tokio-comp"] }
 maxminddb = { version = "0.30", optional = true }
 tokio-splice = { version = "0.2", optional = true }
 libbpf-rs = { version = "0.24", optional = true }
+libc = { version = "0.2", optional = true }
 
 [features]
 default = ["store-redis"]
 store-redis = ["dep:redis"]
 geoip = ["dep:maxminddb"]
-xdp = ["dep:libbpf-rs"]
+xdp = ["dep:libbpf-rs", "dep:libc"]
 io-uring = ["dep:tokio-splice"]

crates/rampart-core/build.rs

@@ -0,0 +1,42 @@
+use std::path::PathBuf;
+use std::process::Command;
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let has_xdp_feature = std::env::var("CARGO_FEATURE_XDP").is_ok();
+    if !has_xdp_feature {
+        return Ok(());
+    }
+
+    let manifest_dir = PathBuf::from(std::env::var("CARGO_MANIFEST_DIR")?);
+    let xdp_dir = manifest_dir.join("../../xdp");
+    let out_dir = PathBuf::from(std::env::var("OUT_DIR")?);
+
+    let src = xdp_dir.join("xdp_filter.c");
+    let dst = out_dir.join("xdp_filter.o");
+
+    println!("cargo:rerun-if-changed={}", src.display());
+
+    let host_arch = std::env::var("HOST").unwrap_or_default();
+    let status = Command::new("clang")
+        .args([
+            "-O2",
+            "-g",
+            "-target",
+            "bpf",
+            "-mcpu=v3",
+            "-c",
+            src.to_str().ok_or("src path is not valid UTF-8")?,
+            "-o",
+            dst.to_str().ok_or("dst path is not valid UTF-8")?,
+            &format!("-I{}", xdp_dir.display()),
+            &format!("-I/usr/include/{}-linux-gnu", host_arch),
+        ])
+        .status()?;
+
+    if !status.success() {
+        return Err("XDP C compilation failed (see clang errors above)".into());
+    }
+
+    println!("cargo:rerun-if-env-changed=CARGO_FEATURE_XDP");
+    Ok(())
+}

crates/rampart-core/src/config.rs

@@ -23,6 +23,10 @@ pub struct Config {
     pub logging: LoggingConfig,
     #[serde(default)]
     pub metrics: MetricsConfig,
+    #[serde(default)]
+    pub pow: PowConfig,
+    #[serde(default)]
+    pub whitelist: Vec<String>,
 }
 
 #[derive(Debug, Clone, Default, Deserialize)]
@@ -139,6 +143,7 @@ pub struct StoreConfig {
     pub redis_url: Option<String>,
     #[serde(default = "default_blacklist_cache_ttl")]
     pub blacklist_cache_ttl_secs: u64,
+    pub clickhouse_url: Option<String>,
 }
 
 fn default_blacklist_cache_ttl() -> u64 {
@@ -150,6 +155,7 @@ impl Default for StoreConfig {
         Self {
             redis_url: None,
             blacklist_cache_ttl_secs: 300,
+            clickhouse_url: None,
         }
     }
 }
@@ -247,6 +253,31 @@ impl Default for DeathCodeConfig {
     }
 }
 
+#[derive(Debug, Clone, Deserialize)]
+pub struct PowConfig {
+    #[serde(default = "default_pow_enabled")]
+    pub enabled: bool,
+    #[serde(default = "default_pow_difficulty")]
+    pub difficulty: u8,
+}
+
+fn default_pow_enabled() -> bool {
+    true
+}
+
+fn default_pow_difficulty() -> u8 {
+    4
+}
+
+impl Default for PowConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            difficulty: 4,
+        }
+    }
+}
+
 impl Config {
     pub fn from_file(path: &str) -> anyhow::Result<Self> {
         let contents = fs::read_to_string(path)?;

crates/rampart-core/src/filter/geo.rs

@@ -1,5 +1,6 @@
 #[cfg(feature = "geoip")]
 pub struct GeoIp {
+    #[allow(dead_code)]
     reader: maxminddb::Reader<Vec<u8>>,
 }
 

crates/rampart-core/src/lib.rs

@@ -2,8 +2,10 @@ pub mod config;
 pub mod crypto;
 pub mod filter;
 pub mod metrics;
+pub mod pow;
 pub mod proxy;
 pub mod store;
+pub mod traffic;
 
 #[cfg(feature = "xdp")]
 pub mod xdp;

crates/rampart-core/src/main.rs

@@ -2,8 +2,9 @@ use rampart_core::config::Config;
 use rampart_core::filter::blacklist::Blacklist;
 use rampart_core::filter::rate_limit::RateLimiter;
 use rampart_core::metrics;
+use rampart_core::pow::difficulty::DifficultyAdjuster;
 use rampart_core::proxy::listener::ProxyListener;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::sync::watch;
 use tracing_subscriber::EnvFilter;
@@ -63,10 +64,32 @@ async fn main() -> anyhow::Result<()> {
         });
     }
 
+    #[cfg(feature = "xdp")]
+    if config.xdp.enabled {
+        use rampart_core::xdp::{XdpFilter, XdpMetrics};
+
+        let mut filter = XdpFilter::new(&config.xdp.interface);
+        filter.load()?;
+        let xdp_metrics = XdpMetrics::register()?;
+
+        let sd = shutdown_rx.clone();
+        std::thread::spawn(move || {
+            while !*sd.borrow() {
+                filter.drain_events();
+                if let Ok(stats) = filter.get_stats() {
+                    xdp_metrics.update(&stats);
+                }
+                std::thread::sleep(Duration::from_secs(5));
+            }
+            filter.unload().ok();
+        });
+    }
+
     tracing::info!("Rampart edge starting on {}:{}", config.bind.address, config.bind.port);
     tracing::info!("Backend: {}:{}", config.backend.address, config.backend.port);
 
-    let listener = ProxyListener::new(config, rate_limiter, blacklist);
+    let adjuster = Arc::new(Mutex::new(DifficultyAdjuster::default()));
+    let listener = ProxyListener::new(config, rate_limiter, blacklist, adjuster);
     listener.run(shutdown_rx).await
 }
 

crates/rampart-core/src/metrics.rs

@@ -24,6 +24,15 @@ pub static DEATH_CODE_BANS_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
         .expect("DEATH_CODE_BANS_TOTAL")
 });
 
+pub static POW_CHALLENGES_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    register_int_counter_vec!("rampart_pow_challenges_total", "PoW challenges issued", &["result"])
+        .expect("POW_CHALLENGES_TOTAL")
+});
+
+pub static POW_CURRENT_DIFFICULTY: LazyLock<IntGauge> = LazyLock::new(|| {
+    register_int_gauge!("rampart_pow_current_difficulty", "Current PoW difficulty").expect("POW_CURRENT_DIFFICULTY")
+});
+
 pub async fn run_metrics_server(addr: &str) {
     let listener = match TcpListener::bind(addr).await {
         Ok(l) => l,

crates/rampart-core/src/pow/challenge.rs

@@ -0,0 +1,30 @@
+use rand::RngCore;
+use std::time::Instant;
+
+pub struct Challenge {
+    pub token: [u8; 32],
+    pub created_at: Instant,
+    pub difficulty: u8,
+    pub used: bool,
+}
+
+impl Challenge {
+    pub fn generate(difficulty: u8) -> Self {
+        let mut token = [0u8; 32];
+        rand::thread_rng().fill_bytes(&mut token);
+        Self {
+            token,
+            created_at: Instant::now(),
+            difficulty,
+            used: false,
+        }
+    }
+
+    pub fn is_expired(&self) -> bool {
+        self.created_at.elapsed().as_secs() >= 30
+    }
+
+    pub fn challenge_string(&self) -> String {
+        hex::encode(self.token)
+    }
+}

crates/rampart-core/src/pow/difficulty.rs

@@ -0,0 +1,68 @@
+use crate::metrics;
+use std::collections::VecDeque;
+use std::time::Instant;
+
+pub struct DifficultyAdjuster {
+    window: VecDeque<Instant>,
+    min: u8,
+    max: u8,
+    current: u8,
+}
+
+impl DifficultyAdjuster {
+    pub fn new(min: u8, max: u8) -> Self {
+        Self {
+            window: VecDeque::new(),
+            min: min.max(4),
+            max: max.min(10),
+            current: min.max(4),
+        }
+    }
+
+    pub fn record_connection(&mut self) {
+        let now = Instant::now();
+        self.window.push_back(now);
+        while let Some(&t) = self.window.front() {
+            if now.duration_since(t).as_secs() >= 1 {
+                self.window.pop_front();
+            } else {
+                break;
+            }
+        }
+        let new_diff = self.compute_difficulty();
+        if self.current != new_diff {
+            tracing::info!(
+                old = self.current,
+                new = new_diff,
+                window = self.window.len(),
+                "pow: difficulty adjusted"
+            );
+            self.current = new_diff;
+            metrics::POW_CURRENT_DIFFICULTY.set(self.current as i64);
+        }
+    }
+
+    pub fn current_difficulty(&self) -> u8 {
+        metrics::POW_CURRENT_DIFFICULTY.set(self.current as i64);
+        self.current
+    }
+
+    fn compute_difficulty(&self) -> u8 {
+        let cps = self.window.len();
+        if cps > 500 {
+            self.max.max(self.min)
+        } else if cps > 200 {
+            8
+        } else if cps > 50 {
+            6
+        } else {
+            self.min
+        }
+    }
+}
+
+impl Default for DifficultyAdjuster {
+    fn default() -> Self {
+        Self::new(4, 16)
+    }
+}

crates/rampart-core/src/pow/mod.rs

@@ -0,0 +1,4 @@
+pub mod challenge;
+pub mod difficulty;
+pub mod solver;
+pub mod verifier;

crates/rampart-core/src/pow/solver.rs

@@ -0,0 +1,17 @@
+use sha2::{Digest, Sha256};
+
+const ALLOWED: &[u8] = b"0123";
+
+pub fn solve(challenge: &str, difficulty: u8) -> Option<String> {
+    let d = difficulty as usize;
+    for nonce in 0..u64::MAX {
+        let nonce_str = nonce.to_string();
+        let input = format!("{challenge}{nonce_str}");
+        let hash = Sha256::digest(input.as_bytes());
+        let hex_hash = hex::encode(hash);
+        if hex_hash.as_bytes().iter().take(d).all(|c| ALLOWED.contains(c)) {
+            return Some(nonce_str);
+        }
+    }
+    None
+}

crates/rampart-core/src/pow/verifier.rs

@@ -0,0 +1,31 @@
+use crate::pow::challenge::Challenge;
+use sha2::{Digest, Sha256};
+use subtle::ConstantTimeEq;
+
+const ALLOWED: [u8; 4] = [b'0', b'1', b'2', b'3'];
+
+pub fn verify(challenge: &mut Challenge, nonce: &str) -> bool {
+    if challenge.used {
+        return false;
+    }
+    if challenge.is_expired() {
+        return false;
+    }
+    if nonce.len() > 64 {
+        return false;
+    }
+
+    let input = format!("{}{}", challenge.challenge_string(), nonce);
+    let hash = Sha256::digest(input.as_bytes());
+    let hex_hash = hex::encode(hash);
+    let d = challenge.difficulty as usize;
+    let ok = hex_hash.as_bytes().iter().take(d).all(|c| {
+        let r = c.ct_eq(&ALLOWED[0]) | c.ct_eq(&ALLOWED[1]) | c.ct_eq(&ALLOWED[2]) | c.ct_eq(&ALLOWED[3]);
+        r.unwrap_u8() == 1
+    });
+    if !ok {
+        return false;
+    }
+    challenge.used = true;
+    true
+}

crates/rampart-core/src/proxy/listener.rs

@@ -1,9 +1,10 @@
 use crate::config::Config;
 use crate::filter::blacklist::Blacklist;
 use crate::filter::rate_limit::RateLimiter;
+use crate::pow::difficulty::DifficultyAdjuster;
 use crate::proxy::tunnel::ConnectionHandler;
 use socket2::{Domain, Socket, Type};
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
 use tokio::net::TcpListener;
 use tokio::sync::watch;
 
@@ -11,14 +12,21 @@ pub struct ProxyListener {
     config: Arc<Config>,
     rate_limiter: Arc<RateLimiter>,
     blacklist: Arc<Blacklist>,
+    adjuster: Arc<Mutex<DifficultyAdjuster>>,
 }
 
 impl ProxyListener {
-    pub fn new(config: Arc<Config>, rate_limiter: Arc<RateLimiter>, blacklist: Arc<Blacklist>) -> Self {
+    pub fn new(
+        config: Arc<Config>,
+        rate_limiter: Arc<RateLimiter>,
+        blacklist: Arc<Blacklist>,
+        adjuster: Arc<Mutex<DifficultyAdjuster>>,
+    ) -> Self {
         Self {
             config,
             rate_limiter,
             blacklist,
+            adjuster,
         }
     }
 
@@ -32,12 +40,14 @@ impl ProxyListener {
             let config = self.config.clone();
             let rate_limiter = self.rate_limiter.clone();
             let blacklist = self.blacklist.clone();
+            let adjuster = self.adjuster.clone();
             let shutdown = shutdown.clone();
             handles.push(tokio::spawn(accept_loop(
                 listener,
                 config,
                 rate_limiter,
                 blacklist,
+                adjuster,
                 shutdown,
             )));
         }
@@ -64,6 +74,7 @@ async fn accept_loop(
     config: Arc<Config>,
     rate_limiter: Arc<RateLimiter>,
     blacklist: Arc<Blacklist>,
+    adjuster: Arc<Mutex<DifficultyAdjuster>>,
     mut shutdown: watch::Receiver<bool>,
 ) -> anyhow::Result<()> {
     loop {
@@ -83,7 +94,7 @@ async fn accept_loop(
                         continue;
                     }
                 };
-                let handler = ConnectionHandler::new(config.clone(), rate_limiter.clone(), blacklist.clone());
+                let handler = ConnectionHandler::new(config.clone(), rate_limiter.clone(), blacklist.clone(), adjuster.clone());
                 tokio::spawn(async move {
                     if let Err(e) = handler.handle(stream, peer_addr).await {
                         tracing::debug!("connection from {peer_addr}: {e}");

crates/rampart-core/src/proxy/mod.rs

@@ -1,3 +1,4 @@
 pub mod handshake;
 pub mod listener;
+pub mod pow;
 pub mod tunnel;

crates/rampart-core/src/proxy/pow.rs

@@ -0,0 +1,37 @@
+use crate::pow::challenge::Challenge;
+use std::net::Ipv4Addr;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::time::{Duration, timeout};
+
+pub async fn handle_pow(stream: &mut TcpStream, peer_ip: Ipv4Addr, difficulty: u8) -> anyhow::Result<bool> {
+    if difficulty == 0 {
+        tracing::debug!("pow: difficulty 0, skipping for {peer_ip}");
+        return Ok(true);
+    }
+
+    let mut challenge = Challenge::generate(difficulty);
+    let challenge_str = challenge.challenge_string();
+    let line = format!("{challenge_str}\n");
+    stream.write_all(line.as_bytes()).await?;
+
+    let mut buf = [0u8; 65];
+    let n = timeout(Duration::from_secs(10), stream.read(&mut buf)).await??;
+    if n == 0 {
+        tracing::debug!("pow: no response from {peer_ip}");
+        return Ok(false);
+    }
+
+    let nonce = std::str::from_utf8(&buf[..n.min(64)]).unwrap_or("").trim();
+    if nonce.is_empty() || nonce.len() > 64 {
+        tracing::debug!("pow: invalid nonce from {peer_ip}");
+        return Ok(false);
+    }
+
+    let valid = crate::pow::verifier::verify(&mut challenge, nonce);
+    tracing::debug!(
+        "pow: verification {} for {peer_ip}",
+        if valid { "passed" } else { "failed" }
+    );
+    Ok(valid)
+}

crates/rampart-core/src/proxy/tunnel.rs

@@ -4,8 +4,11 @@ use crate::filter::blacklist::Blacklist;
 use crate::filter::death_code;
 use crate::filter::rate_limit::RateLimiter;
 use crate::metrics;
+use crate::pow::difficulty::DifficultyAdjuster;
 use crate::proxy::handshake::{McHandshake, read_varint};
-use std::sync::Arc;
+use crate::proxy::pow::handle_pow;
+use std::net::Ipv4Addr;
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::TcpStream;
@@ -14,14 +17,21 @@ pub struct ConnectionHandler {
     config: Arc<Config>,
     rate_limiter: Arc<RateLimiter>,
     blacklist: Arc<Blacklist>,
+    adjuster: Arc<Mutex<DifficultyAdjuster>>,
 }
 
 impl ConnectionHandler {
-    pub fn new(config: Arc<Config>, rate_limiter: Arc<RateLimiter>, blacklist: Arc<Blacklist>) -> Self {
+    pub fn new(
+        config: Arc<Config>,
+        rate_limiter: Arc<RateLimiter>,
+        blacklist: Arc<Blacklist>,
+        adjuster: Arc<Mutex<DifficultyAdjuster>>,
+    ) -> Self {
         Self {
             config,
             rate_limiter,
             blacklist,
+            adjuster,
         }
     }
 
@@ -40,6 +50,31 @@ impl ConnectionHandler {
             return Ok(());
         }
 
+        let pow_config = &self.config.pow;
+        let peer_ip = Ipv4Addr::from_bits(ip_u32);
+        if pow_config.enabled && pow_config.difficulty > 0 && !self.config.whitelist.contains(&peer_ip.to_string()) {
+            self.adjuster
+                .lock()
+                .expect("adjuster lock poisoned")
+                .record_connection();
+            let diff = self
+                .adjuster
+                .lock()
+                .expect("adjuster lock poisoned")
+                .current_difficulty();
+            let result = handle_pow(&mut client, peer_ip, diff).await?;
+            if !result {
+                metrics::POW_CHALLENGES_TOTAL.with_label_values(&["failed"]).inc();
+                tracing::debug!("pow: failed for {peer_ip}, dropping connection");
+                return Ok(());
+            }
+            metrics::POW_CHALLENGES_TOTAL.with_label_values(&["passed"]).inc();
+            metrics::POW_CURRENT_DIFFICULTY.set(diff as i64);
+        } else if pow_config.enabled && pow_config.difficulty > 0 {
+            metrics::POW_CHALLENGES_TOTAL.with_label_values(&["skipped"]).inc();
+            metrics::POW_CURRENT_DIFFICULTY.set(pow_config.difficulty as i64);
+        }
+
         if !self.rate_limiter.check(ip_u32) {
             metrics::RATE_LIMIT_HITS.with_label_values(&["hit"]).inc();
             metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();

crates/rampart-core/src/store/clickhouse.rs

@@ -0,0 +1,89 @@
+use chrono::{DateTime, Utc};
+use serde::Serialize;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tokio::sync::watch;
+
+const BATCH_SIZE: usize = 1000;
+const FLUSH_INTERVAL: Duration = Duration::from_secs(1);
+
+#[derive(Debug, Clone, Serialize)]
+pub struct ClickHouseEvent {
+    pub timestamp: DateTime<Utc>,
+    pub event_type: String,
+    pub ip: String,
+    pub data_float: f64,
+    pub data_int: i64,
+    pub data_string: String,
+}
+
+pub struct ClickHouseWriter {
+    url: String,
+    client: reqwest::Client,
+    buffer: Vec<ClickHouseEvent>,
+}
+
+impl ClickHouseWriter {
+    pub fn new(url: &str) -> Self {
+        Self {
+            url: url.to_string(),
+            client: reqwest::Client::new(),
+            buffer: Vec::with_capacity(BATCH_SIZE),
+        }
+    }
+
+    pub async fn push(&mut self, event: ClickHouseEvent) -> anyhow::Result<()> {
+        self.buffer.push(event);
+        if self.buffer.len() >= BATCH_SIZE {
+            self.flush().await?;
+        }
+        Ok(())
+    }
+
+    pub async fn flush(&mut self) -> anyhow::Result<()> {
+        if self.buffer.is_empty() {
+            return Ok(());
+        }
+        let events = std::mem::take(&mut self.buffer);
+        let json = serde_json::to_string(&events)?;
+        let response = self
+            .client
+            .post(&self.url)
+            .query(&[("query", "INSERT INTO rampart_events FORMAT JSONEachRow")])
+            .header("Content-Type", "application/json")
+            .body(json)
+            .send()
+            .await?;
+        let status = response.status();
+        if !status.is_success() {
+            let text = response.text().await?;
+            anyhow::bail!("clickhouse insert failed ({}): {}", status, text);
+        }
+        tracing::debug!("flushed {} events to clickhouse", events.len());
+        Ok(())
+    }
+}
+
+pub fn start_flush_task(writer: Arc<Mutex<ClickHouseWriter>>, mut shutdown: watch::Receiver<bool>) {
+    tokio::spawn(async move {
+        loop {
+            tokio::select! {
+                _ = tokio::time::sleep(FLUSH_INTERVAL) => {
+                    if let Err(e) = writer.lock().await.flush().await {
+                        tracing::error!("clickhouse flush error: {e}");
+                    }
+                }
+                _ = shutdown.changed() => {
+                    if *shutdown.borrow() {
+                        tracing::info!("flushing clickhouse on shutdown");
+                        if let Err(e) = writer.lock().await.flush().await {
+                            tracing::error!("clickhouse final flush error: {e}");
+                        }
+                        return;
+                    }
+                }
+            }
+        }
+    });
+}

crates/rampart-core/src/store/mod.rs

@@ -3,6 +3,8 @@ pub mod redis;
 #[cfg(feature = "store-redis")]
 pub use redis::start_blacklist_sync;
 
+pub mod clickhouse;
+
 #[allow(async_fn_in_trait)]
 pub trait StateStore: Send + Sync {
     async fn get(&self, key: &str) -> anyhow::Result<Option<String>>;

crates/rampart-core/src/traffic/alert.rs

@@ -0,0 +1,80 @@
+use std::fmt;
+use std::net::Ipv4Addr;
+use std::time::Instant;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AlertLevel {
+    Info,
+    Warning,
+    Critical,
+}
+
+impl fmt::Display for AlertLevel {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            AlertLevel::Info => write!(f, "INFO"),
+            AlertLevel::Warning => write!(f, "WARNING"),
+            AlertLevel::Critical => write!(f, "CRITICAL"),
+        }
+    }
+}
+
+#[derive(Clone)]
+pub struct Alert {
+    pub level: AlertLevel,
+    pub message: String,
+    pub ip: Option<Ipv4Addr>,
+    pub pps: f64,
+    pub timestamp: Instant,
+}
+
+impl Alert {
+    pub fn new(level: AlertLevel, message: String, ip: Option<Ipv4Addr>, pps: f64) -> Self {
+        Self {
+            level,
+            message,
+            ip,
+            pps,
+            timestamp: Instant::now(),
+        }
+    }
+}
+
+impl fmt::Display for Alert {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match &self.ip {
+            Some(ip) => write!(
+                f,
+                "[{}] {} | IP: {} | PPS: {:.2}",
+                self.level, self.message, ip, self.pps
+            ),
+            None => write!(f, "[{}] {} | PPS: {:.2}", self.level, self.message, self.pps),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_alert_display_with_ip() {
+        let alert = Alert::new(
+            AlertLevel::Critical,
+            "possible attack".into(),
+            Some(Ipv4Addr::new(192, 168, 1, 1)),
+            100500.0,
+        );
+        let s = alert.to_string();
+        assert!(s.contains("CRITICAL"));
+        assert!(s.contains("192.168.1.1"));
+    }
+
+    #[test]
+    fn test_alert_display_without_ip() {
+        let alert = Alert::new(AlertLevel::Info, "traffic spike".into(), None, 5000.0);
+        let s = alert.to_string();
+        assert!(s.contains("INFO"));
+        assert!(s.contains("traffic spike"));
+    }
+}

crates/rampart-core/src/traffic/detector.rs

@@ -0,0 +1,93 @@
+use crate::traffic::profiler::TrafficProfiler;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum AttackStatus {
+    Normal,
+    Suspicious,
+    UnderAttack,
+}
+
+pub struct AttackDetector {
+    profiler: TrafficProfiler,
+    consecutive_anomalies: u32,
+}
+
+impl Default for AttackDetector {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl AttackDetector {
+    pub fn new() -> Self {
+        Self {
+            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);
+
+        if score > 3.0 {
+            self.consecutive_anomalies += 1;
+        } else {
+            self.consecutive_anomalies = 0;
+        }
+
+        if self.consecutive_anomalies >= 3 {
+            AttackStatus::UnderAttack
+        } else if score > 2.0 {
+            AttackStatus::Suspicious
+        } else {
+            AttackStatus::Normal
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_attack_status_normal() {
+        let mut d = AttackDetector::new();
+        for _ in 0..10 {
+            d.analyze(100.0);
+        }
+        assert_eq!(d.analyze(100.0), AttackStatus::Normal);
+    }
+
+    #[test]
+    fn test_attack_status_suspicious() {
+        let mut d = AttackDetector::new();
+        for _ in 0..10 {
+            d.analyze(1.0);
+        }
+        assert_eq!(d.analyze(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(10_000.0);
+        d.analyze(10_000.0);
+        assert_eq!(d.analyze(10_000.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(100.0);
+        d.analyze(100.0);
+        d.analyze(1.0);
+        assert_eq!(d.analyze(1.0), AttackStatus::Normal);
+    }
+}

crates/rampart-core/src/traffic/ewma.rs

@@ -0,0 +1,57 @@
+use std::time::Instant;
+
+pub struct Ewma {
+    value: f64,
+    last: Instant,
+    alpha: f64,
+}
+
+impl Ewma {
+    pub fn new(alpha: f64) -> Self {
+        Self {
+            value: 0.0,
+            last: Instant::now(),
+            alpha,
+        }
+    }
+
+    pub fn update(&mut self, sample: f64) {
+        let now = Instant::now();
+        let elapsed = now.duration_since(self.last).as_secs_f64();
+        let steps = elapsed.max(1.0);
+        let weight = (1.0 - self.alpha).powf(steps);
+        self.value = self.value * weight + sample * (1.0 - weight);
+        self.last = now;
+    }
+
+    pub fn value(&self) -> f64 {
+        self.value
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_ewma_initial() {
+        let ewma = Ewma::new(0.125);
+        assert_eq!(ewma.value(), 0.0);
+    }
+
+    #[test]
+    fn test_ewma_update() {
+        let mut ewma = Ewma::new(1.0);
+        ewma.update(100.0);
+        assert!((ewma.value() - 100.0).abs() < 1.0);
+    }
+
+    #[test]
+    fn test_ewma_convergence() {
+        let mut ewma = Ewma::new(0.5);
+        for _ in 0..10 {
+            ewma.update(50.0);
+        }
+        assert!((ewma.value() - 50.0).abs() < 1.0);
+    }
+}

crates/rampart-core/src/traffic/mod.rs

@@ -0,0 +1,5 @@
+pub mod alert;
+pub mod detector;
+pub mod ewma;
+pub mod profiler;
+pub mod reputation;

crates/rampart-core/src/traffic/profiler.rs

@@ -0,0 +1,85 @@
+use crate::traffic::ewma::Ewma;
+use std::time::Instant;
+
+pub struct TrafficProfiler {
+    slots: [Ewma; 168],
+    current_slot: usize,
+    epoch: Instant,
+}
+
+impl Default for TrafficProfiler {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl TrafficProfiler {
+    pub fn new() -> Self {
+        Self {
+            slots: std::array::from_fn(|_| Ewma::new(0.125)),
+            current_slot: 0,
+            epoch: Instant::now(),
+        }
+    }
+
+    fn slot_index(&self) -> usize {
+        (self.epoch.elapsed().as_secs() / 3600) as usize % 168
+    }
+
+    pub fn record(&mut self, pps: f64) {
+        self.current_slot = self.slot_index();
+        self.slots[self.current_slot].update(pps);
+    }
+
+    pub fn baseline(&self) -> f64 {
+        self.slots[self.slot_index()].value()
+    }
+
+    pub fn anomaly_score(&self, pps: f64) -> f64 {
+        let base = self.baseline();
+        if base <= 0.0 {
+            return 0.0;
+        }
+        (pps / base).min(10.0)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_profiler_record_and_baseline() {
+        let mut p = TrafficProfiler::new();
+        p.record(1000.0);
+        p.record(1100.0);
+        let base = p.baseline();
+        assert!(base > 0.0);
+    }
+
+    #[test]
+    fn test_anomaly_score_normal() {
+        let mut p = TrafficProfiler::new();
+        for _ in 0..10 {
+            p.record(100.0);
+        }
+        let score = p.anomaly_score(100.0);
+        assert!(score < 2.0);
+    }
+
+    #[test]
+    fn test_anomaly_score_capped() {
+        let mut p = TrafficProfiler::new();
+        for _ in 0..10 {
+            p.record(1.0);
+        }
+        let score = p.anomaly_score(1_000_000.0);
+        assert!((score - 10.0).abs() < 0.001);
+    }
+
+    #[test]
+    fn test_anomaly_score_zero_baseline() {
+        let p = TrafficProfiler::new();
+        assert_eq!(p.anomaly_score(100.0), 0.0);
+    }
+}

crates/rampart-core/src/traffic/reputation.rs

@@ -0,0 +1,98 @@
+use dashmap::DashMap;
+use std::net::Ipv4Addr;
+use std::sync::Arc;
+
+pub struct IpReputation {
+    scores: Arc<DashMap<Ipv4Addr, i32>>,
+}
+
+impl Default for IpReputation {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl IpReputation {
+    pub fn new() -> Self {
+        Self {
+            scores: Arc::new(DashMap::new()),
+        }
+    }
+
+    pub fn record_good(&self, ip: Ipv4Addr) {
+        let mut entry = self.scores.entry(ip).or_insert(0);
+        *entry = (*entry + 1).min(100);
+    }
+
+    pub fn record_bad(&self, ip: Ipv4Addr) {
+        let mut entry = self.scores.entry(ip).or_insert(0);
+        *entry = (*entry - 10).max(-100);
+    }
+
+    pub fn score(&self, ip: Ipv4Addr) -> i32 {
+        self.scores.get(&ip).map(|v| *v).unwrap_or(0)
+    }
+
+    pub fn is_trusted(&self, ip: Ipv4Addr) -> bool {
+        self.score(ip) > 50
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::Ipv4Addr;
+
+    #[test]
+    fn test_reputation_initial_score() {
+        let rep = IpReputation::new();
+        assert_eq!(rep.score(Ipv4Addr::new(192, 168, 1, 1)), 0);
+    }
+
+    #[test]
+    fn test_reputation_good() {
+        let rep = IpReputation::new();
+        let ip = Ipv4Addr::new(10, 0, 0, 1);
+        rep.record_good(ip);
+        assert_eq!(rep.score(ip), 1);
+    }
+
+    #[test]
+    fn test_reputation_bad() {
+        let rep = IpReputation::new();
+        let ip = Ipv4Addr::new(10, 0, 0, 2);
+        rep.record_bad(ip);
+        assert_eq!(rep.score(ip), -10);
+    }
+
+    #[test]
+    fn test_reputation_cap_positive() {
+        let rep = IpReputation::new();
+        let ip = Ipv4Addr::new(10, 0, 0, 3);
+        for _ in 0..200 {
+            rep.record_good(ip);
+        }
+        assert_eq!(rep.score(ip), 100);
+    }
+
+    #[test]
+    fn test_reputation_cap_negative() {
+        let rep = IpReputation::new();
+        let ip = Ipv4Addr::new(10, 0, 0, 4);
+        for _ in 0..20 {
+            rep.record_bad(ip);
+        }
+        assert_eq!(rep.score(ip), -100);
+    }
+
+    #[test]
+    fn test_is_trusted() {
+        let rep = IpReputation::new();
+        let ip = Ipv4Addr::new(10, 0, 0, 5);
+        assert!(!rep.is_trusted(ip));
+        for _ in 0..51 {
+            rep.record_good(ip);
+        }
+        assert!(rep.is_trusted(ip));
+    }
+}

crates/rampart-core/src/xdp/filter.rs

@@ -0,0 +1,151 @@
+use anyhow::{Context, Result, bail};
+use libbpf_rs::{MapCore, MapFlags, Object, ObjectBuilder, RingBuffer, RingBufferBuilder, Xdp, XdpFlags};
+use std::net::Ipv4Addr;
+use std::os::unix::io::AsFd;
+
+use super::XdpStats;
+
+pub struct XdpFilter {
+    obj: Option<Object>,
+    ringbuf: Option<RingBuffer<'static>>,
+    ifindex: i32,
+    interface: String,
+}
+
+impl XdpFilter {
+    pub fn new(interface: &str) -> Self {
+        Self {
+            obj: None,
+            ringbuf: None,
+            ifindex: 0,
+            interface: interface.to_string(),
+        }
+    }
+
+    pub fn load(&mut self) -> Result<()> {
+        let bpf_obj = include_bytes!(concat!(env!("OUT_DIR"), "/xdp_filter.o"));
+        let obj = ObjectBuilder::default()
+            .open_memory(bpf_obj)
+            .context("Failed to open XDP object")?
+            .load()
+            .context("Failed to load XDP object (verifier error?)")?;
+
+        let ifindex = unsafe { libc::if_nametoindex(self.interface.as_ptr() as *const libc::c_char) };
+        if ifindex == 0 {
+            bail!("interface '{}' not found", self.interface);
+        }
+
+        let prog = obj
+            .progs()
+            .find(|p| p.name() == "rampart_xdp_filter")
+            .context("XDP program 'rampart_xdp_filter' not found")?;
+        Xdp::new(prog.as_fd()).attach(ifindex as i32, XdpFlags::NONE)?;
+
+        let rbuf = build_ringbuf(&obj)?;
+
+        self.obj = Some(obj);
+        self.ringbuf = Some(rbuf);
+        self.ifindex = ifindex as i32;
+        tracing::info!("XDP filter attached to {}", self.interface);
+        Ok(())
+    }
+
+    pub fn unload(&mut self) -> Result<()> {
+        if self.ifindex != 0 {
+            let fd = unsafe { std::os::unix::io::BorrowedFd::borrow_raw(std::os::unix::io::RawFd::from(-1)) };
+            let _ = Xdp::new(fd).detach(self.ifindex, XdpFlags::NONE);
+        }
+        self.ringbuf = None;
+        self.obj = None;
+        self.ifindex = 0;
+        tracing::info!("XDP filter detached from {}", self.interface);
+        Ok(())
+    }
+
+    pub fn drain_events(&self) {
+        if let Some(rb) = &self.ringbuf {
+            let _ = rb.consume();
+        }
+    }
+
+    fn find_map<'a>(&'a self, name: &str) -> Result<impl MapCore + 'a> {
+        self.obj
+            .as_ref()
+            .context("XDP not loaded")?
+            .maps()
+            .find(|m| m.name() == name)
+            .with_context(|| format!("map '{}' not found", name))
+    }
+
+    pub fn ban_ip(&self, ip: Ipv4Addr) -> Result<()> {
+        let map = self.find_map("blacklist_map")?;
+        let mut key = [0u8; 8];
+        key[0] = 32;
+        key[4..8].copy_from_slice(&ip.octets());
+        let now = std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_nanos() as u64;
+        map.update(&key, &(now + 300_000_000_000).to_le_bytes(), MapFlags::ANY)?;
+        Ok(())
+    }
+
+    pub fn unban_ip(&self, ip: Ipv4Addr) -> Result<()> {
+        let map = self.find_map("blacklist_map")?;
+        let mut key = [0u8; 8];
+        key[0] = 32;
+        key[4..8].copy_from_slice(&ip.octets());
+        map.delete(&key)?;
+        Ok(())
+    }
+
+    pub fn get_stats(&self) -> Result<XdpStats> {
+        let map = self.find_map("stats_map")?;
+        let sum = |idx: u32| -> u64 {
+            let key = idx.to_le_bytes();
+            match map.lookup(&key, MapFlags::ANY) {
+                Ok(Some(v)) => v
+                    .chunks_exact(8)
+                    .map(|c| u64::from_le_bytes(c.try_into().expect("chunk size 8")))
+                    .sum(),
+                _ => 0,
+            }
+        };
+        Ok(XdpStats {
+            total: sum(0),
+            tcp_mc: sum(1),
+            whitelist: sum(2),
+            blacklist: sum(3),
+            syn_throttle: sum(4),
+            passed: sum(5),
+            dropped: sum(6),
+            verified: sum(7),
+        })
+    }
+}
+
+unsafe impl Send for XdpFilter {}
+
+impl Drop for XdpFilter {
+    fn drop(&mut self) {
+        let _ = self.unload();
+    }
+}
+
+fn build_ringbuf(obj: &Object) -> Result<RingBuffer<'static>> {
+    let map = obj
+        .maps()
+        .find(|m| m.name() == "events_map")
+        .context("events_map not found")?;
+    let mut builder = RingBufferBuilder::new();
+    builder.add(&map, |data: &[u8]| {
+        if data.len() >= 16 {
+            let ty = u32::from_ne_bytes(data[0..4].try_into().expect("4 bytes for type"));
+            let ip4 = u32::from_ne_bytes(data[4..8].try_into().expect("4 bytes for ip"));
+            let val = u64::from_ne_bytes(data[8..16].try_into().expect("8 bytes for val"));
+            tracing::debug!(event = ty, src_ip = ip4, data = val, "xdp event");
+        }
+        0
+    })?;
+    Ok(builder.build()?)
+}

crates/rampart-core/src/xdp/metrics.rs

@@ -0,0 +1,50 @@
+use anyhow::{Context, Result};
+use prometheus::{IntGauge, register};
+
+use super::XdpStats;
+
+pub struct XdpMetrics {
+    total: IntGauge,
+    tcp_mc: IntGauge,
+    whitelist: IntGauge,
+    blacklist: IntGauge,
+    syn_throttle: IntGauge,
+    passed: IntGauge,
+    dropped: IntGauge,
+    verified: IntGauge,
+}
+
+impl XdpMetrics {
+    fn new_gauge(name: &str, help: &str) -> Result<IntGauge> {
+        let gauge = IntGauge::new(name, help)?;
+        register(Box::new(gauge.clone())).context(format!("failed to register {name}"))?;
+        Ok(gauge)
+    }
+
+    pub fn register() -> Result<Self> {
+        let m = Self {
+            total: Self::new_gauge("rampart_xdp_total", "Total XDP packets processed")?,
+            tcp_mc: Self::new_gauge("rampart_xdp_tcp_mc", "TCP packets matched to Minecraft profile")?,
+            whitelist: Self::new_gauge("rampart_xdp_whitelist", "Whitelisted packets")?,
+            blacklist: Self::new_gauge("rampart_xdp_blacklist", "Blacklisted packets")?,
+            syn_throttle: Self::new_gauge("rampart_xdp_syn_throttle", "SYN packets rate-limited")?,
+            passed: Self::new_gauge("rampart_xdp_passed", "Packets passed to upper layers")?,
+            dropped: Self::new_gauge("rampart_xdp_dropped", "Packets dropped by filter")?,
+            verified: Self::new_gauge("rampart_xdp_verified", "Packets challenge-verified")?,
+        };
+        tracing::info!("XDP Prometheus metrics registered");
+        Ok(m)
+    }
+
+    pub fn update(&self, stats: &XdpStats) {
+        self.total.set(stats.total as i64);
+        self.tcp_mc.set(stats.tcp_mc as i64);
+        self.whitelist.set(stats.whitelist as i64);
+        self.blacklist.set(stats.blacklist as i64);
+        self.syn_throttle.set(stats.syn_throttle as i64);
+        self.passed.set(stats.passed as i64);
+        self.dropped.set(stats.dropped as i64);
+        self.verified.set(stats.verified as i64);
+        tracing::debug!("XDP metrics updated");
+    }
+}

crates/rampart-core/src/xdp/mod.rs

@@ -1,41 +1,17 @@
-#[cfg(feature = "xdp")]
-pub struct XdpFilter {
-    interface: String,
-}
+mod stats;
+pub use stats::XdpStats;
 
 #[cfg(feature = "xdp")]
-impl XdpFilter {
-    pub fn new(interface: &str) -> Self {
-        Self {
-            interface: interface.to_string(),
-        }
-    }
+mod filter;
+#[cfg(feature = "xdp")]
+pub use filter::XdpFilter;
 
-    pub fn load(&self) -> anyhow::Result<()> {
-        tracing::info!("XDP filter loaded on {}", self.interface);
-        Ok(())
-    }
-
-    pub fn unload(&self) -> anyhow::Result<()> {
-        tracing::info!("XDP filter unloaded from {}", self.interface);
-        Ok(())
-    }
-}
+#[cfg(feature = "xdp")]
+mod metrics;
+#[cfg(feature = "xdp")]
+pub use metrics::XdpMetrics;
 
 #[cfg(not(feature = "xdp"))]
-pub struct XdpFilter;
-
+mod noop;
 #[cfg(not(feature = "xdp"))]
-impl XdpFilter {
-    pub fn new(_interface: &str) -> Self {
-        Self
-    }
-
-    pub fn load(&self) -> anyhow::Result<()> {
-        Ok(())
-    }
-
-    pub fn unload(&self) -> anyhow::Result<()> {
-        Ok(())
-    }
-}
+pub use noop::*;

crates/rampart-core/src/xdp/noop.rs

@@ -0,0 +1,26 @@
+use anyhow::Result;
+use std::net::Ipv4Addr;
+
+pub struct XdpFilter;
+
+impl XdpFilter {
+    pub fn new(_interface: &str) -> Self {
+        Self
+    }
+    pub fn load(&mut self) -> Result<()> {
+        Ok(())
+    }
+    pub fn unload(&mut self) -> Result<()> {
+        Ok(())
+    }
+    pub fn drain_events(&self) {}
+    pub fn ban_ip(&self, _ip: Ipv4Addr) -> Result<()> {
+        Ok(())
+    }
+    pub fn unban_ip(&self, _ip: Ipv4Addr) -> Result<()> {
+        Ok(())
+    }
+    pub fn get_stats(&self) -> Result<super::XdpStats> {
+        Ok(super::XdpStats::default())
+    }
+}

crates/rampart-core/src/xdp/stats.rs

@@ -0,0 +1,11 @@
+#[derive(Debug, Clone, Copy, Default)]
+pub struct XdpStats {
+    pub total: u64,
+    pub tcp_mc: u64,
+    pub whitelist: u64,
+    pub blacklist: u64,
+    pub syn_throttle: u64,
+    pub passed: u64,
+    pub dropped: u64,
+    pub verified: u64,
+}