Initial commit: Rampart v0.2.0

Multi-layer DDoS protection for Minecraft servers.

- rampart-core: Edge node with XDP/eBPF + Rust L7 filtering
- rampart-manager: REST API with JWT auth, Redis sync
- rampart-cli: CLI tool for operators
- velocity-plugin: Domain check, HMAC verify, server registry, load balancer
- paper-plugin: Auto-registration, heartbeat, HMAC verify
- dashboard: React + Vite web UI for management

crates/rampart-core/Cargo.toml

@@ -0,0 +1,39 @@
+[package]
+name = "rampart-core"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+tokio.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+thiserror.workspace = true
+anyhow.workspace = true
+dashmap.workspace = true
+crossbeam.workspace = true
+hex.workspace = true
+sha2.workspace = true
+hmac.workspace = true
+subtle.workspace = true
+socket2 = { workspace = true, features = ["all"] }
+prometheus.workspace = true
+toml.workspace = true
+futures.workspace = true
+
+redis = { version = "0.27", optional = true, features = ["tokio-comp"] }
+maxminddb = { version = "0.30", optional = true }
+tokio-splice = { version = "0.2", optional = true }
+libbpf-rs = { version = "0.24", optional = true }
+
+[features]
+default = ["store-redis"]
+store-redis = ["dep:redis"]
+geoip = ["dep:maxminddb"]
+xdp = ["dep:libbpf-rs"]
+io-uring = ["dep:tokio-splice"]

crates/rampart-core/src/config.rs

