feat!: universal redesign — drop Minecraft stack, single-crate architecture

- remove Java plugins (velocity/paper), dashboard, all MC-specific code (handshake, death_code, varint, hostname-HMAC); available in history pre-v0.2 - merge crates/* into one package with src/bin/{rampart,rampart-manager,rampart-cli} - ProtocolHandler trait + registry (no implementations yet), universal PoW kept - XDP: universal L3/L4 filter (xdp/core/) + pluggable hook API (xdp/hooks/), fix IPv6 saddr bug; clang build verified - docs: bilingual knowledge base (docs/kb/: attacks x4, defense-levels, practice x3), rewrite README/architecture for universal concept - TODO.md v4.0: <=300-line module limit, competitor benchmark section (ref/) - deploy/CI/docs cleanup: no MC references, new binary names cargo build/clippy(-D warnings)/test green (55 tests)
2026-08-24 01:50:22 +02:00 · 2026-08-24 01:50:22 +02:00 · 15f474486a
commit 15f474486a
parent 0b53ed720b
179 changed files with 5044 additions and 11519 deletions
--- a/src/cli/commands/blacklist.rs
+++ b/src/cli/commands/blacklist.rs
@ -0,0 +1,69 @@
+use serde::Deserialize;
+
+#[derive(Deserialize)]
+struct AddResponse {
+    status: String,
+    target: String,
+}
+
+#[derive(Deserialize)]
+struct BlacklistItem {
+    target: String,
+    reason: String,
+}
+
+#[derive(Deserialize)]
+struct BlacklistResponse {
+    items: Vec<BlacklistItem>,
+    total: usize,
+}
+
+pub async fn add(target: String, reason: Option<String>) -> anyhow::Result<()> {
+    let manager_url = std::env::var("RAMPART_MANAGER").unwrap_or_else(|_| "http://localhost:8080".to_string());
+
+    let body = serde_json::json!({
+        "target": target,
+        "type": "ip",
+        "reason": reason.unwrap_or_else(|| "manual".to_string()),
+    });
+
+    let client = reqwest::Client::new();
+    match client
+        .post(format!("{manager_url}/api/v1/blacklist"))
+        .json(&body)
+        .send()
+        .await
+    {
+        Ok(resp) => {
+            if let Ok(add_resp) = resp.json::<AddResponse>().await {
+                println!("[OK] {}: {}", add_resp.status, add_resp.target);
+            }
+        },
+        Err(e) => println!("[FAIL] {e}"),
+    }
+    Ok(())
+}
+
+pub async fn remove(target: String) -> anyhow::Result<()> {
+    println!("Removing {target} from blacklist...");
+    println!("(not implemented in v0.1)");
+    Ok(())
+}
+
+pub async fn list() -> anyhow::Result<()> {
+    let manager_url = std::env::var("RAMPART_MANAGER").unwrap_or_else(|_| "http://localhost:8080".to_string());
+
+    match reqwest::get(format!("{manager_url}/api/v1/blacklist")).await {
+        Ok(resp) => {
+            if let Ok(list) = resp.json::<BlacklistResponse>().await {
+                println!("Blacklist ({} entries)", list.total);
+                println!("----------------------");
+                for item in &list.items {
+                    println!("  {} ({})", item.target, item.reason);
+                }
+            }
+        },
+        Err(e) => println!("[FAIL] {e}"),
+    }
+    Ok(())
+}
--- a/src/cli/commands/config.rs
+++ b/src/cli/commands/config.rs
@ -0,0 +1,24 @@
+pub async fn run(key: Option<String>, value: Option<String>) -> anyhow::Result<()> {
+    match (key, value) {
+        (Some(k), Some(v)) => {
+            println!("Setting {k} = {v}");
+            Ok(())
+        },
+        (Some(k), None) => {
+            println!("Reading config key: {k}");
+            println!("(not implemented in v0.1)");
+            Ok(())
+        },
+        (None, Some(_)) | (None, None) => {
+            println!("Configuration");
+            println!("=============\n");
+            println!("Use: rampart config <key> [value]");
+            println!();
+            println!("Example keys:");
+            println!("  workers.count");
+            println!("  limits.rate_limit_login_pps");
+            println!("  limits.max_connections_per_ip");
+            Ok(())
+        },
+    }
+}
--- a/src/cli/commands/doctor.rs
+++ b/src/cli/commands/doctor.rs
@ -0,0 +1,45 @@
+pub async fn run() -> anyhow::Result<()> {
+    println!("Rampart Diagnostics");
+    println!("=====================\n");
+
+    let mut all_ok = true;
+
+    let manager_url = std::env::var("RAMPART_MANAGER").unwrap_or_else(|_| "http://localhost:8080".to_string());
+
+    match reqwest::get(format!("{manager_url}/api/v1/health")).await {
+        Ok(resp) if resp.status().is_success() => {
+            println!("[OK] Manager API");
+        },
+        _ => {
+            println!("[FAIL] Manager API");
+            all_ok = false;
+        },
+    }
+
+    match reqwest::get(format!("{manager_url}/api/v1/blacklist")).await {
+        Ok(resp) if resp.status().is_success() => {
+            println!("[OK] Blacklist API");
+        },
+        _ => {
+            println!("[WARN] Blacklist API unavailable");
+        },
+    }
+
+    match reqwest::get(format!("{manager_url}/api/v1/servers")).await {
+        Ok(resp) if resp.status().is_success() => {
+            println!("[OK] Servers API");
+        },
+        _ => {
+            println!("[WARN] Servers API unavailable");
+        },
+    }
+
+    println!();
+    if all_ok {
+        println!("All checks passed.");
+    } else {
+        println!("Some checks failed. Run with --verbose for details.");
+    }
+
+    Ok(())
+}
--- a/src/cli/commands/drain.rs
+++ b/src/cli/commands/drain.rs
@ -0,0 +1,7 @@
+pub async fn run(node: &str) -> anyhow::Result<()> {
+    println!("Draining node: {node}");
+    println!("Waiting for active connections to drain...");
+    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+    println!("Node {node} drained successfully.");
+    Ok(())
+}
--- a/src/cli/commands/emergency.rs
+++ b/src/cli/commands/emergency.rs
@ -0,0 +1,11 @@
+pub async fn enable() -> anyhow::Result<()> {
+    println!("Emergency mode ENABLED");
+    println!("Only whitelisted IPs will be allowed through.");
+    Ok(())
+}
+
+pub async fn disable() -> anyhow::Result<()> {
+    println!("Emergency mode DISABLED");
+    println!("Normal filtering resumed.");
+    Ok(())
+}
--- a/src/cli/commands/mod.rs
+++ b/src/cli/commands/mod.rs
@ -0,0 +1,6 @@
+pub mod blacklist;
+pub mod config;
+pub mod doctor;
+pub mod drain;
+pub mod emergency;
+pub mod status;
--- a/src/cli/commands/status.rs
+++ b/src/cli/commands/status.rs
@ -0,0 +1,23 @@
+pub async fn run() -> anyhow::Result<()> {
+    println!("Rampart Status");
+    println!("================\n");
+
+    let manager_url = std::env::var("RAMPART_MANAGER").unwrap_or_else(|_| "http://localhost:8080".to_string());
+
+    match reqwest::get(format!("{manager_url}/api/v1/health")).await {
+        Ok(resp) => {
+            if let Ok(body) = resp.json::<serde_json::Value>().await {
+                println!(
+                    "Manager:     {} (v{})",
+                    body["status"].as_str().unwrap_or("unknown"),
+                    body["version"].as_str().unwrap_or("?")
+                );
+            }
+        },
+        Err(e) => println!("Manager:     unreachable ({e})"),
+    }
+
+    println!();
+    println!("To check individual components, run: rampart doctor");
+    Ok(())
+}
--- a/src/cli/mod.rs
+++ b/src/cli/mod.rs
@ -0,0 +1,3 @@
+//! CLI-команды для bin/rampart-cli.
+#[allow(clippy::print_stdout, clippy::print_stderr)]
+pub mod commands;
--- a/src/config/mod.rs
+++ b/src/config/mod.rs
@ -0,0 +1,143 @@
+//! Единый конфиг платформы Rampart.
+mod sections;
+
+pub use sections::{
+    BackendConfig, BanConfig, BindConfig, LimitsConfig, LoggingConfig, MetricsConfig, PowConfig, StoreConfig,
+    WorkerConfig, XdpConfig,
+};
+
+use serde::Deserialize;
+use std::fs;
+
+#[derive(Debug, Clone, Default, Deserialize)]
+pub struct Config {
+    #[serde(default)]
+    pub bind: BindConfig,
+    #[serde(default)]
+    pub backend: BackendConfig,
+    #[serde(default)]
+    pub workers: WorkerConfig,
+    #[serde(default)]
+    pub limits: LimitsConfig,
+    #[serde(default)]
+    pub ban: BanConfig,
+    #[serde(default)]
+    pub store: StoreConfig,
+    #[serde(default)]
+    pub xdp: XdpConfig,
+    #[serde(default)]
+    pub logging: LoggingConfig,
+    #[serde(default)]
+    pub metrics: MetricsConfig,
+    #[serde(default)]
+    pub pow: PowConfig,
+    #[serde(default)]
+    pub whitelist: Vec<String>,
+}
+
+impl Config {
+    /// Парсит конфиг из TOML-строки и валидирует семантику.
+    ///
+    /// # Errors
+    /// Возвращает ошибку при невалидном TOML или нарушении инвариантов
+    /// (пустой список upstream-бэкендов, некорректные адреса).
+    pub fn parse_str(contents: &str) -> anyhow::Result<Self> {
+        let config: Config = toml::from_str(contents)?;
+        config.validate()?;
+        Ok(config)
+    }
+
+    /// Читает и парсит конфиг из файла.
+    ///
+    /// # Errors
+    /// Возвращает ошибку при невозможности прочитать файл или невалидном содержимом.
+    pub fn from_file(path: &str) -> anyhow::Result<Self> {
+        let contents = fs::read_to_string(path)?;
+        Self::parse_str(&contents)
+    }
+
+    fn validate(&self) -> anyhow::Result<()> {
+        if self.backend.upstreams.is_empty() {
+            anyhow::bail!("backend.upstreams must not be empty");
+        }
+        for upstream in &self.backend.upstreams {
+            if upstream.parse::<std::net::SocketAddr>().is_err() {
+                anyhow::bail!("invalid upstream address: {upstream}");
+            }
+        }
+        for entry in &self.whitelist {
+            if entry.parse::<std::net::IpAddr>().is_err() {
+                anyhow::bail!("invalid whitelist entry: {entry}");
+            }
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn empty_config_uses_defaults() {
+        let config = Config::parse_str("").expect("empty config should parse");
+        assert_eq!(config.bind.address, "0.0.0.0");
+        assert_eq!(config.bind.port, 25565);
+        assert_eq!(config.backend.upstreams, vec!["127.0.0.1:25566"]);
+        assert!(!config.pow.enabled);
+        assert!(!config.xdp.enabled);
+    }
+
+    #[test]
+    fn full_config_parses() {
+        let config = Config::parse_str(
+            r#"
+[bind]
+address = "10.0.0.1"
+port = 443
+
+[backend]
+upstreams = ["192.168.1.10:8080", "192.168.1.11:8080"]
+
+[pow]
+enabled = true
+difficulty = 6
+
+whitelist = ["203.0.113.7"]
+"#,
+        )
+        .expect("config should parse");
+        assert_eq!(config.bind.port, 443);
+        assert_eq!(config.backend.upstreams.len(), 2);
+        assert!(config.pow.enabled);
+        assert_eq!(config.pow.difficulty, 6);
+    }
+
+    #[test]
+    fn empty_upstreams_rejected() {
+        let result = Config::parse_str(
+            r#"
+[backend]
+upstreams = []
+"#,
+        );
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn invalid_upstream_rejected() {
+        let result = Config::parse_str(
+            r#"
+[backend]
+upstreams = ["not-an-addr"]
+"#,
+        );
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn invalid_whitelist_entry_rejected() {
+        let result = Config::parse_str("whitelist = [\"999.999.1.1\"]");
+        assert!(result.is_err());
+    }
+}
--- a/src/config/sections.rs
+++ b/src/config/sections.rs
@ -0,0 +1,228 @@
+//! Секции единого конфига платформы.
+
+use serde::Deserialize;
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct BindConfig {
+    #[serde(default = "default_bind_address")]
+    pub address: String,
+    #[serde(default = "default_bind_port")]
+    pub port: u16,
+}
+
+impl Default for BindConfig {
+    fn default() -> Self {
+        Self {
+            address: default_bind_address(),
+            port: default_bind_port(),
+        }
+    }
+}
+
+fn default_bind_address() -> String {
+    "0.0.0.0".to_string()
+}
+fn default_bind_port() -> u16 {
+    25565
+}
+
+/// Generic upstream-бэкенды (список addr:port) за edge-нодой.
+#[derive(Debug, Clone, Deserialize)]
+pub struct BackendConfig {
+    #[serde(default = "default_upstreams")]
+    pub upstreams: Vec<String>,
+}
+
+impl Default for BackendConfig {
+    fn default() -> Self {
+        Self {
+            upstreams: default_upstreams(),
+        }
+    }
+}
+
+fn default_upstreams() -> Vec<String> {
+    vec!["127.0.0.1:25566".to_string()]
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct WorkerConfig {
+    #[serde(default = "default_worker_count")]
+    pub count: usize,
+}
+
+impl Default for WorkerConfig {
+    fn default() -> Self {
+        Self { count: 4 }
+    }
+}
+
+fn default_worker_count() -> usize {
+    4
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct LimitsConfig {
+    #[serde(default = "default_handshake_timeout")]
+    pub handshake_timeout_secs: u64,
+    #[serde(default = "default_max_connections_per_ip")]
+    pub max_connections_per_ip: u32,
+    #[serde(default = "default_rate_limit_pps")]
+    pub rate_limit_pps: f64,
+    #[serde(default = "default_rate_limit_burst")]
+    pub rate_limit_burst: f64,
+}
+
+impl Default for LimitsConfig {
+    fn default() -> Self {
+        Self {
+            handshake_timeout_secs: default_handshake_timeout(),
+            max_connections_per_ip: default_max_connections_per_ip(),
+            rate_limit_pps: default_rate_limit_pps(),
+            rate_limit_burst: default_rate_limit_burst(),
+        }
+    }
+}
+
+fn default_handshake_timeout() -> u64 {
+    5
+}
+fn default_max_connections_per_ip() -> u32 {
+    10
+}
+fn default_rate_limit_pps() -> f64 {
+    5.0
+}
+fn default_rate_limit_burst() -> f64 {
+    10.0
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct BanConfig {
+    #[serde(default = "default_ban_duration")]
+    pub ban_duration_secs: u64,
+}
+
+impl Default for BanConfig {
+    fn default() -> Self {
+        Self {
+            ban_duration_secs: default_ban_duration(),
+        }
+    }
+}
+
+fn default_ban_duration() -> u64 {
+    3600
+}
+
+#[derive(Debug, Clone, Deserialize)]
+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>,
+}
+
+impl Default for StoreConfig {
+    fn default() -> Self {
+        Self {
+            redis_url: None,
+            blacklist_cache_ttl_secs: default_blacklist_cache_ttl(),
+            clickhouse_url: None,
+        }
+    }
+}
+
+fn default_blacklist_cache_ttl() -> u64 {
+    300
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct XdpConfig {
+    #[serde(default)]
+    pub enabled: bool,
+    #[serde(default = "default_xdp_interface")]
+    pub interface: String,
+}
+
+impl Default for XdpConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            interface: default_xdp_interface(),
+        }
+    }
+}
+
+fn default_xdp_interface() -> String {
+    "eth0".to_string()
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct LoggingConfig {
+    #[serde(default = "default_log_level")]
+    pub level: String,
+    #[serde(default = "default_log_format")]
+    pub format: String,
+}
+
+impl Default for LoggingConfig {
+    fn default() -> Self {
+        Self {
+            level: default_log_level(),
+            format: default_log_format(),
+        }
+    }
+}
+
+fn default_log_level() -> String {
+    "info".to_string()
+}
+fn default_log_format() -> String {
+    "text".to_string()
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct MetricsConfig {
+    #[serde(default = "default_metrics_enabled")]
+    pub enabled: bool,
+    #[serde(default = "default_metrics_port")]
+    pub port: u16,
+}
+
+impl Default for MetricsConfig {
+    fn default() -> Self {
+        Self {
+            enabled: default_metrics_enabled(),
+            port: default_metrics_port(),
+        }
+    }
+}
+
+fn default_metrics_enabled() -> bool {
+    true
+}
+fn default_metrics_port() -> u16 {
+    9090
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct PowConfig {
+    #[serde(default)]
+    pub enabled: bool,
+    #[serde(default = "default_pow_difficulty")]
+    pub difficulty: u8,
+}
+
+impl Default for PowConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            difficulty: default_pow_difficulty(),
+        }
+    }
+}
+
+fn default_pow_difficulty() -> u8 {
+    4
+}
--- a/src/engine/challenge.rs
+++ b/src/engine/challenge.rs
@ -0,0 +1,260 @@
+//! Универсальный SHA-256 hashcash: генерация challenge, решатель,
+//! верификатор и адаптивная сложность.
+
+use crate::metrics;
+use rand::RngCore;
+use sha2::{Digest, Sha256};
+use std::collections::VecDeque;
+use std::net::IpAddr;
+use std::time::{Duration, Instant};
+use subtle::ConstantTimeEq;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+
+const CHALLENGE_TTL_SECS: u64 = 30;
+const MAX_NONCE_LEN: usize = 64;
+const ALLOWED_HEX_PREFIX: [u8; 4] = *b"0123";
+
+/// Активная PoW-задача, выданная одному соединению.
+pub struct Challenge {
+    token: [u8; 32],
+    created_at: Instant,
+    difficulty: u8,
+    used: bool,
+}
+
+impl Challenge {
+    #[must_use]
+    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,
+        }
+    }
+
+    #[must_use]
+    pub fn is_expired(&self) -> bool {
+        self.created_at.elapsed().as_secs() >= CHALLENGE_TTL_SECS
+    }
+
+    #[must_use]
+    pub fn challenge_string(&self) -> String {
+        hex::encode(self.token)
+    }
+
+    /// Проверяет nonce и помечает challenge использованным при успехе.
+    #[must_use]
+    pub fn verify(&mut self, nonce: &str) -> bool {
+        if self.used || self.is_expired() || nonce.len() > MAX_NONCE_LEN {
+            return false;
+        }
+
+        let input = format!("{}{nonce}", self.challenge_string());
+        let hash = Sha256::digest(input.as_bytes());
+        let hex_hash = hex::encode(hash);
+        let d = self.difficulty as usize;
+        let ok = hex_hash.as_bytes().iter().take(d).all(|c| {
+            let r = c.ct_eq(&ALLOWED_HEX_PREFIX[0])
+                | c.ct_eq(&ALLOWED_HEX_PREFIX[1])
+                | c.ct_eq(&ALLOWED_HEX_PREFIX[2])
+                | c.ct_eq(&ALLOWED_HEX_PREFIX[3]);
+            r.unwrap_u8() == 1
+        });
+        if ok {
+            self.used = true;
+        }
+        ok
+    }
+}
+
+/// Переборный решатель hashcash (для клиентов и тестов).
+#[must_use]
+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_HEX_PREFIX.contains(c))
+        {
+            return Some(nonce_str);
+        }
+    }
+    None
+}
+
+/// Адаптирует сложность PoW к текущему темпу подключений.
+pub struct DifficultyAdjuster {
+    window: VecDeque<Instant>,
+    min: u8,
+    max: u8,
+    current: u8,
+}
+
+impl Default for DifficultyAdjuster {
+    fn default() -> Self {
+        Self::new(4, 10)
+    }
+}
+
+impl DifficultyAdjuster {
+    #[must_use]
+    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);
+        }
+    }
+
+    #[must_use]
+    pub fn current_difficulty(&self) -> u8 {
+        metrics::POW_CURRENT_DIFFICULTY.set(self.current as i64);
+        self.current
+    }
+
+    fn compute_difficulty(&self) -> u8 {
+        match self.window.len() {
+            cps if cps > 500 => self.max.max(self.min),
+            cps if cps > 200 => 8,
+            cps if cps > 50 => 6,
+            _ => self.min,
+        }
+    }
+}
+
+/// Проводит текстовый PoW-gate в потоке: выдаёт challenge и проверяет ответ.
+///
+/// # Errors
+/// Возвращает ошибку только при сбое ввода-вывода; неверное решение — `Ok(false)`.
+pub async fn enforce(stream: &mut TcpStream, peer_ip: IpAddr, 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 line = format!("{}\n", challenge.challenge_string());
+    stream.write_all(line.as_bytes()).await?;
+
+    let mut buf = [0u8; MAX_NONCE_LEN + 1];
+    let n = tokio::time::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(MAX_NONCE_LEN)]).unwrap_or("").trim();
+    if nonce.is_empty() || nonce.len() > MAX_NONCE_LEN {
+        tracing::debug!("pow: invalid nonce from {peer_ip}");
+        return Ok(false);
+    }
+
+    let valid = challenge.verify(nonce);
+    tracing::debug!(
+        "pow: verification {} for {peer_ip}",
+        if valid { "passed" } else { "failed" }
+    );
+    Ok(valid)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn solve_then_verify_roundtrip() {
+        let mut challenge = Challenge::generate(2);
+        let nonce = solve(&challenge.challenge_string(), 2).expect("solver should find a nonce");
+        assert!(challenge.verify(&nonce), "verifier must accept solver output");
+    }
+
+    #[test]
+    fn verify_rejects_wrong_nonce() {
+        let mut challenge = Challenge::generate(4);
+        let challenge_str = challenge.challenge_string();
+
+        // Подбираем nonce, который независимо рассчитанный SHA-256
+        // заведомо отклоняет — тест детерминирован.
+        let mut wrong_nonce = None;
+        for i in 0..10_000u32 {
+            let candidate = format!("bad-{i}");
+            let hex_hash = hex::encode(Sha256::digest(format!("{challenge_str}{candidate}").as_bytes()));
+            if !hex_hash
+                .as_bytes()
+                .iter()
+                .take(4)
+                .all(|c| ALLOWED_HEX_PREFIX.contains(c))
+            {
+                wrong_nonce = Some(candidate);
+                break;
+            }
+        }
+        let wrong_nonce = wrong_nonce.expect("a failing nonce must exist among candidates");
+        assert!(!challenge.verify(&wrong_nonce));
+    }
+
+    #[test]
+    fn verify_rejects_replay() {
+        let mut challenge = Challenge::generate(2);
+        let nonce = solve(&challenge.challenge_string(), 2).expect("solved");
+        assert!(challenge.verify(&nonce));
+        assert!(!challenge.verify(&nonce), "challenge must be single-use");
+    }
+
+    #[test]
+    fn verify_rejects_oversized_nonce() {
+        let mut challenge = Challenge::generate(2);
+        let big_nonce = "0".repeat(MAX_NONCE_LEN + 1);
+        assert!(!challenge.verify(&big_nonce));
+    }
+
+    #[test]
+    fn adjuster_raises_difficulty_under_load() {
+        let mut adjuster = DifficultyAdjuster::new(4, 10);
+        for _ in 0..600 {
+            adjuster.record_connection();
+        }
+        assert_eq!(adjuster.current_difficulty(), 10);
+    }
+
+    #[test]
+    fn adjuster_stays_minimal_when_idle() {
+        let mut adjuster = DifficultyAdjuster::new(4, 10);
+        adjuster.record_connection();
+        assert_eq!(adjuster.current_difficulty(), 4);
+    }
+}
--- a/src/engine/listener.rs
+++ b/src/engine/listener.rs
@ -0,0 +1,73 @@
+//! Multi-worker TCP accept-loop с SO_REUSEPORT и graceful shutdown.
+
+use crate::config::Config;
+use crate::engine::tunnel::Gateway;
+use socket2::{Domain, Socket, Type};
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tokio::sync::watch;
+
+/// Запускает `workers.count` accept-loop'ов на одном адресе.
+///
+/// # Errors
+/// Возвращает ошибку при невозможности забиндить адрес или падении одного
+/// из accept-loop'ов.
+pub async fn run(config: Arc<Config>, gateway: Arc<Gateway>, shutdown: watch::Receiver<bool>) -> anyhow::Result<()> {
+    let _ = &config;
+    let addr: std::net::SocketAddr = format!("{}:{}", config.bind.address, config.bind.port).parse()?;
+    let workers = config.workers.count.max(1);
+
+    let mut handles = Vec::with_capacity(workers);
+    for _ in 0..workers {
+        let listener = build_listener(addr)?;
+        handles.push(tokio::spawn(accept_loop(
+            listener,
+            Arc::clone(&gateway),
+            shutdown.clone(),
+        )));
+    }
+
+    for h in handles {
+        h.await?;
+    }
+    Ok(())
+}
+
+fn build_listener(addr: std::net::SocketAddr) -> anyhow::Result<TcpListener> {
+    let socket = Socket::new(Domain::IPV4, Type::STREAM, None)?;
+    socket.set_reuse_port(true)?;
+    socket.set_reuse_address(true)?;
+    socket.set_nonblocking(true)?;
+    socket.bind(&addr.into())?;
+    socket.listen(65535)?;
+    Ok(TcpListener::from_std(socket.into())?)
+}
+
+async fn accept_loop(listener: TcpListener, gateway: Arc<Gateway>, mut shutdown: watch::Receiver<bool>) {
+    loop {
+        tokio::select! {
+            biased;
+            _ = shutdown.changed() => {
+                if *shutdown.borrow() {
+                    tracing::info!("shutdown signal received, stopping accept loop");
+                    return;
+                }
+            }
+            result = listener.accept() => {
+                let (stream, peer_addr) = match result {
+                    Ok(conn) => conn,
+                    Err(e) => {
+                        tracing::error!("accept error: {e}");
+                        continue;
+                    }
+                };
+                let gateway = Arc::clone(&gateway);
+                tokio::spawn(async move {
+                    if let Err(e) = gateway.handle(stream, peer_addr).await {
+                        tracing::debug!("connection from {peer_addr}: {e}");
+                    }
+                });
+            }
+        }
+    }
+}
--- a/src/engine/mod.rs
+++ b/src/engine/mod.rs
@ -0,0 +1,5 @@
+//! Edge-движок: listener, generic TCP tunnel, PoW gate.
+
+pub mod challenge;
+pub mod listener;
+pub mod tunnel;
--- a/src/engine/tunnel.rs
+++ b/src/engine/tunnel.rs
@ -0,0 +1,192 @@
+//! Generic TCP proxy: фильтры входа → protocol handshake → upstream.
+
+use crate::config::Config;
+use crate::engine::challenge::{DifficultyAdjuster, enforce as enforce_pow};
+use crate::filter::blacklist::Blacklist;
+use crate::filter::rate_limit::RateLimiter;
+use crate::metrics;
+use crate::protocol::ProtocolRegistry;
+use crate::store::clickhouse::{ClickHouseEvent, ClickHouseWriter};
+use crate::traffic::reputation::IpReputation;
+use crate::xdp::XdpFilter;
+use std::collections::HashSet;
+use std::net::IpAddr;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
+use std::time::Duration;
+use tokio::net::TcpStream;
+use tokio::sync::Mutex as TokioMutex;
+
+/// Общее состояние edge-движка, разделяемое между accept-loop'ами.
+pub struct Gateway {
+    pub config: Arc<Config>,
+    pub rate_limiter: Arc<RateLimiter>,
+    pub blacklist: Arc<Blacklist>,
+    pub adjuster: Arc<Mutex<DifficultyAdjuster>>,
+    pub whitelist: Arc<HashSet<IpAddr>>,
+    pub reputation: Arc<IpReputation>,
+    pub xdp: Option<Arc<Mutex<XdpFilter>>>,
+    pub clickhouse: Option<Arc<TokioMutex<ClickHouseWriter>>>,
+    pub allowed_1s: Arc<AtomicU64>,
+    pub registry: Arc<ProtocolRegistry>,
+    upstream_cursor: AtomicUsize,
+}
+
+impl Gateway {
+    #[allow(clippy::too_many_arguments)]
+    pub fn new(
+        config: Arc<Config>,
+        rate_limiter: Arc<RateLimiter>,
+        blacklist: Arc<Blacklist>,
+        adjuster: Arc<Mutex<DifficultyAdjuster>>,
+        whitelist: Arc<HashSet<IpAddr>>,
+        reputation: Arc<IpReputation>,
+        xdp: Option<Arc<Mutex<XdpFilter>>>,
+        clickhouse: Option<Arc<TokioMutex<ClickHouseWriter>>>,
+        allowed_1s: Arc<AtomicU64>,
+        registry: Arc<ProtocolRegistry>,
+    ) -> Self {
+        Self {
+            config,
+            rate_limiter,
+            blacklist,
+            adjuster,
+            whitelist,
+            reputation,
+            xdp,
+            clickhouse,
+            allowed_1s,
+            registry,
+            upstream_cursor: AtomicUsize::new(0),
+        }
+    }
+
+    /// Полный конвейер обработки входящего соединения.
+    ///
+    /// # Errors
+    /// Возвращает ошибку только при сбое ввода-вывода апстрима; блокировки
+    /// клиента ошибкой не считаются.
+    pub async fn handle(&self, mut client: TcpStream, peer_addr: std::net::SocketAddr) -> anyhow::Result<()> {
+        let peer_ip = peer_addr.ip();
+
+        if self.blacklist.is_blocked(peer_ip) {
+            metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();
+            return Ok(());
+        }
+
+        if !self.pass_pow_gate(&mut client, peer_ip).await? {
+            return Ok(());
+        }
+
+        if !self.rate_limiter.check(peer_ip) {
+            self.block_rate_limit(peer_ip).await;
+            return Ok(());
+        }
+
+        let handler = self.registry.primary()?;
+        let upstream = match handler.handshake(&mut client).await {
+            Ok(upstream) => upstream,
+            Err(e) => {
+                tracing::debug!("handshake rejected from {peer_addr}: {e}");
+                metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();
+                return Ok(());
+            },
+        };
+
+        if !self.rate_limiter.check(peer_ip) {
+            self.block_rate_limit(peer_ip).await;
+            return Ok(());
+        }
+
+        metrics::CONNECTIONS_TOTAL.with_label_values(&["allowed"]).inc();
+        self.allowed_1s.fetch_add(1, Ordering::Relaxed);
+        self.reputation.record_good(peer_ip);
+
+        let mut backend = TcpStream::connect(self.next_upstream(&upstream.addr)).await?;
+        tokio::io::copy_bidirectional(&mut client, &mut backend).await?;
+        Ok(())
+    }
+
+    async fn pass_pow_gate(&self, client: &mut TcpStream, peer_ip: IpAddr) -> anyhow::Result<bool> {
+        let pow_config = &self.config.pow;
+        if !(pow_config.enabled && pow_config.difficulty > 0) {
+            return Ok(true);
+        }
+        if self.whitelist.contains(&peer_ip) {
+            metrics::POW_CHALLENGES_TOTAL.with_label_values(&["skipped"]).inc();
+            metrics::POW_CURRENT_DIFFICULTY.set(pow_config.difficulty as i64);
+            return Ok(true);
+        }
+
+        let diff = {
+            let mut adjuster = self.adjuster.lock().expect("adjuster lock poisoned");
+            adjuster.record_connection();
+            adjuster.current_difficulty()
+        };
+        let result = enforce_pow(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(false);
+        }
+        metrics::POW_CHALLENGES_TOTAL.with_label_values(&["passed"]).inc();
+        metrics::POW_CURRENT_DIFFICULTY.set(diff as i64);
+        Ok(true)
+    }
+
+    fn next_upstream<'a>(&'a self, negotiated: &'a str) -> &'a str {
+        let upstreams = &self.config.backend.upstreams;
+        if upstreams.len() > 1 {
+            let idx = self.upstream_cursor.fetch_add(1, Ordering::Relaxed) % upstreams.len();
+            return upstreams[idx].as_str();
+        }
+        if let Some(first) = upstreams.first() {
+            return first.as_str();
+        }
+        negotiated
+    }
+
+    async fn block_rate_limit(&self, ip: IpAddr) {
+        metrics::RATE_LIMIT_HITS.with_label_values(&["hit"]).inc();
+        metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();
+        self.reputation.record_bad(ip);
+        if self.reputation.score(ip) < -40 {
+            let duration_secs = Duration::from_secs(self.config.ban.ban_duration_secs);
+            self.blacklist.add(ip, duration_secs, "low_reputation");
+            self.xdp_ban(ip, duration_secs.as_secs());
+            self.push_event("block", ip, "low_reputation").await;
+            tracing::info!("low reputation ban {ip}: rate-limit abuse");
+        }
+    }
+
+    fn xdp_ban(&self, ip: IpAddr, duration_secs: u64) {
+        let Some(xdp) = &self.xdp else {
+            return;
+        };
+        let IpAddr::V4(ip_v4) = ip else {
+            return;
+        };
+        match xdp.lock().expect("xdp lock poisoned").ban_ip(ip_v4, duration_secs) {
+            Ok(()) => tracing::debug!("xdp ban {ip_v4} for {duration_secs}s"),
+            Err(e) => tracing::warn!("xdp ban failed for {ip_v4}: {e}"),
+        }
+    }
+
+    async fn push_event(&self, event_type: &str, ip: IpAddr, data_string: &str) {
+        let Some(writer) = &self.clickhouse else {
+            return;
+        };
+        let event = ClickHouseEvent {
+            timestamp: chrono::Utc::now(),
+            event_type: event_type.to_string(),
+            ip: ip.to_string(),
+            data_float: 0.0,
+            data_int: 0,
+            data_string: data_string.to_string(),
+        };
+        if let Err(e) = writer.lock().await.push(event).await {
+            tracing::debug!("clickhouse push error: {e}");
+        }
+    }
+}
--- a/src/filter/blacklist.rs
+++ b/src/filter/blacklist.rs
@ -0,0 +1,104 @@
+use dashmap::DashMap;
+use std::net::IpAddr;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+struct BanEntry {
+    expires: Instant,
+    _reason: String,
+}
+
+pub struct Blacklist {
+    entries: Arc<DashMap<IpAddr, BanEntry>>,
+}
+
+impl Default for Blacklist {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Blacklist {
+    pub fn new() -> Self {
+        Self {
+            entries: Arc::new(DashMap::new()),
+        }
+    }
+
+    pub fn is_blocked(&self, ip: IpAddr) -> bool {
+        if let Some(entry) = self.entries.get(&ip) {
+            if entry.expires > Instant::now() {
+                return true;
+            }
+            drop(entry);
+            self.entries.remove(&ip);
+        }
+        false
+    }
+
+    pub fn add(&self, ip: IpAddr, duration: Duration, reason: &str) {
+        self.entries.insert(
+            ip,
+            BanEntry {
+                expires: Instant::now() + duration,
+                _reason: reason.to_string(),
+            },
+        );
+    }
+
+    pub fn remove(&self, ip: IpAddr) {
+        self.entries.remove(&ip);
+    }
+
+    pub fn len(&self) -> usize {
+        self.entries.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.entries.is_empty()
+    }
+
+    pub fn clear_expired(&self) {
+        self.entries.retain(|_, entry| entry.expires > Instant::now());
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::{IpAddr, Ipv4Addr};
+
+    fn ip(octets: [u8; 4]) -> IpAddr {
+        IpAddr::V4(Ipv4Addr::from(octets))
+    }
+
+    #[test]
+    fn test_blacklist_block() {
+        let bl = Blacklist::new();
+        bl.add(ip([1, 2, 3, 4]), Duration::from_secs(60), "test");
+        assert!(bl.is_blocked(ip([1, 2, 3, 4])));
+    }
+
+    #[test]
+    fn test_blacklist_not_blocked() {
+        let bl = Blacklist::new();
+        bl.add(ip([1, 2, 3, 4]), Duration::from_secs(60), "test");
+        assert!(!bl.is_blocked(ip([5, 6, 7, 8])));
+    }
+
+    #[test]
+    fn test_blacklist_expired() {
+        let bl = Blacklist::new();
+        bl.add(ip([1, 2, 3, 4]), Duration::from_millis(1), "test");
+        std::thread::sleep(Duration::from_millis(2));
+        assert!(!bl.is_blocked(ip([1, 2, 3, 4])));
+    }
+
+    #[test]
+    fn test_blacklist_remove() {
+        let bl = Blacklist::new();
+        bl.add(ip([1, 2, 3, 4]), Duration::from_secs(60), "test");
+        bl.remove(ip([1, 2, 3, 4]));
+        assert!(!bl.is_blocked(ip([1, 2, 3, 4])));
+    }
+}
--- a/src/filter/geo.rs
+++ b/src/filter/geo.rs
@ -0,0 +1,35 @@
+#[cfg(feature = "geoip")]
+pub struct GeoIp {
+    #[allow(dead_code)]
+    reader: maxminddb::Reader<Vec<u8>>,
+}
+
+#[cfg(feature = "geoip")]
+impl GeoIp {
+    pub fn new(db_path: &str) -> anyhow::Result<Self> {
+        let reader = maxminddb::Reader::open_readfile(db_path)?;
+        Ok(Self { reader })
+    }
+}
+
+pub enum IpCategory {
+    Residential,
+    Datacenter,
+    Mobile,
+    Vpn,
+    Tor,
+    Unknown,
+}
+
+impl std::fmt::Display for IpCategory {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Residential => write!(f, "residential"),
+            Self::Datacenter => write!(f, "datacenter"),
+            Self::Mobile => write!(f, "mobile"),
+            Self::Vpn => write!(f, "vpn"),
+            Self::Tor => write!(f, "tor"),
+            Self::Unknown => write!(f, "unknown"),
+        }
+    }
+}
--- a/src/filter/mod.rs
+++ b/src/filter/mod.rs
@ -0,0 +1,5 @@
+//! Фильтры входящего трафика.
+
+pub mod blacklist;
+pub mod geo;
+pub mod rate_limit;
--- a/src/filter/rate_limit.rs
+++ b/src/filter/rate_limit.rs
@ -0,0 +1,152 @@
+use dashmap::DashMap;
+use std::net::IpAddr;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+const MAX_BUCKETS: usize = 1_000_000;
+const EVICTION_IDLE: Duration = Duration::from_secs(600);
+const SWEEP_INTERVAL: Duration = Duration::from_secs(60);
+
+struct Bucket {
+    tokens: f64,
+    last_refill: Instant,
+    last_access: Instant,
+}
+
+pub struct RateLimiter {
+    buckets: Arc<DashMap<IpAddr, Bucket>>,
+    max_tokens: f64,
+    refill_rate: f64,
+    _refill_interval: Duration,
+    epoch: Instant,
+    last_sweep_elapsed: AtomicU64,
+    eviction_idle: Duration,
+}
+
+impl RateLimiter {
+    pub fn new(rate_per_sec: f64, burst: f64) -> Self {
+        Self::with_eviction(rate_per_sec, burst, EVICTION_IDLE)
+    }
+
+    fn with_eviction(rate_per_sec: f64, burst: f64, eviction_idle: Duration) -> Self {
+        Self {
+            buckets: Arc::new(DashMap::new()),
+            max_tokens: burst,
+            refill_rate: rate_per_sec,
+            _refill_interval: Duration::from_secs(1),
+            epoch: Instant::now(),
+            last_sweep_elapsed: AtomicU64::new(0),
+            eviction_idle,
+        }
+    }
+
+    pub fn check(&self, ip: IpAddr) -> bool {
+        let now = Instant::now();
+        let mut entry = self.buckets.entry(ip).or_insert_with(|| Bucket {
+            tokens: self.max_tokens,
+            last_refill: now,
+            last_access: now,
+        });
+
+        let elapsed = now.duration_since(entry.last_refill);
+        let refill = elapsed.as_secs_f64() * self.refill_rate;
+        entry.tokens = (entry.tokens + refill).min(self.max_tokens);
+        entry.last_refill = now;
+        entry.last_access = now;
+
+        if entry.tokens >= 1.0 {
+            entry.tokens -= 1.0;
+            true
+        } else {
+            false
+        }
+    }
+
+    pub fn len(&self) -> usize {
+        self.buckets.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.buckets.is_empty()
+    }
+
+    /// Эвиктит простаивающие бакеты. Запускается, когда бакетов больше
+    /// MAX_BUCKETS либо по расписанию (раз в SWEEP_INTERVAL).
+    pub fn sweep(&self) {
+        let now = Instant::now();
+        let elapsed_secs = now.duration_since(self.epoch).as_secs();
+        let last = self.last_sweep_elapsed.load(Ordering::Relaxed);
+        let due = last == 0 || elapsed_secs.saturating_sub(last) >= SWEEP_INTERVAL.as_secs();
+        let over_cap = self.buckets.len() > MAX_BUCKETS;
+        if !over_cap && !due {
+            return;
+        }
+        self.buckets
+            .retain(|_, b| now.duration_since(b.last_access) < self.eviction_idle);
+        self.last_sweep_elapsed.store(elapsed_secs, Ordering::Relaxed);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::{IpAddr, Ipv4Addr};
+
+    fn test_ip(octet: u8) -> IpAddr {
+        IpAddr::V4(Ipv4Addr::new(10, 0, 0, octet))
+    }
+
+    #[test]
+    fn test_rate_limit_under() {
+        let limiter = RateLimiter::new(10.0, 10.0);
+        assert!(limiter.check(test_ip(1)));
+    }
+
+    #[test]
+    fn test_rate_limit_over() {
+        let limiter = RateLimiter::new(1.0, 1.0);
+        assert!(limiter.check(test_ip(1)));
+        assert!(!limiter.check(test_ip(1)));
+    }
+
+    #[test]
+    fn test_rate_limit_burst() {
+        let limiter = RateLimiter::new(1.0, 5.0);
+        for _ in 0..5 {
+            assert!(limiter.check(test_ip(2)));
+        }
+        assert!(!limiter.check(test_ip(2)));
+    }
+
+    #[test]
+    fn test_rate_limit_refill() {
+        let limiter = RateLimiter::new(100.0, 1.0);
+        assert!(limiter.check(test_ip(3)));
+        assert!(!limiter.check(test_ip(3)));
+        std::thread::sleep(Duration::from_millis(20));
+        assert!(limiter.check(test_ip(3)));
+    }
+
+    #[test]
+    fn test_sweep_removes_idle_keeps_active() {
+        let limiter = RateLimiter::with_eviction(1.0, 10.0, Duration::from_millis(20));
+        limiter.check(test_ip(1));
+        limiter.check(test_ip(2));
+        std::thread::sleep(Duration::from_millis(50));
+        limiter.check(test_ip(2));
+        limiter.sweep();
+        assert_eq!(limiter.len(), 1);
+        assert!(!limiter.buckets.contains_key(&test_ip(1)));
+        assert!(limiter.buckets.contains_key(&test_ip(2)));
+    }
+
+    #[test]
+    fn test_sweep_does_not_remove_active() {
+        let limiter = RateLimiter::with_eviction(1.0, 10.0, Duration::from_millis(50));
+        limiter.check(test_ip(1));
+        limiter.sweep();
+        assert_eq!(limiter.len(), 1);
+        assert!(limiter.buckets.contains_key(&test_ip(1)));
+    }
+}
--- a/src/lib.rs
+++ b/src/lib.rs
@ -0,0 +1,11 @@
+//! Rampart — универсальная платформа сетевой защиты (L3/L4/L7).
+pub mod cli;
+pub mod config;
+pub mod engine;
+pub mod filter;
+pub mod manager;
+pub mod metrics;
+pub mod protocol;
+pub mod store;
+pub mod traffic;
+pub mod xdp;
--- a/src/manager/api/auth.rs
+++ b/src/manager/api/auth.rs
@ -0,0 +1,104 @@
+use crate::manager::AppState;
+use axum::{
+    Json,
+    extract::{ConnectInfo, State},
+    http::StatusCode,
+};
+use dashmap::DashMap;
+use serde::Deserialize;
+use std::{
+    net::{IpAddr, SocketAddr},
+    sync::Arc,
+    time::{Duration, Instant},
+};
+use subtle::ConstantTimeEq;
+
+const LOGIN_WINDOW: Duration = Duration::from_secs(60);
+const LOGIN_MAX_ATTEMPTS: u32 = 5;
+
+#[derive(Deserialize)]
+pub struct LoginRequest {
+    pub password: String,
+}
+
+pub async fn login(
+    State(state): State<Arc<AppState>>,
+    ConnectInfo(addr): ConnectInfo<SocketAddr>,
+    Json(req): Json<LoginRequest>,
+) -> Result<Json<serde_json::Value>, (StatusCode, Json<serde_json::Value>)> {
+    if !allow_login_attempt(&state.login_limiter, addr.ip()) {
+        return Err((
+            StatusCode::TOO_MANY_REQUESTS,
+            Json(serde_json::json!({"error": "too many requests"})),
+        ));
+    }
+
+    if !verify_password(&req.password, &state.api_password) {
+        return Err((
+            StatusCode::UNAUTHORIZED,
+            Json(serde_json::json!({"error": "invalid password"})),
+        ));
+    }
+
+    match crate::manager::auth::create_token(&state.jwt_secret, state.jwt_expiration, &state.jwt_audience) {
+        Ok(token) => Ok(Json(serde_json::json!({"token": token}))),
+        Err(_) => Err((
+            StatusCode::INTERNAL_SERVER_ERROR,
+            Json(serde_json::json!({"error": "token creation failed"})),
+        )),
+    }
+}
+
+fn verify_password(provided: &str, expected: &str) -> bool {
+    let provided = provided.as_bytes();
+    let expected = expected.as_bytes();
+    let len_match = (provided.len() as u64).ct_eq(&(expected.len() as u64));
+    let min_len = provided.len().min(expected.len());
+    let bytes_match = provided[..min_len].ct_eq(&expected[..min_len]);
+    bool::from(len_match & bytes_match)
+}
+
+fn allow_login_attempt(limiter: &DashMap<IpAddr, (Instant, u32)>, ip: IpAddr) -> bool {
+    let now = Instant::now();
+    let mut slot = limiter.entry(ip).or_insert((now, 0));
+    let (last_reset, attempts) = &mut *slot;
+    if now.duration_since(*last_reset) >= LOGIN_WINDOW {
+        *last_reset = now;
+        *attempts = 1;
+    } else if *attempts >= LOGIN_MAX_ATTEMPTS {
+        return false;
+    } else {
+        *attempts += 1;
+    }
+    true
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn verify_password_matches() {
+        assert!(verify_password("s3cret-pass", "s3cret-pass"));
+    }
+
+    #[test]
+    fn verify_password_rejects_wrong() {
+        assert!(!verify_password("wrong-pass", "s3cret-pass"));
+    }
+
+    #[test]
+    fn verify_password_rejects_different_length() {
+        assert!(!verify_password("short", "a-longer-password"));
+    }
+
+    #[test]
+    fn allow_login_attempt_respects_limit() {
+        let limiter = DashMap::new();
+        let ip = IpAddr::from([127, 0, 0, 1]);
+        for _ in 0..LOGIN_MAX_ATTEMPTS {
+            assert!(allow_login_attempt(&limiter, ip));
+        }
+        assert!(!allow_login_attempt(&limiter, ip));
+    }
+}
--- a/src/manager/api/blacklist.rs
+++ b/src/manager/api/blacklist.rs
@ -0,0 +1,94 @@
+use crate::manager::AppState;
+use axum::{Json, extract::State};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BlacklistEntry {
+    pub target: String,
+    #[serde(rename = "type")]
+    pub entry_type: String,
+    pub reason: String,
+    pub created_at: String,
+    pub expires_at: Option<String>,
+}
+
+#[derive(Serialize)]
+pub struct BlacklistResponse {
+    pub items: Vec<BlacklistEntry>,
+    pub total: usize,
+}
+
+#[derive(Deserialize)]
+#[allow(dead_code)]
+pub struct AddBlacklistRequest {
+    pub target: String,
+    #[serde(rename = "type")]
+    pub entry_type: String,
+    pub reason: String,
+    pub duration_secs: Option<u64>,
+}
+
+pub async fn list_blacklist(State(state): State<Arc<AppState>>) -> Json<BlacklistResponse> {
+    let mut conn = match state.redis_client.get_multiplexed_async_connection().await {
+        Ok(c) => c,
+        Err(_) => {
+            return Json(BlacklistResponse {
+                items: vec![],
+                total: 0,
+            });
+        },
+    };
+
+    let members: Vec<String> = match redis::cmd("SMEMBERS")
+        .arg("rampart:blacklist")
+        .query_async(&mut conn)
+        .await
+    {
+        Ok(m) => m,
+        Err(_) => {
+            return Json(BlacklistResponse {
+                items: vec![],
+                total: 0,
+            });
+        },
+    };
+
+    let items: Vec<BlacklistEntry> = members
+        .into_iter()
+        .map(|target| BlacklistEntry {
+            target,
+            entry_type: "ip".to_string(),
+            reason: "manual".to_string(),
+            created_at: chrono::Utc::now().to_rfc3339(),
+            expires_at: None,
+        })
+        .collect();
+
+    let total = items.len();
+    Json(BlacklistResponse { items, total })
+}
+
+pub async fn add_blacklist(
+    State(state): State<Arc<AppState>>,
+    Json(req): Json<AddBlacklistRequest>,
+) -> Json<serde_json::Value> {
+    let mut conn = match state.redis_client.get_multiplexed_async_connection().await {
+        Ok(c) => c,
+        Err(_) => return Json(serde_json::json!({"error": "redis unavailable"})),
+    };
+
+    let _: () = redis::cmd("SADD")
+        .arg("rampart:blacklist")
+        .arg(&req.target)
+        .query_async(&mut conn)
+        .await
+        .unwrap_or_default();
+    tracing::info!("Added to blacklist: {} ({})", req.target, req.reason);
+
+    Json(serde_json::json!({
+        "status": "added",
+        "target": req.target,
+        "reason": req.reason
+    }))
+}
--- a/src/manager/api/health.rs
+++ b/src/manager/api/health.rs
@ -0,0 +1,18 @@
+use crate::manager::AppState;
+use axum::Json;
+use axum::extract::State;
+use serde::Serialize;
+use std::sync::Arc;
+
+#[derive(Serialize)]
+pub struct HealthResponse {
+    pub status: String,
+    pub version: String,
+}
+
+pub async fn health_check(State(_state): State<Arc<AppState>>) -> Json<HealthResponse> {
+    Json(HealthResponse {
+        status: "healthy".to_string(),
+        version: env!("CARGO_PKG_VERSION").to_string(),
+    })
+}
--- a/src/manager/api/mod.rs
+++ b/src/manager/api/mod.rs
@ -0,0 +1,5 @@
+pub mod auth;
+pub mod blacklist;
+pub mod health;
+pub mod nodes;
+pub mod servers;
--- a/src/manager/api/nodes.rs
+++ b/src/manager/api/nodes.rs
@ -0,0 +1,43 @@
+use crate::manager::AppState;
+use axum::{Json, extract::State};
+use redis::AsyncCommands;
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct NodeInfo {
+    pub id: String,
+    pub role: String,
+    pub ip: String,
+    pub status: String,
+    pub last_heartbeat: String,
+}
+
+#[derive(Serialize)]
+pub struct NodesResponse {
+    pub nodes: Vec<NodeInfo>,
+}
+
+pub async fn list_nodes(State(state): State<Arc<AppState>>) -> Json<NodesResponse> {
+    let mut conn = match state.redis_client.get_multiplexed_async_connection().await {
+        Ok(c) => c,
+        Err(_) => return Json(NodesResponse { nodes: vec![] }),
+    };
+
+    let keys: Vec<String> = match redis::cmd("KEYS").arg("rampart:nodes:*").query_async(&mut conn).await {
+        Ok(k) => k,
+        Err(_) => return Json(NodesResponse { nodes: vec![] }),
+    };
+
+    let mut nodes = Vec::with_capacity(keys.len());
+    for key in &keys {
+        let raw: Option<String> = conn.get(key).await.unwrap_or(None);
+        if let Some(json) = raw
+            && let Ok(node) = serde_json::from_str::<NodeInfo>(&json)
+        {
+            nodes.push(node);
+        }
+    }
+
+    Json(NodesResponse { nodes })
+}
--- a/src/manager/api/servers.rs
+++ b/src/manager/api/servers.rs
@ -0,0 +1,44 @@
+use crate::manager::AppState;
+use axum::{Json, extract::State};
+use redis::AsyncCommands;
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ServerEntry {
+    pub name: String,
+    #[serde(rename = "type")]
+    pub server_type: String,
+    pub ip: String,
+    pub port: u16,
+    pub status: String,
+}
+
+#[derive(Serialize)]
+pub struct ServersResponse {
+    pub servers: Vec<ServerEntry>,
+}
+
+pub async fn list_servers(State(state): State<Arc<AppState>>) -> Json<ServersResponse> {
+    let mut conn = match state.redis_client.get_multiplexed_async_connection().await {
+        Ok(c) => c,
+        Err(_) => return Json(ServersResponse { servers: vec![] }),
+    };
+
+    let keys: Vec<String> = match redis::cmd("KEYS").arg("rampart:servers:*").query_async(&mut conn).await {
+        Ok(k) => k,
+        Err(_) => return Json(ServersResponse { servers: vec![] }),
+    };
+
+    let mut servers = Vec::with_capacity(keys.len());
+    for key in &keys {
+        let raw: Option<String> = conn.get(key).await.unwrap_or(None);
+        if let Some(json) = raw
+            && let Ok(server) = serde_json::from_str::<ServerEntry>(&json)
+        {
+            servers.push(server);
+        }
+    }
+
+    Json(ServersResponse { servers })
+}
--- a/src/manager/auth.rs
+++ b/src/manager/auth.rs
@ -0,0 +1,111 @@
+use axum::Json;
+use axum::body::Body;
+use axum::http::{Request, StatusCode};
+use axum::middleware::Next;
+use axum::response::Response;
+use jsonwebtoken::{DecodingKey, EncodingKey, Header, Validation, decode, encode};
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct Claims {
+    pub sub: String,
+    pub aud: String,
+    pub role: String,
+    pub exp: usize,
+    pub iat: usize,
+}
+
+pub fn create_token(secret: &str, expiration: u64, audience: &str) -> Result<String, jsonwebtoken::errors::Error> {
+    let now = chrono::Utc::now().timestamp() as usize;
+    let claims = Claims {
+        sub: "rampart-admin".to_string(),
+        aud: audience.to_string(),
+        role: "admin".to_string(),
+        exp: now + expiration as usize,
+        iat: now,
+    };
+    encode(&Header::default(), &claims, &EncodingKey::from_secret(secret.as_ref()))
+}
+
+pub fn verify_token(token: &str, secret: &str, audience: &str) -> Result<Claims, jsonwebtoken::errors::Error> {
+    let mut validation = Validation::default();
+    validation.set_audience(&[audience]);
+    let token_data = decode::<Claims>(token, &DecodingKey::from_secret(secret.as_ref()), &validation)?;
+    Ok(token_data.claims)
+}
+
+pub async fn auth_middleware(
+    request: Request<Body>,
+    next: Next,
+) -> Result<Response, (StatusCode, Json<serde_json::Value>)> {
+    let auth_header = request
+        .headers()
+        .get(axum::http::header::AUTHORIZATION)
+        .and_then(|value| value.to_str().ok())
+        .and_then(|value| value.strip_prefix("Bearer "));
+
+    let token = match auth_header {
+        Some(t) => t,
+        None => {
+            return Err((
+                StatusCode::UNAUTHORIZED,
+                Json(serde_json::json!({"error": "unauthorized"})),
+            ));
+        },
+    };
+
+    let state = match request.extensions().get::<Arc<crate::manager::AppState>>() {
+        Some(s) => s,
+        None => {
+            return Err((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                Json(serde_json::json!({"error": "internal error"})),
+            ));
+        },
+    };
+
+    if verify_token(token, &state.jwt_secret, &state.jwt_audience).is_err() {
+        return Err((
+            StatusCode::UNAUTHORIZED,
+            Json(serde_json::json!({"error": "unauthorized"})),
+        ));
+    }
+
+    Ok(next.run(request).await)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn valid_secret() -> String {
+        "this-is-a-test-secret-32-bytes-long!".to_string()
+    }
+
+    #[test]
+    fn create_verify_roundtrip_passes() {
+        let secret = valid_secret();
+        let token = create_token(&secret, 3600, "rampart").expect("token creation should succeed");
+        let claims = verify_token(&token, &secret, "rampart").expect("verification should succeed");
+        assert_eq!(claims.sub, "rampart-admin");
+        assert_eq!(claims.aud, "rampart");
+        assert_eq!(claims.role, "admin");
+        assert!(claims.exp > claims.iat);
+    }
+
+    #[test]
+    fn verify_rejects_wrong_audience() {
+        let secret = valid_secret();
+        let token = create_token(&secret, 3600, "rampart").expect("token creation should succeed");
+        assert!(verify_token(&token, &secret, "other").is_err());
+    }
+
+    #[test]
+    fn verify_rejects_wrong_secret() {
+        let secret = valid_secret();
+        let other_secret = "another-test-secret-also-32-bytes-long!".to_string();
+        let token = create_token(&secret, 3600, "rampart").expect("token creation should succeed");
+        assert!(verify_token(&token, &other_secret, "rampart").is_err());
+    }
+}
--- a/src/manager/mod.rs
+++ b/src/manager/mod.rs
@ -0,0 +1,21 @@
+//! Management-плоскость Rampart: REST API, аутентификация, синхронизация узлов.
+pub mod api;
+pub mod auth;
+pub mod sync;
+
+use dashmap::DashMap;
+use std::net::IpAddr;
+use std::sync::Arc;
+use std::time::Instant;
+
+/// Общее состояние management-API.
+pub struct AppState {
+    pub redis_client: redis::Client,
+    pub jwt_secret: String,
+    pub jwt_audience: String,
+    pub jwt_expiration: u64,
+    pub api_password: String,
+    pub login_limiter: DashMap<IpAddr, (Instant, u32)>,
+}
+
+pub type SharedState = Arc<AppState>;
--- a/src/manager/sync/heartbeat.rs
+++ b/src/manager/sync/heartbeat.rs
@ -0,0 +1,44 @@
+use crate::manager::AppState;
+use redis::AsyncCommands;
+use std::sync::Arc;
+use tokio::time::{Duration, interval};
+
+pub async fn start_heartbeat_check(state: Arc<AppState>) {
+    let mut ticker = interval(Duration::from_secs(30));
+    loop {
+        ticker.tick().await;
+        if let Err(e) = check_nodes(&state).await {
+            tracing::warn!("heartbeat check failed: {e}");
+        }
+    }
+}
+
+async fn check_nodes(state: &AppState) -> anyhow::Result<()> {
+    let mut conn = state.redis_client.get_multiplexed_async_connection().await?;
+    let keys: Vec<String> = redis::cmd("KEYS").arg("rampart:nodes:*").query_async(&mut conn).await?;
+
+    let now = chrono::Utc::now().timestamp();
+
+    for key in &keys {
+        let raw: Option<String> = conn.get(key).await?;
+        if let Some(json) = raw
+            && let Ok(mut node) = serde_json::from_str::<serde_json::Value>(&json)
+        {
+            let hb = node["last_heartbeat"]
+                .as_str()
+                .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
+                .map(|t| t.timestamp())
+                .unwrap_or(0);
+            if now - hb > 60 {
+                if let Some(obj) = node.as_object_mut() {
+                    obj.insert("status".to_string(), serde_json::Value::String("offline".to_string()));
+                    if let Ok(updated) = serde_json::to_string(&node) {
+                        let _: () = conn.set(key.as_str(), updated).await.unwrap_or_default();
+                    }
+                }
+                tracing::warn!("Node {key} is offline (heartbeat expired)");
+            }
+        }
+    }
+    Ok(())
+}
--- a/src/manager/sync/mod.rs
+++ b/src/manager/sync/mod.rs
@ -0,0 +1 @@
+pub mod heartbeat;
--- a/src/metrics.rs
+++ b/src/metrics.rs
@ -0,0 +1,69 @@
+use prometheus::{Encoder, IntCounterVec, IntGauge, register_int_counter_vec, register_int_gauge};
+use std::sync::LazyLock;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+
+pub static CONNECTIONS_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    register_int_counter_vec!("rampart_connections_total", "Total connections handled", &["result"])
+        .expect("CONNECTIONS_TOTAL")
+});
+
+pub static RATE_LIMIT_HITS: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    register_int_counter_vec!("rampart_rate_limit_hits", "Rate limit hits", &["action"]).expect("RATE_LIMIT_HITS")
+});
+
+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 static ATTACK_STATUS: LazyLock<IntGauge> = LazyLock::new(|| {
+    register_int_gauge!(
+        "rampart_attack_status",
+        "Attack detector status (0=normal, 1=suspicious, 2=under attack)"
+    )
+    .expect("ATTACK_STATUS")
+});
+
+/// Отдаёт Prometheus-метрики по голому HTTP/0.9-совместимому ответу.
+pub async fn run_metrics_server(addr: &str) {
+    let listener = match TcpListener::bind(addr).await {
+        Ok(l) => l,
+        Err(e) => {
+            tracing::error!("Failed to bind metrics server: {e}");
+            return;
+        },
+    };
+    loop {
+        let (mut stream, _) = match listener.accept().await {
+            Ok(s) => s,
+            Err(e) => {
+                tracing::error!("Metrics accept error: {e}");
+                continue;
+            },
+        };
+        tokio::spawn(async move {
+            let mut buf = [0u8; 1024];
+            if stream.read(&mut buf).await.is_err() {
+                return;
+            }
+            let metric_families = prometheus::gather();
+            let encoder = prometheus::TextEncoder::new();
+            let mut payload = Vec::new();
+            if encoder.encode(&metric_families, &mut payload).is_err() {
+                return;
+            }
+            let header = format!(
+                "HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
+                payload.len()
+            );
+            let mut response = header.into_bytes();
+            response.extend_from_slice(&payload);
+            let _ = stream.write_all(&response).await;
+        });
+    }
+}
--- a/src/protocol/mod.rs
+++ b/src/protocol/mod.rs
@ -0,0 +1,130 @@
+use std::future::Future;
+use std::pin::Pin;
+use tokio::net::TcpStream;
+
+/// Результат успешного protocol-handshake: адрес upstream-бэкенда,
+/// на который нужно проксировать соединение.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Upstream {
+    pub addr: String,
+}
+
+impl Upstream {
+    #[must_use]
+    pub fn new(addr: impl Into<String>) -> Self {
+        Self { addr: addr.into() }
+    }
+}
+
+/// Точка расширения платформы: реализация для конкретного L7-протокола.
+///
+/// Контракт минимальный: обработчик проводит handshake с клиентом
+/// (чтение/запись в поток до момента, когда можно проксировать) и
+/// возвращает целевой [`Upstream`]. Реализации поставляются фичами
+/// (`--features protocol-http`, внешние plugin-крейты) — в ядре их нет.
+///
+/// Метод асинхронен через boxed future, чтобы трейт оставался
+/// dyn-совместимым (`ProtocolRegistry` хранит `Box<dyn ProtocolHandler>`).
+pub trait ProtocolHandler: Send + Sync {
+    /// Имя протокола (для метрик, логов и реестра).
+    fn name(&self) -> &'static str;
+
+    /// Проводит handshake с клиентом и определяет upstream.
+    ///
+    /// # Errors
+    /// Возвращает ошибку при невалидном клиентском потоке или отказе в доступе.
+    fn handshake<'a>(
+        &'a self,
+        stream: &'a mut TcpStream,
+    ) -> Pin<Box<dyn Future<Output = anyhow::Result<Upstream>> + Send + 'a>>;
+}
+
+/// Реестр скомпилированных протокольных обработчиков.
+#[derive(Default)]
+pub struct ProtocolRegistry {
+    handlers: Vec<Box<dyn ProtocolHandler>>,
+}
+
+impl ProtocolRegistry {
+    #[must_use]
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn register(&mut self, handler: Box<dyn ProtocolHandler>) {
+        self.handlers.push(handler);
+    }
+
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.handlers.is_empty()
+    }
+
+    #[must_use]
+    pub fn len(&self) -> usize {
+        self.handlers.len()
+    }
+
+    #[must_use]
+    pub fn names(&self) -> Vec<&'static str> {
+        self.handlers.iter().map(|h| h.name()).collect()
+    }
+
+    #[must_use]
+    pub fn get(&self, name: &str) -> Option<&dyn ProtocolHandler> {
+        self.handlers.iter().find(|h| h.name() == name).map(|h| h.as_ref())
+    }
+
+    /// Возвращает основной обработчик.
+    ///
+    /// # Errors
+    /// Возвращает ошибку, если ни один протокол не скомпилирован —
+    /// edge-нода обязана fail-fast на старте в этом случае.
+    pub fn primary(&self) -> anyhow::Result<&dyn ProtocolHandler> {
+        let Some(handler) = self.handlers.first() else {
+            anyhow::bail!(
+                "no protocol plugins compiled; build with --features protocol-http \
+                 or link an external ProtocolHandler implementation"
+            );
+        };
+        Ok(handler.as_ref())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct DummyHandler;
+
+    impl ProtocolHandler for DummyHandler {
+        fn name(&self) -> &'static str {
+            "dummy"
+        }
+
+        fn handshake<'a>(
+            &'a self,
+            _stream: &'a mut TcpStream,
+        ) -> Pin<Box<dyn Future<Output = anyhow::Result<Upstream>> + Send + 'a>> {
+            Box::pin(async { Ok(Upstream::new("127.0.0.1:9000")) })
+        }
+    }
+
+    #[test]
+    fn empty_registry_fails_fast() {
+        let registry = ProtocolRegistry::new();
+        assert!(registry.is_empty());
+        assert!(registry.primary().is_err());
+    }
+
+    #[test]
+    fn registered_handler_is_resolvable() {
+        let mut registry = ProtocolRegistry::new();
+        registry.register(Box::new(DummyHandler));
+        assert_eq!(registry.len(), 1);
+        assert_eq!(registry.names(), vec!["dummy"]);
+        assert!(registry.primary().is_ok());
+        assert!(registry.get("dummy").is_some());
+        assert!(registry.get("missing").is_none());
+    }
+}
--- a/src/store/clickhouse.rs
+++ b/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;
+                    }
+                }
+            }
+        }
+    });
+}
--- a/src/store/mod.rs
+++ b/src/store/mod.rs
@ -0,0 +1,26 @@
+#[cfg(feature = "store-redis")]
+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 {
+    /// # Errors
+    /// Возвращает ошибку при недоступности хранилища.
+    async fn get(&self, key: &str) -> anyhow::Result<Option<String>>;
+
+    /// # Errors
+    /// Возвращает ошибку при недоступности хранилища.
+    async fn set(&self, key: &str, value: &str) -> anyhow::Result<()>;
+
+    /// # Errors
+    /// Возвращает ошибку при недоступности хранилища.
+    async fn del(&self, key: &str) -> anyhow::Result<()>;
+
+    /// # Errors
+    /// Возвращает ошибку при недоступности хранилища.
+    async fn publish(&self, channel: &str, message: &str) -> anyhow::Result<()>;
+}
--- a/src/store/redis.rs
+++ b/src/store/redis.rs
@ -0,0 +1,128 @@
+use crate::filter::blacklist::Blacklist;
+use crate::store::StateStore;
+use futures::StreamExt;
+use redis::AsyncCommands;
+use redis::Msg;
+use serde::Deserialize;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::watch;
+
+pub struct RedisStore {
+    client: redis::Client,
+}
+
+#[derive(Deserialize)]
+struct BlacklistEvent {
+    ip: String,
+    action: String,
+    #[serde(default = "default_duration")]
+    duration_secs: u64,
+}
+
+fn default_duration() -> u64 {
+    300
+}
+
+impl RedisStore {
+    pub fn new(url: &str) -> anyhow::Result<Self> {
+        let client = redis::Client::open(url)?;
+        Ok(Self { client })
+    }
+}
+
+pub async fn start_blacklist_sync(
+    client: &redis::Client,
+    blacklist: Arc<Blacklist>,
+    mut shutdown: watch::Receiver<bool>,
+) {
+    #[allow(deprecated)]
+    let conn = match client.get_async_connection().await {
+        Ok(c) => c,
+        Err(e) => {
+            tracing::error!("failed to connect to Redis for blacklist sync: {e}");
+            return;
+        },
+    };
+
+    let mut pubsub = conn.into_pubsub();
+    if let Err(e) = pubsub.subscribe("rampart:blacklist:events").await {
+        tracing::error!("failed to subscribe to blacklist events: {e}");
+        return;
+    }
+    tracing::info!("subscribed to rampart:blacklist:events");
+
+    loop {
+        let mut stream = pubsub.on_message();
+        let msg_fut = stream.next();
+        tokio::pin!(msg_fut);
+
+        tokio::select! {
+            _ = shutdown.changed() => {
+                if *shutdown.borrow() {
+                    tracing::info!("shutting down blacklist subscriber");
+                    return;
+                }
+            }
+            result = &mut msg_fut => {
+                match result {
+                    Some(msg) => {
+                        if let Err(e) = handle_event(&msg, &blacklist) {
+                            tracing::error!("blacklist event error: {e}");
+                        }
+                    }
+                    None => {
+                        tracing::error!("pubsub stream ended");
+                        tokio::time::sleep(Duration::from_secs(1)).await;
+                        return;
+                    }
+                }
+            }
+        }
+    }
+}
+
+fn handle_event(msg: &Msg, blacklist: &Blacklist) -> anyhow::Result<()> {
+    let payload: String = msg.get_payload()?;
+    let event: BlacklistEvent = serde_json::from_str(&payload)?;
+
+    let ip: std::net::IpAddr = event.ip.parse()?;
+
+    match event.action.as_str() {
+        "ban" => {
+            blacklist.add(ip, Duration::from_secs(event.duration_secs), "redis");
+            tracing::info!("blacklist add via Redis: {}", event.ip);
+        },
+        "unban" => {
+            blacklist.remove(ip);
+            tracing::info!("blacklist remove via Redis: {}", event.ip);
+        },
+        a => anyhow::bail!("unknown action: {a}"),
+    }
+    Ok(())
+}
+
+impl StateStore for RedisStore {
+    async fn get(&self, key: &str) -> anyhow::Result<Option<String>> {
+        let mut conn = self.client.get_multiplexed_async_connection().await?;
+        Ok(conn.get(key).await?)
+    }
+
+    async fn set(&self, key: &str, value: &str) -> anyhow::Result<()> {
+        let mut conn = self.client.get_multiplexed_async_connection().await?;
+        let _: () = conn.set(key, value).await?;
+        Ok(())
+    }
+
+    async fn del(&self, key: &str) -> anyhow::Result<()> {
+        let mut conn = self.client.get_multiplexed_async_connection().await?;
+        let _: () = conn.del(key).await?;
+        Ok(())
+    }
+
+    async fn publish(&self, channel: &str, message: &str) -> anyhow::Result<()> {
+        let mut conn = self.client.get_multiplexed_async_connection().await?;
+        let _: () = conn.publish(channel, message).await?;
+        Ok(())
+    }
+}
--- a/src/traffic/alert.rs
+++ b/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"));
+    }
+}
--- a/src/traffic/detector.rs
+++ b/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);
+    }
+}
--- a/src/traffic/ewma.rs
+++ b/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);
+    }
+}
--- a/src/traffic/mod.rs
+++ b/src/traffic/mod.rs
@ -0,0 +1,7 @@
+//! Профилирование трафика, детектор атак и репутация IP.
+
+pub mod alert;
+pub mod detector;
+pub mod ewma;
+pub mod profiler;
+pub mod reputation;
--- a/src/traffic/profiler.rs
+++ b/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);
+    }
+}
--- a/src/traffic/reputation.rs
+++ b/src/traffic/reputation.rs
@ -0,0 +1,110 @@
+use dashmap::DashMap;
+use std::net::IpAddr;
+use std::sync::Arc;
+
+pub struct IpReputation {
+    scores: Arc<DashMap<IpAddr, 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: IpAddr) {
+        let mut entry = self.scores.entry(ip).or_insert(0);
+        *entry = (*entry + 1).min(100);
+    }
+
+    pub fn record_bad(&self, ip: IpAddr) {
+        let mut entry = self.scores.entry(ip).or_insert(0);
+        *entry = (*entry - 10).max(-100);
+    }
+
+    pub fn score(&self, ip: IpAddr) -> i32 {
+        self.scores.get(&ip).map(|v| *v).unwrap_or(0)
+    }
+
+    pub fn is_trusted(&self, ip: IpAddr) -> bool {
+        self.score(ip) > 50
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};
+
+    fn v4(octets: [u8; 4]) -> IpAddr {
+        IpAddr::V4(Ipv4Addr::from(octets))
+    }
+
+    #[test]
+    fn test_reputation_initial_score() {
+        let rep = IpReputation::new();
+        assert_eq!(rep.score(v4([192, 168, 1, 1])), 0);
+    }
+
+    #[test]
+    fn test_reputation_good() {
+        let rep = IpReputation::new();
+        let ip = v4([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 = v4([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 = v4([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 = v4([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 = v4([10, 0, 0, 5]);
+        assert!(!rep.is_trusted(ip));
+        for _ in 0..51 {
+            rep.record_good(ip);
+        }
+        assert!(rep.is_trusted(ip));
+    }
+
+    #[test]
+    fn test_reputation_v6() {
+        let rep = IpReputation::new();
+        let ip = IpAddr::V6(Ipv6Addr::LOCALHOST);
+        rep.record_bad(ip);
+        assert_eq!(rep.score(ip), -10);
+    }
+}
--- a/src/xdp/filter.rs
+++ b/src/xdp/filter.rs
@ -0,0 +1,156 @@
+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"), "/universal_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_universal_filter")
+            .context("XDP program 'rampart_universal_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, duration_secs: u64) -> 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 + duration_secs * 1_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: sum(1),
+            whitelist: sum(2),
+            blacklist: sum(3),
+            syn_throttle: sum(4),
+            passed: sum(5),
+            dropped: sum(6),
+            udp: sum(7),
+            rate_limit: sum(8),
+        })
+    }
+}
+
+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()?)
+}
--- a/src/xdp/metrics.rs
+++ b/src/xdp/metrics.rs
@ -0,0 +1,53 @@
+use anyhow::{Context, Result};
+use prometheus::{IntGauge, register};
+
+use super::XdpStats;
+
+pub struct XdpMetrics {
+    total: IntGauge,
+    tcp: IntGauge,
+    whitelist: IntGauge,
+    blacklist: IntGauge,
+    syn_throttle: IntGauge,
+    passed: IntGauge,
+    dropped: IntGauge,
+    udp: IntGauge,
+    rate_limit: 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: Self::new_gauge("rampart_xdp_tcp", "TCP packets processed")?,
+            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")?,
+            udp: Self::new_gauge("rampart_xdp_udp", "UDP packets processed")?,
+            rate_limit: Self::new_gauge("rampart_xdp_rate_limit", "Packets rate-limited by window counter")?,
+        };
+        tracing::info!("XDP Prometheus metrics registered");
+        Ok(m)
+    }
+
+    pub fn update(&self, stats: &XdpStats) {
+        self.total.set(stats.total as i64);
+        self.tcp.set(stats.tcp 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.udp.set(stats.udp as i64);
+        self.rate_limit.set(stats.rate_limit as i64);
+        tracing::debug!("XDP metrics updated");
+    }
+}
--- a/src/xdp/mod.rs
+++ b/src/xdp/mod.rs
@ -0,0 +1,18 @@
+//! Загрузчик XDP-программы (L3/L4-уровень защиты).
+mod stats;
+pub use stats::XdpStats;
+
+#[cfg(feature = "xdp")]
+mod filter;
+#[cfg(feature = "xdp")]
+pub use filter::XdpFilter;
+
+#[cfg(feature = "xdp")]
+mod metrics;
+#[cfg(feature = "xdp")]
+pub use metrics::XdpMetrics;
+
+#[cfg(not(feature = "xdp"))]
+mod noop;
+#[cfg(not(feature = "xdp"))]
+pub use noop::XdpFilter;
--- a/src/xdp/noop.rs
+++ b/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, _duration_secs: u64) -> Result<()> {
+        Ok(())
+    }
+    pub fn unban_ip(&self, _ip: Ipv4Addr) -> Result<()> {
+        Ok(())
+    }
+    pub fn get_stats(&self) -> Result<super::XdpStats> {
+        Ok(super::XdpStats::default())
+    }
+}
--- a/src/xdp/stats.rs
+++ b/src/xdp/stats.rs
@ -0,0 +1,13 @@
+/// Счётчики XDP-программы; индексы соответствуют STAT_* из xdp/core/maps.h.
+#[derive(Debug, Clone, Copy, Default)]
+pub struct XdpStats {
+    pub total: u64,
+    pub tcp: u64,
+    pub whitelist: u64,
+    pub blacklist: u64,
+    pub syn_throttle: u64,
+    pub passed: u64,
+    pub dropped: u64,
+    pub udp: u64,
+    pub rate_limit: u64,
+}