@@ -0,0 +1,256 @@
+use serde::Deserialize;
+use std::fs;
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct Config {
+    #[serde(default)]
+    pub bind: BindConfig,
+    #[serde(default)]
+    pub backend: BackendConfig,
+    #[serde(default)]
+    pub hmac: HmacConfig,
+    #[serde(default)]
+    pub workers: WorkerConfig,
+    #[serde(default)]
+    pub limits: LimitsConfig,
+    #[serde(default)]
+    pub store: StoreConfig,
+    #[serde(default)]
+    pub xdp: XdpConfig,
+    #[serde(default)]
+    pub death_code: DeathCodeConfig,
+    #[serde(default)]
+    pub logging: LoggingConfig,
+    #[serde(default)]
+    pub metrics: MetricsConfig,
+}
+
+#[derive(Debug, Clone, Default, Deserialize)]
+pub struct BindConfig {
+    #[serde(default = "default_bind_address")]
+    pub address: String,
+    #[serde(default = "default_bind_port")]
+    pub port: u16,
+}
+
+fn default_bind_address() -> String {
+    "0.0.0.0".to_string()
+}
+fn default_bind_port() -> u16 {
+    25565
+}
+
+#[derive(Debug, Clone, Default, Deserialize)]
+pub struct BackendConfig {
+    #[serde(default = "default_backend_address")]
+    pub address: String,
+    #[serde(default = "default_backend_port")]
+    pub port: u16,
+}
+
+fn default_backend_address() -> String {
+    "127.0.0.1".to_string()
+}
+fn default_backend_port() -> u16 {
+    25566
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct HmacConfig {
+    #[serde(default)]
+    pub secret: String,
+    #[serde(default = "default_key_rotation")]
+    pub key_rotation_interval_secs: u64,
+}
+
+fn default_key_rotation() -> u64 {
+    3600
+}
+
+impl Default for HmacConfig {
+    fn default() -> Self {
+        Self {
+            secret: String::new(),
+            key_rotation_interval_secs: 3600,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct WorkerConfig {
+    #[serde(default = "default_worker_count")]
+    pub count: usize,
+}
+
+fn default_worker_count() -> usize {
+    4
+}
+
+impl Default for WorkerConfig {
+    fn default() -> Self {
+        Self { count: 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_login")]
+    pub rate_limit_login_pps: f64,
+    #[serde(default = "default_rate_limit_status")]
+    pub rate_limit_status_pps: f64,
+    #[serde(default = "default_rate_limit_burst")]
+    pub rate_limit_burst: f64,
+}
+
+fn default_handshake_timeout() -> u64 {
+    5
+}
+fn default_max_connections_per_ip() -> u32 {
+    10
+}
+fn default_rate_limit_login() -> f64 {
+    5.0
+}
+fn default_rate_limit_status() -> f64 {
+    2.0
+}
+fn default_rate_limit_burst() -> f64 {
+    10.0
+}
+
+impl Default for LimitsConfig {
+    fn default() -> Self {
+        Self {
+            handshake_timeout_secs: 5,
+            max_connections_per_ip: 10,
+            rate_limit_login_pps: 5.0,
+            rate_limit_status_pps: 2.0,
+            rate_limit_burst: 10.0,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct StoreConfig {
+    pub redis_url: Option<String>,
+    #[serde(default = "default_blacklist_cache_ttl")]
+    pub blacklist_cache_ttl_secs: u64,
+}
+
+fn default_blacklist_cache_ttl() -> u64 {
+    300
+}
+
+impl Default for StoreConfig {
+    fn default() -> Self {
+        Self {
+            redis_url: None,
+            blacklist_cache_ttl_secs: 300,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct XdpConfig {
+    #[serde(default)]
+    pub enabled: bool,
+    #[serde(default = "default_xdp_interface")]
+    pub interface: String,
+}
+
+fn default_xdp_interface() -> String {
+    "eth0".to_string()
+}
+
+impl Default for XdpConfig {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            interface: "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,
+}
+
+fn default_log_level() -> String {
+    "info".to_string()
+}
+fn default_log_format() -> String {
+    "text".to_string()
+}
+
+impl Default for LoggingConfig {
+    fn default() -> Self {
+        Self {
+            level: "info".to_string(),
+            format: "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,
+}
+
+fn default_metrics_enabled() -> bool {
+    true
+}
+fn default_metrics_port() -> u16 {
+    9090
+}
+
+impl Default for MetricsConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            port: 9090,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Deserialize)]
+pub struct DeathCodeConfig {
+    #[serde(default = "default_death_code_enabled")]
+    pub enabled: bool,
+    #[serde(default = "default_death_code_ban_duration")]
+    pub ban_duration_secs: u64,
+}
+
+fn default_death_code_enabled() -> bool {
+    true
+}
+fn default_death_code_ban_duration() -> u64 {
+    3600
+}
+
+impl Default for DeathCodeConfig {
+    fn default() -> Self {
+        Self {
+            enabled: true,
+            ban_duration_secs: 3600,
+        }
+    }
+}
+
+impl Config {
+    pub fn from_file(path: &str) -> anyhow::Result<Self> {
+        let contents = fs::read_to_string(path)?;
+        let config: Config = toml::from_str(&contents)?;
+        Ok(config)
+    }
+}

crates/rampart-core/src/crypto/hmac.rs

@@ -0,0 +1,73 @@
+use hmac::{Hmac, Mac};
+use sha2::Sha256;
+use subtle::ConstantTimeEq;
+
+type HmacSha256 = Hmac<Sha256>;
+
+pub fn sign(hostname: &str, secret: &[u8]) -> String {
+    let mut mac = HmacSha256::new_from_slice(secret).expect("HMAC accepts any key length");
+    mac.update(hostname.as_bytes());
+    hex::encode(mac.finalize().into_bytes())
+}
+
+pub fn verify(hostname: &str, provided_sig: &str, secret: &[u8]) -> bool {
+    let expected = sign(hostname, secret);
+    expected.as_bytes().ct_eq(provided_sig.as_bytes()).into()
+}
+
+pub fn sign_hostname(raw: &str, secret: &[u8]) -> String {
+    let domain = raw.split('\0').next().unwrap_or(raw);
+    let sig = sign(domain, secret);
+    format!("{raw}\0shield\0{sig}")
+}
+
+pub fn parse_hostname(raw: &str) -> (String, Option<String>) {
+    let parts: Vec<&str> = raw.split('\0').collect();
+    let domain = parts[0].to_string();
+    let hmac = parts
+        .iter()
+        .position(|&p| p == "shield")
+        .and_then(|i| parts.get(i + 1))
+        .map(|s| s.to_string());
+    (domain, hmac)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_sign_verify() {
+        let secret = b"test_secret_32_bytes_long_here!!";
+        let hostname = "play.example.com";
+        let sig = sign(hostname, secret);
+        assert!(verify(hostname, &sig, secret));
+    }
+
+    #[test]
+    fn test_verify_wrong_secret() {
+        let secret = b"test_secret_32_bytes_long_here!!";
+        let wrong = b"wrong_secret_32_bytes_long_here!!!";
+        let hostname = "play.example.com";
+        let sig = sign(hostname, wrong);
+        assert!(!verify(hostname, &sig, secret));
+    }
+
+    #[test]
+    fn test_sign_hostname_suffix() {
+        let secret = b"test_secret";
+        let result = sign_hostname("play.example.com", secret);
+        assert!(result.starts_with("play.example.com\0shield\0"));
+        let sig = result.split("\0shield\0").nth(1).unwrap();
+        assert_eq!(sig.len(), 64);
+    }
+
+    #[test]
+    fn test_verify_constant_time() {
+        let secret = b"test_secret_32_bytes_long_here!!";
+        let hostname = "play.example.com";
+        let sig = sign(hostname, secret);
+        assert!(!verify("play.example.co", &sig, secret));
+        assert!(verify(hostname, &sig, secret));
+    }
+}

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

@@ -0,0 +1 @@
+pub mod hmac;

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

@@ -0,0 +1,98 @@
+use dashmap::DashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+struct BanEntry {
+    expires: Instant,
+    _reason: String,
+}
+
+pub struct Blacklist {
+    entries: Arc<DashMap<u32, 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: u32) -> 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: u32, duration: Duration, reason: &str) {
+        self.entries.insert(
+            ip,
+            BanEntry {
+                expires: Instant::now() + duration,
+                _reason: reason.to_string(),
+            },
+        );
+    }
+
+    pub fn remove(&self, ip: u32) {
+        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::*;
+
+    #[test]
+    fn test_blacklist_block() {
+        let bl = Blacklist::new();
+        bl.add(0x01020304, Duration::from_secs(60), "test");
+        assert!(bl.is_blocked(0x01020304));
+    }
+
+    #[test]
+    fn test_blacklist_not_blocked() {
+        let bl = Blacklist::new();
+        bl.add(0x01020304, Duration::from_secs(60), "test");
+        assert!(!bl.is_blocked(0x05060708));
+    }
+
+    #[test]
+    fn test_blacklist_expired() {
+        let bl = Blacklist::new();
+        bl.add(0x01020304, Duration::from_millis(1), "test");
+        std::thread::sleep(Duration::from_millis(2));
+        assert!(!bl.is_blocked(0x01020304));
+    }
+
+    #[test]
+    fn test_blacklist_remove() {
+        let bl = Blacklist::new();
+        bl.add(0x01020304, Duration::from_secs(60), "test");
+        bl.remove(0x01020304);
+        assert!(!bl.is_blocked(0x01020304));
+    }
+}

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

@@ -0,0 +1,207 @@
+use crate::proxy::handshake::{read_string, read_varint};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum DeathCode {
+    EmptyPacket,
+    PacketTooShort,
+    InvalidPacketId,
+    NegativeProtocolVersion,
+    NonCanonicalVarint,
+    NullByteInHostname,
+    UnprintableHostname,
+    MalformedPacket,
+}
+
+impl DeathCode {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Self::EmptyPacket => "empty_packet",
+            Self::PacketTooShort => "packet_too_short",
+            Self::InvalidPacketId => "invalid_packet_id",
+            Self::NegativeProtocolVersion => "negative_protocol_version",
+            Self::NonCanonicalVarint => "non_canonical_varint",
+            Self::NullByteInHostname => "null_byte_in_hostname",
+            Self::UnprintableHostname => "unprintable_hostname",
+            Self::MalformedPacket => "malformed_packet",
+        }
+    }
+}
+
+pub fn detect(buf: &[u8]) -> Option<DeathCode> {
+    if buf.is_empty() {
+        return Some(DeathCode::EmptyPacket);
+    }
+
+    if buf.len() < 3 {
+        return Some(DeathCode::PacketTooShort);
+    }
+
+    let (packet_len, after_len) = match read_varint(buf, 0) {
+        Ok(r) => r,
+        Err(_) => return Some(DeathCode::MalformedPacket),
+    };
+
+    if !is_canonical_varint(buf, 0) {
+        return Some(DeathCode::NonCanonicalVarint);
+    }
+
+    if packet_len <= 0 || (after_len + packet_len as usize) > buf.len() {
+        return Some(DeathCode::MalformedPacket);
+    }
+
+    let (packet_id, after_id) = match read_varint(buf, after_len) {
+        Ok(r) => r,
+        Err(_) => return Some(DeathCode::MalformedPacket),
+    };
+
+    if !is_canonical_varint(buf, after_len) {
+        return Some(DeathCode::NonCanonicalVarint);
+    }
+
+    if packet_id != 0x00 {
+        return Some(DeathCode::InvalidPacketId);
+    }
+
+    let (_protocol_version, after_pv) = match read_varint(buf, after_id) {
+        Ok(r) => r,
+        Err(_) => return Some(DeathCode::MalformedPacket),
+    };
+
+    if !is_canonical_varint(buf, after_id) {
+        return Some(DeathCode::NonCanonicalVarint);
+    }
+
+    let (server_address, _) = match read_string(buf, after_pv) {
+        Ok(r) => r,
+        Err(_) => return Some(DeathCode::MalformedPacket),
+    };
+
+    if !is_canonical_varint(buf, after_pv) {
+        return Some(DeathCode::NonCanonicalVarint);
+    }
+
+    if server_address.contains('\0') {
+        return Some(DeathCode::NullByteInHostname);
+    }
+
+    if !server_address.chars().all(|c| c.is_ascii_graphic() || c == '.') {
+        return Some(DeathCode::UnprintableHostname);
+    }
+
+    None
+}
+
+fn is_canonical_varint(buf: &[u8], start: usize) -> bool {
+    let mut value: u32 = 0;
+    let mut shift = 0;
+    let mut bytes_used = 0;
+
+    for (i, &byte) in buf[start..].iter().enumerate() {
+        if i >= 5 {
+            return false;
+        }
+        bytes_used = i + 1;
+        value |= ((byte & 0x7F) as u32) << shift;
+        shift += 7;
+        if (byte & 0x80) == 0 {
+            break;
+        }
+    }
+
+    if bytes_used >= 5 {
+        return false;
+    }
+
+    let min_varint = |val: u32| -> usize {
+        if val == 0 {
+            return 1;
+        }
+        let mut bits = 32 - val.leading_zeros();
+        let mut bytes = 0;
+        while bits > 0 {
+            bytes += 1;
+            bits = bits.saturating_sub(7);
+        }
+        bytes.max(1)
+    };
+
+    bytes_used == min_varint(value)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn write_varint(buf: &mut Vec<u8>, mut value: i32) {
+        loop {
+            if (value & !0x7F) == 0 {
+                buf.push(value as u8);
+                return;
+            }
+            buf.push((value as u8 & 0x7F) | 0x80);
+            value >>= 7;
+        }
+    }
+
+    fn build_handshake_raw(hostname: &str) -> Vec<u8> {
+        let addr = hostname.as_bytes();
+        let mut buf = Vec::new();
+        buf.push(0x00);
+        write_varint(&mut buf, 765);
+        write_varint(&mut buf, addr.len() as i32);
+        buf.extend_from_slice(addr);
+        buf.extend_from_slice(&[0x63, 0xDD]);
+        buf.push(0x02);
+
+        let len = buf.len() as i32;
+        let mut pkt = Vec::new();
+        write_varint(&mut pkt, len);
+        pkt.extend_from_slice(&buf);
+        pkt
+    }
+
+    #[test]
+    fn test_valid_handshake() {
+        let pkt = build_handshake_raw("play.example.com");
+        assert_eq!(detect(&pkt), None);
+    }
+
+    #[test]
+    fn test_empty_packet() {
+        assert_eq!(detect(&[]), Some(DeathCode::EmptyPacket));
+    }
+
+    #[test]
+    fn test_null_byte_in_hostname() {
+        let pkt = build_handshake_raw("play.example.com\0extra");
+        assert_eq!(detect(&pkt), Some(DeathCode::NullByteInHostname));
+    }
+
+    #[test]
+    fn test_non_canonical_varint() {
+        let pkt: Vec<u8> = vec![0x08, 0x00, 0x80, 0x00, 0x02, b'e', b'x', 0x63, 0xDD, 0x02];
+        assert_eq!(detect(&pkt), Some(DeathCode::NonCanonicalVarint));
+    }
+
+    #[test]
+    fn test_invalid_packet_id() {
+        let addr = b"play.example.com";
+        let mut buf = Vec::new();
+        write_varint(&mut buf, 1);
+        buf.extend_from_slice(addr);
+        buf.extend_from_slice(&[0x63, 0xDD]);
+        buf.push(0x02);
+
+        let len = buf.len() as i32;
+        let mut pkt = Vec::new();
+        write_varint(&mut pkt, len);
+        pkt.extend_from_slice(&buf);
+        assert_eq!(detect(&pkt), Some(DeathCode::InvalidPacketId));
+    }
+
+    #[test]
+    fn test_unprintable_hostname() {
+        let pkt = build_handshake_raw("play\x01example.com");
+        assert_eq!(detect(&pkt), Some(DeathCode::UnprintableHostname));
+    }
+}

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

@@ -0,0 +1,34 @@
+#[cfg(feature = "geoip")]
+pub struct GeoIp {
+    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"),
+        }
+    }
+}

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

@@ -0,0 +1,4 @@
+pub mod blacklist;
+pub mod death_code;
+pub mod geo;
+pub mod rate_limit;

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

@@ -0,0 +1,90 @@
+use dashmap::DashMap;
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+struct Bucket {
+    tokens: f64,
+    last_refill: Instant,
+}
+
+pub struct RateLimiter {
+    buckets: Arc<DashMap<u32, Bucket>>,
+    max_tokens: f64,
+    refill_rate: f64,
+    _refill_interval: Duration,
+}
+
+impl RateLimiter {
+    pub fn new(rate_per_sec: f64, burst: f64) -> Self {
+        Self {
+            buckets: Arc::new(DashMap::new()),
+            max_tokens: burst,
+            refill_rate: rate_per_sec,
+            _refill_interval: Duration::from_secs(1),
+        }
+    }
+
+    pub fn check(&self, ip: u32) -> bool {
+        let mut entry = self.buckets.entry(ip).or_insert_with(|| Bucket {
+            tokens: self.max_tokens,
+            last_refill: Instant::now(),
+        });
+
+        let now = Instant::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;
+
+        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()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_rate_limit_under() {
+        let limiter = RateLimiter::new(10.0, 10.0);
+        assert!(limiter.check(1));
+    }
+
+    #[test]
+    fn test_rate_limit_over() {
+        let limiter = RateLimiter::new(1.0, 1.0);
+        assert!(limiter.check(1));
+        assert!(!limiter.check(1));
+    }
+
+    #[test]
+    fn test_rate_limit_burst() {
+        let limiter = RateLimiter::new(1.0, 5.0);
+        for _ in 0..5 {
+            assert!(limiter.check(2));
+        }
+        assert!(!limiter.check(2));
+    }
+
+    #[test]
+    fn test_rate_limit_refill() {
+        let limiter = RateLimiter::new(100.0, 1.0);
+        assert!(limiter.check(3));
+        assert!(!limiter.check(3));
+        std::thread::sleep(Duration::from_millis(20));
+        assert!(limiter.check(3));
+    }
+}

crates/rampart-core/src/lib.rs

@@ -0,0 +1,9 @@
+pub mod config;
+pub mod crypto;
+pub mod filter;
+pub mod metrics;
+pub mod proxy;
+pub mod store;
+
+#[cfg(feature = "xdp")]
+pub mod xdp;

crates/rampart-core/src/main.rs

@@ -0,0 +1,82 @@
+use rampart_core::config::Config;
+use rampart_core::filter::blacklist::Blacklist;
+use rampart_core::filter::rate_limit::RateLimiter;
+use rampart_core::metrics;
+use rampart_core::proxy::listener::ProxyListener;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::watch;
+use tracing_subscriber::EnvFilter;
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(EnvFilter::from_default_env().add_directive("rampart_core=info".parse()?))
+        .init();
+
+    let config_path = std::env::var("RAMPART_CONFIG").unwrap_or_else(|_| "/etc/rampart/config.toml".to_string());
+    let config = Config::from_file(&config_path)?;
+    let config = Arc::new(config);
+
+    let rate_limiter = Arc::new(RateLimiter::new(
+        config.limits.rate_limit_login_pps,
+        config.limits.rate_limit_burst,
+    ));
+    let blacklist = Arc::new(Blacklist::new());
+
+    let (shutdown_tx, shutdown_rx) = watch::channel(false);
+
+    let sig_tx = shutdown_tx.clone();
+    tokio::spawn(async move {
+        wait_for_signal().await;
+        tracing::info!("shutdown signal received, draining connections...");
+        let _ = sig_tx.send(true);
+        tokio::time::sleep(Duration::from_secs(5)).await;
+        tracing::info!("shutdown timeout reached, exiting");
+        std::process::exit(0);
+    });
+
+    #[cfg(feature = "store-redis")]
+    if let Some(redis_url) = &config.store.redis_url
+        && !redis_url.is_empty()
+    {
+        let bl = blacklist.clone();
+        let sd = shutdown_rx.clone();
+        let url = redis_url.clone();
+        tokio::spawn(async move {
+            let client = match redis::Client::open(url.as_str()) {
+                Ok(c) => c,
+                Err(e) => {
+                    tracing::warn!("Invalid redis_url: {e}, blacklist sync disabled");
+                    return;
+                },
+            };
+            rampart_core::store::start_blacklist_sync(&client, bl, sd).await;
+        });
+    }
+
+    if config.metrics.enabled {
+        let metrics_addr = format!("0.0.0.0:{}", config.metrics.port);
+        tracing::info!("Metrics server listening on {metrics_addr}");
+        tokio::spawn(async move {
+            metrics::run_metrics_server(&metrics_addr).await;
+        });
+    }
+
+    tracing::info!("Rampart edge starting on {}:{}", config.bind.address, config.bind.port);
+    tracing::info!("Backend: {}:{}", config.backend.address, config.backend.port);
+
+    let listener = ProxyListener::new(config, rate_limiter, blacklist);
+    listener.run(shutdown_rx).await
+}
+
+async fn wait_for_signal() {
+    let ctrl_c = tokio::signal::ctrl_c();
+    let mut term = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+        .expect("failed to install SIGTERM handler");
+
+    tokio::select! {
+        _ = ctrl_c => {}
+        _ = term.recv() => {}
+    }
+}

crates/rampart-core/src/metrics.rs

@@ -0,0 +1,63 @@
+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 ACTIVE_CONNECTIONS: LazyLock<IntGauge> = LazyLock::new(|| {
+    register_int_gauge!("rampart_active_connections", "Active connections").expect("ACTIVE_CONNECTIONS")
+});
+
+pub static BLACKLIST_SIZE: LazyLock<IntGauge> =
+    LazyLock::new(|| register_int_gauge!("rampart_blacklist_size", "Blacklist entries").expect("BLACKLIST_SIZE"));
+
+pub static DEATH_CODE_BANS_TOTAL: LazyLock<IntCounterVec> = LazyLock::new(|| {
+    register_int_counter_vec!("rampart_death_code_bans_total", "Death code auto-bans", &["code"])
+        .expect("DEATH_CODE_BANS_TOTAL")
+});
+
+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;
+        });
+    }
+}

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

@@ -0,0 +1,215 @@
+use thiserror::Error;
+
+#[derive(Error, Debug)]
+pub enum ParseError {
+    #[error("Incomplete packet: {0}")]
+    Incomplete(&'static str),
+    #[error("VarInt too big (>5 bytes)")]
+    VarIntTooBig,
+    #[error("VarInt overflow")]
+    VarIntOverflow,
+    #[error("Invalid UTF-8 in string")]
+    InvalidUtf8,
+    #[error("String too long: {0}")]
+    StringTooLong(usize),
+    #[error("Hostname too long")]
+    HostnameTooLong,
+    #[error("Not a handshake packet: id={0}")]
+    NotHandshake(i32),
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub enum NextState {
+    Status,
+    Login,
+    Unknown(i32),
+}
+
+#[derive(Debug, Clone)]
+pub struct McHandshake {
+    pub protocol_version: i32,
+    pub server_address: String,
+    pub server_port: u16,
+    pub next_state: NextState,
+}
+
+impl McHandshake {
+    pub fn parse(buf: &[u8]) -> Result<Self, ParseError> {
+        let mut pos;
+        let (_, after_len) = read_varint(buf, 0)?;
+        pos = after_len;
+
+        let (packet_id, after_id) = read_varint(buf, pos)?;
+        pos = after_id;
+        if packet_id != 0x00 {
+            return Err(ParseError::NotHandshake(packet_id));
+        }
+
+        let (protocol_version, after_pv) = read_varint(buf, pos)?;
+        pos = after_pv;
+
+        let (server_address, after_addr) = read_string(buf, pos)?;
+        pos = after_addr;
+
+        if server_address.len() > 255 {
+            return Err(ParseError::HostnameTooLong);
+        }
+
+        if pos + 2 > buf.len() {
+            return Err(ParseError::Incomplete("missing port"));
+        }
+        let server_port = u16::from_be_bytes([buf[pos], buf[pos + 1]]);
+        pos += 2;
+
+        let (next_state_raw, _) = read_varint(buf, pos)?;
+        let next_state = match next_state_raw {
+            1 => NextState::Status,
+            2 => NextState::Login,
+            n => NextState::Unknown(n),
+        };
+
+        Ok(McHandshake {
+            protocol_version,
+            server_address,
+            server_port,
+            next_state,
+        })
+    }
+
+    pub fn is_login(&self) -> bool {
+        self.next_state == NextState::Login
+    }
+}
+
+pub fn read_varint(buf: &[u8], start: usize) -> Result<(i32, usize), ParseError> {
+    let mut value: i32 = 0;
+    let mut shift = 0;
+
+    for (i, &byte) in buf[start..].iter().enumerate() {
+        if i >= 5 {
+            return Err(ParseError::VarIntTooBig);
+        }
+        let segment = (byte & 0x7F) as i32;
+        if shift >= 32 || (shift == 28 && segment > 0x0F) {
+            return Err(ParseError::VarIntOverflow);
+        }
+        value |= segment << shift;
+        shift += 7;
+        if (byte & 0x80) == 0 {
+            return Ok((value, start + i + 1));
+        }
+    }
+    Err(ParseError::Incomplete("varint"))
+}
+
+pub fn read_string(buf: &[u8], start: usize) -> Result<(String, usize), ParseError> {
+    let (len, after_len) = read_varint(buf, start)?;
+    if !(0..=32767).contains(&len) {
+        return Err(ParseError::StringTooLong(len as usize));
+    }
+    let end = after_len + len as usize;
+    if end > buf.len() {
+        return Err(ParseError::Incomplete("string data"));
+    }
+    let s = std::str::from_utf8(&buf[after_len..end])
+        .map_err(|_| ParseError::InvalidUtf8)?
+        .to_string();
+    Ok((s, end))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_varint_zero() {
+        let buf = vec![0x00];
+        assert_eq!(read_varint(&buf, 0).unwrap(), (0, 1));
+    }
+
+    #[test]
+    fn test_varint_single() {
+        let buf = vec![0x7F];
+        assert_eq!(read_varint(&buf, 0).unwrap(), (127, 1));
+    }
+
+    #[test]
+    fn test_varint_multi() {
+        let buf = vec![0x80, 0x01];
+        assert_eq!(read_varint(&buf, 0).unwrap(), (128, 2));
+    }
+
+    #[test]
+    fn test_varint_max() {
+        let buf = vec![0xFF, 0xFF, 0xFF, 0xFF, 0x07];
+        assert_eq!(read_varint(&buf, 0).unwrap(), (i32::MAX, 5));
+    }
+
+    #[test]
+    fn test_varint_overflow() {
+        let buf = vec![0xFF, 0xFF, 0xFF, 0xFF, 0x10];
+        assert!(matches!(read_varint(&buf, 0), Err(ParseError::VarIntOverflow)));
+    }
+
+    #[test]
+    fn test_varint_incomplete() {
+        let buf = vec![0x80];
+        assert!(matches!(read_varint(&buf, 0), Err(ParseError::Incomplete(_))));
+    }
+
+    #[test]
+    fn test_handshake_login() {
+        let addr = b"play.example.com";
+        let mut buf = Vec::new();
+
+        buf.extend_from_slice(&[0x00]);
+        buf.push(0x00);
+        write_varint(&mut buf, 765);
+        write_varint(&mut buf, addr.len() as i32);
+        buf.extend_from_slice(addr);
+        buf.extend_from_slice(&[0x63, 0xDD]);
+        buf.push(0x02);
+
+        let len = (buf.len() - 1) as u8;
+        buf[0] = len;
+
+        let hs = McHandshake::parse(&buf).unwrap();
+        assert_eq!(hs.protocol_version, 765);
+        assert_eq!(hs.server_address, "play.example.com");
+        assert_eq!(hs.server_port, 25565);
+        assert!(hs.is_login());
+    }
+
+    fn write_varint(buf: &mut Vec<u8>, mut value: i32) {
+        loop {
+            if (value & !0x7F) == 0 {
+                buf.push(value as u8);
+                return;
+            }
+            buf.push((value as u8 & 0x7F) | 0x80);
+            value = (value >> 7) & (i32::MAX >> 6);
+        }
+    }
+
+    #[test]
+    fn test_handshake_status() {
+        let addr = b"play.example";
+        let mut buf = Vec::new();
+        // packet length will be set below
+        buf.push(0x00);
+        buf.push(0x00); // packet ID
+        buf.push(0x02); // protocol version 2
+        write_varint(&mut buf, addr.len() as i32);
+        buf.extend_from_slice(addr);
+        buf.extend_from_slice(&[0x63, 0xDD]); // port 25565
+        buf.push(0x01); // next state = status
+        let len = (buf.len() - 1) as u8;
+        buf[0] = len;
+
+        let hs = McHandshake::parse(&buf).unwrap();
+        assert_eq!(hs.protocol_version, 2);
+        assert_eq!(hs.server_address, "play.example");
+        assert_eq!(hs.server_port, 25565);
+        assert_eq!(hs.next_state, NextState::Status);
+    }
+}

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

@@ -0,0 +1,95 @@
+use crate::config::Config;
+use crate::filter::blacklist::Blacklist;
+use crate::filter::rate_limit::RateLimiter;
+use crate::proxy::tunnel::ConnectionHandler;
+use socket2::{Domain, Socket, Type};
+use std::sync::Arc;
+use tokio::net::TcpListener;
+use tokio::sync::watch;
+
+pub struct ProxyListener {
+    config: Arc<Config>,
+    rate_limiter: Arc<RateLimiter>,
+    blacklist: Arc<Blacklist>,
+}
+
+impl ProxyListener {
+    pub fn new(config: Arc<Config>, rate_limiter: Arc<RateLimiter>, blacklist: Arc<Blacklist>) -> Self {
+        Self {
+            config,
+            rate_limiter,
+            blacklist,
+        }
+    }
+
+    pub async fn run(&self, shutdown: watch::Receiver<bool>) -> anyhow::Result<()> {
+        let addr = format!("{}:{}", self.config.bind.address, self.config.bind.port).parse::<std::net::SocketAddr>()?;
+        let workers = self.config.workers.count.max(1);
+
+        let mut handles = Vec::with_capacity(workers);
+        for _ in 0..workers {
+            let listener = build_listener(addr)?;
+            let config = self.config.clone();
+            let rate_limiter = self.rate_limiter.clone();
+            let blacklist = self.blacklist.clone();
+            let shutdown = shutdown.clone();
+            handles.push(tokio::spawn(accept_loop(
+                listener,
+                config,
+                rate_limiter,
+                blacklist,
+                shutdown,
+            )));
+        }
+
+        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,
+    config: Arc<Config>,
+    rate_limiter: Arc<RateLimiter>,
+    blacklist: Arc<Blacklist>,
+    mut shutdown: watch::Receiver<bool>,
+) -> anyhow::Result<()> {
+    loop {
+        tokio::select! {
+            biased;
+            _ = shutdown.changed() => {
+                if *shutdown.borrow() {
+                    tracing::info!("shutdown signal received, stopping accept loop");
+                    return Ok(());
+                }
+            }
+            result = listener.accept() => {
+                let (stream, peer_addr) = match result {
+                    Ok(conn) => conn,
+                    Err(e) => {
+                        tracing::error!("accept error: {e}");
+                        continue;
+                    }
+                };
+                let handler = ConnectionHandler::new(config.clone(), rate_limiter.clone(), blacklist.clone());
+                tokio::spawn(async move {
+                    if let Err(e) = handler.handle(stream, peer_addr).await {
+                        tracing::debug!("connection from {peer_addr}: {e}");
+                    }
+                });
+            }
+        }
+    }
+}

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

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

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

@@ -0,0 +1,204 @@
+use crate::config::Config;
+use crate::crypto::hmac;
+use crate::filter::blacklist::Blacklist;
+use crate::filter::death_code;
+use crate::filter::rate_limit::RateLimiter;
+use crate::metrics;
+use crate::proxy::handshake::{McHandshake, read_varint};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+
+pub struct ConnectionHandler {
+    config: Arc<Config>,
+    rate_limiter: Arc<RateLimiter>,
+    blacklist: Arc<Blacklist>,
+}
+
+impl ConnectionHandler {
+    pub fn new(config: Arc<Config>, rate_limiter: Arc<RateLimiter>, blacklist: Arc<Blacklist>) -> Self {
+        Self {
+            config,
+            rate_limiter,
+            blacklist,
+        }
+    }
+
+    fn ip_to_u32(addr: std::net::SocketAddr) -> u32 {
+        match addr.ip() {
+            std::net::IpAddr::V4(ip) => ip.to_bits(),
+            _ => 0,
+        }
+    }
+
+    pub async fn handle(&self, mut client: TcpStream, peer_addr: std::net::SocketAddr) -> anyhow::Result<()> {
+        let ip_u32 = Self::ip_to_u32(peer_addr);
+
+        if self.blacklist.is_blocked(ip_u32) {
+            metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();
+            return Ok(());
+        }
+
+        if !self.rate_limiter.check(ip_u32) {
+            metrics::RATE_LIMIT_HITS.with_label_values(&["hit"]).inc();
+            metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();
+            return Ok(());
+        }
+
+        let timeout = Duration::from_secs(self.config.limits.handshake_timeout_secs);
+        let mut buf = vec![0u8; 4096];
+        let n = tokio::time::timeout(timeout, client.read(&mut buf)).await??;
+        if n == 0 {
+            return Ok(());
+        }
+
+        let parsed = McHandshake::parse(&buf[..n]);
+        match parsed {
+            Ok(handshake) => {
+                if !self.rate_limiter.check(ip_u32) {
+                    metrics::RATE_LIMIT_HITS.with_label_values(&["hit"]).inc();
+                    metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();
+                    return Ok(());
+                }
+
+                metrics::CONNECTIONS_TOTAL.with_label_values(&["allowed"]).inc();
+
+                let backend_addr = format!("{}:{}", self.config.backend.address, self.config.backend.port);
+                let mut backend = TcpStream::connect(&backend_addr).await?;
+
+                let signed = hmac::sign_hostname(&handshake.server_address, self.config.hmac.secret.as_bytes());
+                let modified = replace_hostname(&buf[..n], &handshake.server_address, &signed)?;
+                backend.write_all(&modified).await?;
+
+                tokio::io::copy_bidirectional(&mut client, &mut backend).await?;
+            },
+            Err(e) => {
+                tracing::debug!("parse error from {peer_addr}: {e}");
+                metrics::CONNECTIONS_TOTAL.with_label_values(&["blocked"]).inc();
+
+                if self.config.death_code.enabled {
+                    if let Some(code) = death_code::detect(&buf[..n]) {
+                        let duration = Duration::from_secs(self.config.death_code.ban_duration_secs);
+                        self.blacklist.add(ip_u32, duration, code.as_str());
+                        metrics::DEATH_CODE_BANS_TOTAL.with_label_values(&[code.as_str()]).inc();
+                        tracing::info!("death code ban {peer_addr}: {}", code.as_str());
+                    }
+                }
+            },
+        }
+        Ok(())
+    }
+}
+
+fn replace_hostname(original: &[u8], _old_hostname: &str, new_hostname: &str) -> anyhow::Result<Vec<u8>> {
+    let (packet_len, after_packet_len) =
+        read_varint(original, 0).map_err(|_| anyhow::anyhow!("corrupt packet length"))?;
+    let mut pos = after_packet_len;
+
+    let (_packet_id, after_id) = read_varint(original, pos).map_err(|_| anyhow::anyhow!("corrupt packet id"))?;
+    pos = after_id;
+
+    let (_protocol_version, after_pv) =
+        read_varint(original, pos).map_err(|_| anyhow::anyhow!("corrupt protocol version"))?;
+    pos = after_pv;
+
+    let (old_host_len, host_field_start) =
+        read_varint(original, pos).map_err(|_| anyhow::anyhow!("corrupt hostname length"))?;
+    let host_data_end = host_field_start + old_host_len as usize;
+    let old_field_size = host_data_end - pos;
+
+    let new_hostname_bytes = new_hostname.as_bytes();
+    let new_len_field_bytes = varint_bytes(new_hostname_bytes.len() as i32);
+    let new_field_size = new_len_field_bytes.len() + new_hostname_bytes.len();
+    let size_diff = new_field_size as isize - old_field_size as isize;
+    let new_packet_len = (packet_len as isize + size_diff) as i32;
+
+    let cap = original.len().wrapping_add(size_diff as usize);
+    let mut result = Vec::with_capacity(cap);
+
+    result.extend_from_slice(&varint_bytes(new_packet_len));
+    result.extend_from_slice(&original[after_packet_len..pos]);
+    result.extend_from_slice(&new_len_field_bytes);
+    result.extend_from_slice(new_hostname_bytes);
+    result.extend_from_slice(&original[host_data_end..]);
+
+    Ok(result)
+}
+
+fn varint_bytes(mut value: i32) -> Vec<u8> {
+    let mut result = Vec::with_capacity(5);
+    loop {
+        if (value & !0x7F) == 0 {
+            result.push(value as u8);
+            return result;
+        }
+        result.push((value as u8 & 0x7F) | 0x80);
+        value >>= 7;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn build_test_packet(hostname: &str) -> Vec<u8> {
+        let addr = hostname.as_bytes();
+        let mut buf = Vec::new();
+        buf.push(0x00);
+        buf.extend_from_slice(&varint_bytes(765));
+        buf.extend_from_slice(&varint_bytes(addr.len() as i32));
+        buf.extend_from_slice(addr);
+        buf.extend_from_slice(&[0x63, 0xDD]);
+        buf.push(0x02);
+
+        let len = buf.len() as i32;
+        let mut pkt = varint_bytes(len);
+        pkt.extend_from_slice(&buf);
+        pkt
+    }
+
+    #[test]
+    fn test_replace_hostname_basic() {
+        let pkt = build_test_packet("play.example.com");
+        let new_hostname = "play.example.com\0shield\0abcdef1234567890";
+        let modified = replace_hostname(&pkt, "play.example.com", new_hostname).unwrap();
+        assert!(modified.len() > pkt.len());
+
+        let parsed = McHandshake::parse(&modified).unwrap();
+        assert_eq!(parsed.server_address, new_hostname);
+    }
+
+    #[test]
+    fn test_replace_hostname_shorter() {
+        let pkt = build_test_packet("very.long.hostname.example.com");
+        let new_hostname = "short.com";
+        let modified = replace_hostname(&pkt, "very.long.hostname.example.com", new_hostname).unwrap();
+        assert!(modified.len() < pkt.len());
+
+        let parsed = McHandshake::parse(&modified).unwrap();
+        assert_eq!(parsed.server_address, new_hostname);
+    }
+
+    #[test]
+    fn test_replace_hostname_preserves_port_and_protocol() {
+        let pkt = build_test_packet("mc.example.com");
+        let new_hostname = "mc.example.com\0shield\x00deadbeef";
+        let modified = replace_hostname(&pkt, "mc.example.com", new_hostname).unwrap();
+
+        let parsed = McHandshake::parse(&modified).unwrap();
+        assert_eq!(parsed.server_port, 25565);
+        assert_eq!(parsed.protocol_version, 765);
+        assert!(parsed.is_login());
+    }
+
+    #[test]
+    fn test_varint_roundtrip() {
+        let cases = vec![0, 1, 127, 128, 255, 65535, 1000000, i32::MAX];
+        for val in cases {
+            let bytes = varint_bytes(val);
+            let (decoded, _) = read_varint(&bytes, 0).unwrap();
+            assert_eq!(decoded, val, "roundtrip failed for {val}");
+        }
+    }
+}

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

@@ -0,0 +1,29 @@
+#[cfg(feature = "store-redis")]
+pub mod redis;
+#[cfg(feature = "store-redis")]
+pub use redis::start_blacklist_sync;
+
+#[allow(async_fn_in_trait)]
+pub trait StateStore: Send + Sync {
+    async fn get(&self, key: &str) -> anyhow::Result<Option<String>>;
+    async fn set(&self, key: &str, value: &str) -> anyhow::Result<()>;
+    async fn del(&self, key: &str) -> anyhow::Result<()>;
+    async fn publish(&self, channel: &str, message: &str) -> anyhow::Result<()>;
+}
+
+pub struct NoopStore;
+
+impl StateStore for NoopStore {
+    async fn get(&self, _key: &str) -> anyhow::Result<Option<String>> {
+        Ok(None)
+    }
+    async fn set(&self, _key: &str, _value: &str) -> anyhow::Result<()> {
+        Ok(())
+    }
+    async fn del(&self, _key: &str) -> anyhow::Result<()> {
+        Ok(())
+    }
+    async fn publish(&self, _channel: &str, _message: &str) -> anyhow::Result<()> {
+        Ok(())
+    }
+}

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

@@ -0,0 +1,136 @@
+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_parts: Vec<&str> = event.ip.split('.').collect();
+    if ip_parts.len() != 4 {
+        anyhow::bail!("invalid IP: {}", event.ip);
+    }
+    let mut ip_u32: u32 = 0;
+    for part in &ip_parts {
+        let octet: u32 = part.parse()?;
+        ip_u32 = (ip_u32 << 8) | octet;
+    }
+
+    match event.action.as_str() {
+        "ban" => {
+            blacklist.add(ip_u32, Duration::from_secs(event.duration_secs), "redis");
+            tracing::info!("blacklist add via Redis: {}", event.ip);
+        },
+        "unban" => {
+            blacklist.remove(ip_u32);
+            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(())
+    }
+}

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

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