From 6863249ad460b0ebaab0e7b5c3148b6eea2c3029 Mon Sep 17 00:00:00 2001 From: loki5512344 Date: Mon, 24 Aug 2026 12:18:22 +0200 Subject: [PATCH] feat: HTTP protocol plugin + XDP globals patching from config protocol-http (compile-time feature): - HttpHandler: incremental HTTP/1.1 request parse (fragment-safe), header size/timeout limits, method whitelist, Host required, blocked_paths, optional User-Agent requirement - [protocol.http] config section; registry wiring behind feature XDP globals: - src/xdp/globals.rs: XdpGlobals + build_rodata_image() mirroring xdp/core/config.h layout; set via OpenMapMut::set_initial_value() before load ([xdp] section: ports, udp policy, throttle, challenge) - wire set_globals into rampart binary startup fix: gate preflight() behind xdp feature (dead code without it) --- Cargo.toml | 4 +- src/bin/rampart.rs | 49 ++++++--- src/config/mod.rs | 14 +++ src/config/sections.rs | 99 ++++++++++++++++++ src/protocol/http/mod.rs | 112 +++++++++++++++++++++ src/protocol/http/parser.rs | 196 ++++++++++++++++++++++++++++++++++++ src/protocol/mod.rs | 3 + src/xdp/diagnostics.rs | 1 + src/xdp/filter.rs | 29 +++++- src/xdp/globals.rs | 182 +++++++++++++++++++++++++++++++++ src/xdp/mod.rs | 5 + tests/config_parse.rs | 51 ++++++++++ tests/http_protocol.rs | 131 ++++++++++++++++++++++++ tests/xdp_globals.rs | 118 ++++++++++++++++++++++ 14 files changed, 972 insertions(+), 22 deletions(-) create mode 100644 src/protocol/http/mod.rs create mode 100644 src/protocol/http/parser.rs create mode 100644 src/xdp/globals.rs create mode 100644 tests/http_protocol.rs create mode 100644 tests/xdp_globals.rs diff --git a/Cargo.toml b/Cargo.toml index d59d62d..36897e1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -37,8 +37,8 @@ store-redis = ["dep:redis"] geoip = ["dep:maxminddb"] xdp = ["dep:libbpf-rs", "dep:libc"] io-uring = ["dep:tokio-splice"] -# Резерв под реализацию HTTP-протокольного обработчика (модель plugin-by-feature). -# Реализаций пока нет: реестр протоколов пуст и edge-нода требует явного флага. +# HTTP/1.1 protocol handler: парсинг request-head, verdict Allow/Reject, +# туннелирование на upstream (см. src/protocol/http/ и [protocol.http] в конфиге). protocol-http = [] [dependencies] diff --git a/src/bin/rampart.rs b/src/bin/rampart.rs index 0c5e506..08154ef 100644 --- a/src/bin/rampart.rs +++ b/src/bin/rampart.rs @@ -14,6 +14,8 @@ use rampart::traffic::detector::{AttackDetector, AttackStatus}; use rampart::traffic::hook::TrafficHook; use rampart::traffic::reputation::IpReputation; use rampart::xdp::XdpFilter; +#[cfg(feature = "xdp")] +use rampart::xdp::XdpGlobals; use std::collections::HashSet; use std::net::IpAddr; use std::sync::Arc; @@ -33,19 +35,30 @@ fn attack_status_value(status: AttackStatus) -> i64 { /// Собирает реестр протоколов из скомпилированных реализаций. /// -/// В ядре реализаций нет: `protocol-http` — зарезервированная пустая feature, -/// показывающая модель plugin-by-feature. Внешние крейты регистрируют свои -/// обработчики в этом месте. -#[allow(unused_mut)] -fn build_registry() -> ProtocolRegistry { - let mut registry = ProtocolRegistry::new(); +/// Обработчики поставляются фичами: `protocol-http` регистрирует +/// HTTP/1.1-обработчик с политикой из секции `[protocol.http]`. +/// Внешние крейты регистрируют свои обработчики в этом же месте. +fn build_registry(config: &Config) -> ProtocolRegistry { #[cfg(feature = "protocol-http")] - register_http(&mut registry); - registry + { + let mut registry = ProtocolRegistry::new(); + register_http(&mut registry, config); + registry + } + #[cfg(not(feature = "protocol-http"))] + { + let _ = config; + ProtocolRegistry::new() + } } #[cfg(feature = "protocol-http")] -fn register_http(_registry: &mut ProtocolRegistry) {} +fn register_http(registry: &mut ProtocolRegistry, config: &Config) { + registry.register(Box::new(rampart::protocol::http::HttpProtocolHandler::new( + &config.protocol.http, + &config.backend.upstreams, + ))); +} #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -53,16 +66,17 @@ async fn main() -> anyhow::Result<()> { .with_env_filter(EnvFilter::from_default_env().add_directive("rampart=info".parse()?)) .init(); - let registry = Arc::new(build_registry()); + let config_path = std::env::var("RAMPART_CONFIG").unwrap_or_else(|_| "/etc/rampart/config.toml".to_string()); + let config = Arc::new(Config::from_file(&config_path)?); + + let registry = Arc::new(build_registry(&config)); if registry.is_empty() { anyhow::bail!( - "no protocol plugins compiled; build with --features protocol-http \ + "no protocol plugins compiled; available feature flags: protocol-http, \ + store-redis, geoip, xdp, io-uring. Build with --features protocol-http \ or link an external ProtocolHandler implementation" ); } - - let config_path = std::env::var("RAMPART_CONFIG").unwrap_or_else(|_| "/etc/rampart/config.toml".to_string()); - let config = Arc::new(Config::from_file(&config_path)?); let whitelist = build_whitelist(&config)?; let rate_limiter = Arc::new(RateLimiter::new( @@ -239,7 +253,12 @@ fn start_xdp( let filter = XdpFilter::new(&config.xdp.interface); let shared = Arc::new(Mutex::new(filter)); - shared.lock().expect("xdp lock poisoned").load()?; + { + let mut guard = shared.lock().expect("xdp lock poisoned"); + let globals = XdpGlobals::from_config(&config.xdp)?; + guard.set_globals(globals); + guard.load()?; + } let xdp_metrics = XdpMetrics::register()?; let sd = shutdown_rx.clone(); diff --git a/src/config/mod.rs b/src/config/mod.rs index e902d8b..e63c884 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -5,6 +5,8 @@ pub use sections::{ BackendConfig, BanConfig, BindConfig, DetectAlertConfig, DetectAutobanConfig, DetectConfig, DetectPrefixConfig, LimitsConfig, LoggingConfig, MetricsConfig, PowConfig, StoreConfig, WorkerConfig, XdpConfig, }; +#[cfg(feature = "protocol-http")] +pub use sections::{HttpProtocolConfig, ProtocolConfig}; use serde::Deserialize; use std::fs; @@ -33,6 +35,9 @@ pub struct Config { pub pow: PowConfig, #[serde(default)] pub detect: DetectConfig, + #[cfg(feature = "protocol-http")] + #[serde(default)] + pub protocol: ProtocolConfig, #[serde(default)] pub whitelist: Vec, } @@ -75,6 +80,15 @@ impl Config { if self.detect.prefix.enabled && self.detect.prefix.syn_threshold == 0 { anyhow::bail!("detect.prefix.syn_threshold must be positive when enabled"); } + #[cfg(feature = "protocol-http")] + { + if self.protocol.http.max_header_bytes == 0 { + anyhow::bail!("protocol.http.max_header_bytes must be positive"); + } + if self.protocol.http.timeout_secs == 0 { + anyhow::bail!("protocol.http.timeout_secs must be positive"); + } + } Ok(()) } } diff --git a/src/config/sections.rs b/src/config/sections.rs index 7d42790..d558660 100644 --- a/src/config/sections.rs +++ b/src/config/sections.rs @@ -143,6 +143,28 @@ pub struct XdpConfig { pub enabled: bool, #[serde(default = "default_xdp_interface")] pub interface: String, + #[serde(default = "default_xdp_port_start")] + pub protected_port_start: u16, + #[serde(default = "default_xdp_port_end")] + pub protected_port_end: u16, + /// "pass" | "drop" | "rate-limit" → G_UDP_POLICY. + #[serde(default = "default_xdp_udp_policy")] + pub udp_policy: String, + #[serde(default = "default_xdp_udp_hit_count")] + pub udp_rate_hit_count: u32, + #[serde(default = "default_xdp_udp_window_ms")] + pub udp_rate_window_ms: u64, + #[serde(default)] + pub syn_challenge_enabled: bool, + /// 128-битный hex (16 символов) для G_CHALLENGE_SECRET; None = плейсхолдер из config.h. + #[serde(default)] + pub challenge_secret_hex: Option, + #[serde(default = "default_xdp_challenge_timeout_ms")] + pub challenge_timeout_ms: u32, + #[serde(default = "default_xdp_throttle_enabled")] + pub throttle_enabled: bool, + #[serde(default = "default_xdp_events_enabled")] + pub events_enabled: bool, } impl Default for XdpConfig { @@ -150,6 +172,16 @@ impl Default for XdpConfig { Self { enabled: false, interface: default_xdp_interface(), + protected_port_start: default_xdp_port_start(), + protected_port_end: default_xdp_port_end(), + udp_policy: default_xdp_udp_policy(), + udp_rate_hit_count: default_xdp_udp_hit_count(), + udp_rate_window_ms: default_xdp_udp_window_ms(), + syn_challenge_enabled: false, + challenge_secret_hex: None, + challenge_timeout_ms: default_xdp_challenge_timeout_ms(), + throttle_enabled: default_xdp_throttle_enabled(), + events_enabled: default_xdp_events_enabled(), } } } @@ -157,6 +189,30 @@ impl Default for XdpConfig { fn default_xdp_interface() -> String { "eth0".to_string() } +fn default_xdp_port_start() -> u16 { + 1 +} +fn default_xdp_port_end() -> u16 { + 65535 +} +fn default_xdp_udp_policy() -> String { + "pass".to_string() +} +fn default_xdp_udp_hit_count() -> u32 { + 100 +} +fn default_xdp_udp_window_ms() -> u64 { + 1000 +} +fn default_xdp_challenge_timeout_ms() -> u32 { + 3000 +} +fn default_xdp_throttle_enabled() -> bool { + true +} +fn default_xdp_events_enabled() -> bool { + true +} #[derive(Debug, Clone, Deserialize)] pub struct LoggingConfig { @@ -303,3 +359,46 @@ fn default_prefix_window_secs() -> u64 { fn default_prefix_min_unique_sources() -> u64 { 16 } + +/// Секция `[protocol]`: протокольные обработчики (plugin-by-feature). +#[cfg(feature = "protocol-http")] +#[derive(Debug, Clone, Default, Deserialize)] +pub struct ProtocolConfig { + #[serde(default)] + pub http: HttpProtocolConfig, +} + +/// Секция `[protocol.http]`: политика HTTP/1.1-обработчика на edge-ноде. +#[cfg(feature = "protocol-http")] +#[derive(Debug, Clone, Deserialize)] +pub struct HttpProtocolConfig { + #[serde(default = "default_http_max_header_bytes")] + pub max_header_bytes: usize, + #[serde(default = "default_http_timeout_secs")] + pub timeout_secs: u64, + #[serde(default)] + pub blocked_paths: Vec, + #[serde(default)] + pub require_user_agent: bool, +} + +#[cfg(feature = "protocol-http")] +impl Default for HttpProtocolConfig { + fn default() -> Self { + Self { + max_header_bytes: default_http_max_header_bytes(), + timeout_secs: default_http_timeout_secs(), + blocked_paths: Vec::new(), + require_user_agent: false, + } + } +} + +#[cfg(feature = "protocol-http")] +fn default_http_max_header_bytes() -> usize { + 8192 +} +#[cfg(feature = "protocol-http")] +fn default_http_timeout_secs() -> u64 { + 5 +} diff --git a/src/protocol/http/mod.rs b/src/protocol/http/mod.rs new file mode 100644 index 0000000..94cfa0e --- /dev/null +++ b/src/protocol/http/mod.rs @@ -0,0 +1,112 @@ +//! HTTP/1.1 protocol handler (feature `protocol-http`). +//! +//! Первый plugin-реализаций [`ProtocolHandler`]: читает request-head +//! постепенно (TCP приходит фрагментами), проверяет его по политике из +//! секции `[protocol.http]` и возвращает upstream для туннелирования. +//! Голова читается через `peek`, чтобы байты остались в сокете и движок +//! проксировал запрос на backend без потерь. + +mod parser; + +use crate::config::HttpProtocolConfig; +use crate::protocol::{ProtocolHandler, Upstream}; +use std::future::Future; +use std::pin::Pin; +use std::time::Duration; +use tokio::net::TcpStream; + +/// Обработчик HTTP/1.1: verdict Allow → tunnel на upstream, Reject → drop. +pub struct HttpProtocolHandler { + policy: parser::Policy, + timeout: Duration, + upstream: String, +} + +impl HttpProtocolHandler { + /// Собирает обработчик из конфига `[protocol.http]` и списка upstream'ов. + #[must_use] + pub fn new(http: &HttpProtocolConfig, upstreams: &[String]) -> Self { + Self { + policy: parser::Policy { + max_header_bytes: http.max_header_bytes, + blocked_paths: http.blocked_paths.clone(), + require_user_agent: http.require_user_agent, + }, + timeout: Duration::from_secs(http.timeout_secs), + upstream: upstreams.first().cloned().unwrap_or_else(|| "127.0.0.1:80".to_string()), + } + } + + /// Накапливает request-head через `peek` до терминатора или лимита. + /// + /// # Errors + /// Таймаут, закрытие клиента, превышение `max_header_bytes` или ошибка I/O. + async fn read_head(&self, stream: &TcpStream) -> anyhow::Result> { + let limit = self.policy.max_header_bytes; + if limit == 0 { + anyhow::bail!("protocol.http.max_header_bytes must be positive"); + } + let mut buf = vec![0u8; limit]; + let mut len = 0usize; + let accumulate = async { + loop { + if let Some(head_len) = parser::scan_head(&buf[..len]) { + buf.truncate(head_len); + return Ok(buf); + } + if len >= limit { + anyhow::bail!("request head exceeds {limit} bytes"); + } + let n = stream.peek(&mut buf[len..]).await?; + if n == 0 { + anyhow::bail!("client closed connection before head was complete"); + } + len += n; + } + }; + match tokio::time::timeout(self.timeout, accumulate).await { + Ok(result) => result, + Err(_) => anyhow::bail!("timed out reading request head"), + } + } +} + +impl ProtocolHandler for HttpProtocolHandler { + fn name(&self) -> &'static str { + "http" + } + + fn handshake<'a>( + &'a self, + stream: &'a mut TcpStream, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + let head = self.read_head(stream).await?; + match parser::evaluate(&head, &self.policy) { + parser::Verdict::Allow { .. } => Ok(Upstream::new(self.upstream.clone())), + parser::Verdict::Reject { reason } => Err(anyhow::anyhow!("http: {reason}")), + } + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn zero_limit_fails_fast_without_io() { + let config = HttpProtocolConfig::default(); + let mut broken = config.clone(); + broken.max_header_bytes = 0; + let handler = HttpProtocolHandler::new(&broken, &[]); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let client = tokio::net::TcpStream::connect(listener.local_addr().expect("addr")) + .await + .expect("connect"); + let (server, _) = listener.accept().await.expect("accept"); + drop(listener); + assert!(handler.read_head(&server).await.is_err()); + drop(client); + } +} diff --git a/src/protocol/http/parser.rs b/src/protocol/http/parser.rs new file mode 100644 index 0000000..c7c5a69 --- /dev/null +++ b/src/protocol/http/parser.rs @@ -0,0 +1,196 @@ +//! Чистый парсер и валидатор HTTP/1.x request-head для edge-фильтра. +//! +//! Модуль не знает про I/O: на вход — накопленный заголовок, на выходе — +//! [`Verdict`]. Это позволяет покрывать логику юнит-тестами без сокетов. + +/// Политика проверки, снятая с секции `[protocol.http]`. +#[derive(Debug, Clone)] +pub struct Policy { + pub max_header_bytes: usize, + pub blocked_paths: Vec, + pub require_user_agent: bool, +} + +/// Результат проверки головы запроса. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum Verdict { + Allow { method: String, path: String, host: String }, + Reject { reason: String }, +} + +const ALLOWED_METHODS: [&str; 7] = ["GET", "POST", "PUT", "DELETE", "HEAD", "OPTIONS", "PATCH"]; +const HEADER_TERMINATOR: &[u8] = b"\r\n\r\n"; + +/// Ищет конец головы запроса (`\r\n\r\n`) и возвращает её полную длину. +#[must_use] +pub fn scan_head(buf: &[u8]) -> Option { + buf.windows(HEADER_TERMINATOR.len()) + .position(|window| window == HEADER_TERMINATOR) + .map(|pos| pos + HEADER_TERMINATOR.len()) +} + +/// Разбирает и валидирует голову запроса согласно политике. +#[must_use] +pub fn evaluate(head: &[u8], policy: &Policy) -> Verdict { + let Ok(text) = std::str::from_utf8(head) else { + return reject("request head is not valid utf-8"); + }; + let mut lines = text.split("\r\n"); + let Some(request_line) = lines.next().filter(|line| !line.is_empty()) else { + return reject("empty request line"); + }; + + let mut parts = request_line.split(' '); + let (Some(method), Some(target), Some(version)) = (parts.next(), parts.next(), parts.next()) else { + return reject("malformed request line"); + }; + if parts.next().is_some() { + return reject("malformed request line"); + } + if !matches!(version, "HTTP/1.0" | "HTTP/1.1") { + return reject("unsupported http version"); + } + if !ALLOWED_METHODS.contains(&method) { + return reject("method not allowed"); + } + + let mut host: Option<&str> = None; + let mut user_agent: Option<&str> = None; + for line in lines { + if line.is_empty() { + continue; + } + let Some((name, value)) = line.split_once(':') else { + return reject("malformed header line"); + }; + let value = value.trim(); + if name.eq_ignore_ascii_case("host") { + host = Some(value); + } else if name.eq_ignore_ascii_case("user-agent") { + user_agent = Some(value); + } + } + + let path = match target.split_once('?') { + Some((path, _)) => path, + None => target, + }; + if policy + .blocked_paths + .iter() + .any(|prefix| path.starts_with(prefix.as_str())) + { + return reject("path is blocklisted"); + } + + let Some(host) = host.filter(|value| !value.is_empty()) else { + return reject("missing host header"); + }; + if policy.require_user_agent && user_agent.is_none_or(str::is_empty) { + return reject("missing user-agent header"); + } + + Verdict::Allow { + method: method.to_string(), + path: path.to_string(), + host: host.to_string(), + } +} + +fn reject(reason: &str) -> Verdict { + Verdict::Reject { + reason: reason.to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn policy() -> Policy { + Policy { + max_header_bytes: 8192, + blocked_paths: Vec::new(), + require_user_agent: false, + } + } + + #[test] + fn scan_finds_full_head_length() { + assert_eq!(scan_head(b"GET / HTTP/1.1\r\n\r\n"), Some(18)); + assert_eq!(scan_head(b"GET / HTTP/1.1\r\n"), None); + } + + #[test] + fn valid_request_is_allowed() { + let head = b"POST /api/items?limit=5 HTTP/1.1\r\nHost: example.com\r\n\r\n"; + let verdict = evaluate(head, &policy()); + assert_eq!( + verdict, + Verdict::Allow { + method: "POST".to_string(), + path: "/api/items".to_string(), + host: "example.com".to_string() + } + ); + } + + #[test] + fn unknown_method_is_rejected() { + let head = b"TRACE / HTTP/1.1\r\nHost: h\r\n\r\n"; + assert!(matches!(evaluate(head, &policy()), Verdict::Reject { .. })); + } + + #[test] + fn missing_host_is_rejected() { + let head = b"GET / HTTP/1.1\r\nUser-Agent: t\r\n\r\n"; + assert!(matches!(evaluate(head, &policy()), Verdict::Reject { .. })); + } + + #[test] + fn http2_version_is_rejected() { + let head = b"GET / HTTP/2.0\r\nHost: h\r\n\r\n"; + assert!(matches!(evaluate(head, &policy()), Verdict::Reject { .. })); + } + + #[test] + fn malformed_line_is_rejected() { + assert!(matches!( + evaluate(b"nonsense\r\n\r\n", &policy()), + Verdict::Reject { .. } + )); + assert!(matches!( + evaluate(b"GET / HTTP/1.1\r\nbroken\r\n\r\n", &policy()), + Verdict::Reject { .. } + )); + } + + #[test] + fn non_utf8_head_is_rejected() { + assert!(matches!(evaluate(&[0xff, 0xfe], &policy()), Verdict::Reject { .. })); + } + + #[test] + fn blocked_path_prefix_is_rejected() { + let mut custom = policy(); + custom.blocked_paths = vec!["/admin".to_string()]; + assert!(matches!( + evaluate(b"GET /admin/panel HTTP/1.1\r\nHost: h\r\n\r\n", &custom), + Verdict::Reject { .. } + )); + assert!(matches!( + evaluate(b"GET /public HTTP/1.1\r\nHost: h\r\n\r\n", &custom), + Verdict::Allow { .. } + )); + } + + #[test] + fn required_user_agent_enforced() { + let mut custom = policy(); + custom.require_user_agent = true; + let head = b"GET / HTTP/1.1\r\nHost: h\r\n\r\n"; + assert!(matches!(evaluate(head, &custom), Verdict::Reject { .. })); + let head_with_ua = b"GET / HTTP/1.1\r\nHost: h\r\nUser-Agent: curl/8\r\n\r\n"; + assert!(matches!(evaluate(head_with_ua, &custom), Verdict::Allow { .. })); + } +} diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs index cd170e0..1dc82f5 100644 --- a/src/protocol/mod.rs +++ b/src/protocol/mod.rs @@ -1,3 +1,6 @@ +#[cfg(feature = "protocol-http")] +pub mod http; + use std::future::Future; use std::pin::Pin; use tokio::net::TcpStream; diff --git a/src/xdp/diagnostics.rs b/src/xdp/diagnostics.rs index 8c15f89..bd06c46 100644 --- a/src/xdp/diagnostics.rs +++ b/src/xdp/diagnostics.rs @@ -262,6 +262,7 @@ impl EnvironmentReport { /// /// # Errors /// См. [`EnvironmentReport::validate`]. +#[cfg(feature = "xdp")] pub(crate) fn preflight(interface: &str) -> Result<()> { let report = EnvironmentReport::collect(&FilesystemProbe, interface); let kernel = report diff --git a/src/xdp/filter.rs b/src/xdp/filter.rs index 08aa140..fcaff5e 100644 --- a/src/xdp/filter.rs +++ b/src/xdp/filter.rs @@ -1,10 +1,11 @@ use anyhow::{Context, Result, bail}; -use libbpf_rs::{MapCore, MapFlags, Object, ObjectBuilder, RingBuffer, RingBufferBuilder, Xdp, XdpFlags}; +use libbpf_rs::{MapCore, MapFlags, Object, ObjectBuilder, OpenObject, RingBuffer, RingBufferBuilder, Xdp, XdpFlags}; use std::net::IpAddr; use std::net::Ipv4Addr; use std::os::unix::io::AsFd; use super::XdpStats; +use super::globals::{RODATA_MAP_NAME, XdpGlobals}; use crate::traffic::prefix::{PrefixKey, PrefixStatsVal}; pub struct XdpFilter { @@ -12,6 +13,7 @@ pub struct XdpFilter { ringbuf: Option>, ifindex: i32, interface: String, + globals: XdpGlobals, } impl XdpFilter { @@ -21,19 +23,25 @@ impl XdpFilter { ringbuf: None, ifindex: 0, interface: interface.to_string(), + globals: XdpGlobals::default(), } } + /// Задаёт глобалы для патчинга `.rodata` при следующем `load()`. + pub fn set_globals(&mut self, globals: XdpGlobals) { + self.globals = globals; + } + pub fn load(&mut self) -> Result<()> { // Диагностика ДО загрузки: fail-fast на неподдерживаемом ядре/драйвере. super::diagnostics::preflight(&self.interface)?; let bpf_obj = include_bytes!(concat!(env!("OUT_DIR"), "/universal_filter.o")); - let obj = ObjectBuilder::default() + let mut open_obj = ObjectBuilder::default() .open_memory(bpf_obj) - .context("Failed to open XDP object")? - .load() - .context("Failed to load XDP object (verifier error?)")?; + .context("Failed to open XDP object")?; + patch_rodata(&mut open_obj, &self.globals)?; + let obj = open_obj.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 { @@ -182,6 +190,17 @@ fn unix_ns() -> u64 { .as_nanos() as u64 } +/// Патчит карту `rampart_.rodata` (volatile const глобалы) ДО загрузки объекта. +fn patch_rodata(open_obj: &mut OpenObject, globals: &XdpGlobals) -> Result<()> { + let image = globals.build_rodata_image(); + let mut map = open_obj + .maps_mut() + .find(|m| m.name() == RODATA_MAP_NAME) + .with_context(|| format!("map '{RODATA_MAP_NAME}' not found in XDP object"))?; + map.set_initial_value(&image) + .with_context(|| format!("failed to set initial value of '{RODATA_MAP_NAME}'")) +} + fn build_ringbuf(obj: &Object) -> Result> { let map = obj .maps() diff --git a/src/xdp/globals.rs b/src/xdp/globals.rs new file mode 100644 index 0000000..d1db0d9 --- /dev/null +++ b/src/xdp/globals.rs @@ -0,0 +1,182 @@ +//! Патчинг volatile const глобалов XDP-программы (`xdp/core/config.h`) из TOML-конфига. +//! +//! Глобалы живут в секции `.rodata`, которая при загрузке превращается в карту +//! с именем `rampart_.rodata`. Образ патчится до `load()` через +//! `OpenMapMut::set_initial_value()` (см. [`crate::xdp::filter`]). +//! +//! # Схема маппинга `[xdp]` → G_* (единственный источник истины) +//! +//! | Конфиг | Глобал | Тип | +//! |-------------------------|------------------------------|-----| +//! | `protected_port_start` | `G_START_PORT` | u16 | +//! | `protected_port_end` | `G_END_PORT` | u16 | +//! | `udp_policy` | `G_UDP_POLICY` | u8 ("pass"=0, "drop"=1, "rate-limit"=2) | +//! | `udp_rate_hit_count` | `G_UDP_HIT_COUNT` | u32 | +//! | `udp_rate_window_ms` | `G_UDP_WINDOW_NS` | u64 (мс → нс) | +//! | `throttle_enabled` | `G_FEATURE_SYN_THROTTLE` | u8 (bool → 0/1) | +//! | `events_enabled` | `G_FEATURE_EVENTS` | u8 (bool → 0/1) | +//! | `syn_challenge_enabled` | `G_SYN_CHALLENGE_ENABLED` | u8 (bool → 0/1) | +//! | `challenge_secret_hex` | `G_CHALLENGE_SECRET` | u64 (hex, big-endian) | +//! | `challenge_timeout_ms` | `G_CHALLENGE_TIMEOUT_MS` | u32 | +//! +//! Остальные глобалы (`G_SYN_*`, `G_CONNTRACK_IDLE_NS`, `G_PLAYER_IDLE_NS`, +//! `G_BAN_DURATION_NS`, `G_MAX_OUT_OF_ORDER`, `G_CHALLENGE_VERIFIED_TTL_NS`) +//! конфигом пока не управляются и патчатся значениями по умолчанию из config.h. + +use anyhow::{Context, Result, bail}; + +use crate::config::XdpConfig; + +/// Имя карты с образом `.rodata` загруженного BPF-объекта. +pub(crate) const RODATA_MAP_NAME: &str = "rampart_.rodata"; + +/// Размер образа `.rodata` в байтах (см. `build_rodata_image`). +pub const RODATA_SIZE: usize = 96; + +/// Значение `G_UDP_POLICY`: пропускать UDP. +pub const UDP_POLICY_PASS: u8 = 0; +/// Значение `G_UDP_POLICY`: дропать весь UDP. +pub const UDP_POLICY_DROP: u8 = 1; +/// Значение `G_UDP_POLICY`: rate-limit по окну/порогу. +pub const UDP_POLICY_RATE_LIMIT: u8 = 2; + +/// Дефолт-плейсхолдер `G_CHALLENGE_SECRET` из config.h. +const DEFAULT_CHALLENGE_SECRET: u64 = 0xA5A5_A5A5_A5A5_A5A5; + +/// Runtime-значения глобалов XDP-программы; `Default` совпадает с config.h. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct XdpGlobals { + pub start_port: u16, + pub end_port: u16, + pub syn_hit_count: u32, + pub syn_window_ns: u64, + pub syn_ban_duration_ns: u64, + pub conntrack_idle_ns: u64, + pub player_idle_ns: u64, + pub ban_duration_ns: u64, + pub max_out_of_order: u8, + pub udp_policy: u8, + pub udp_hit_count: u32, + pub udp_window_ns: u64, + pub feature_syn_throttle: bool, + pub feature_events: bool, + pub syn_challenge_enabled: bool, + pub challenge_secret: u64, + pub challenge_timeout_ms: u32, + pub challenge_verified_ttl_ns: u64, +} + +impl Default for XdpGlobals { + fn default() -> Self { + Self { + start_port: 1, + end_port: 65535, + syn_hit_count: 10, + syn_window_ns: 3_000_000_000, + syn_ban_duration_ns: 60_000_000_000, + conntrack_idle_ns: 30_000_000_000, + player_idle_ns: 120_000_000_000, + ban_duration_ns: 300_000_000_000, + max_out_of_order: 4, + udp_policy: UDP_POLICY_PASS, + udp_hit_count: 100, + udp_window_ns: 1_000_000_000, + feature_syn_throttle: true, + feature_events: true, + syn_challenge_enabled: false, + challenge_secret: DEFAULT_CHALLENGE_SECRET, + challenge_timeout_ms: 3000, + challenge_verified_ttl_ns: 300_000_000_000, + } + } +} + +impl XdpGlobals { + /// Собирает байтовый образ `.rodata` в порядке объявления глобалов + /// в `xdp/core/config.h` с естественным выравниванием каждого поля. + /// + /// RISK: раскладка предполагает, что линкер кладёт статические переменные + /// в секцию `.rodata` в порядке объявления с natural alignment — на + /// практике это так (каждая `static volatile const` получает своё + /// выравнивание, padding виден в смещениях ниже), но стандарт C этого не + /// гарантирует. TODO: верифицировать смещения на сервере через BTF-dump + /// карты `rampart_.rodata` (`bpftool btf dump`) и зафиксировать их здесь. + pub fn build_rodata_image(&self) -> Vec { + let mut img = vec![0u8; RODATA_SIZE]; + // u16 ×2 без padding: смещения 0..4. + img[0..2].copy_from_slice(&self.start_port.to_le_bytes()); + img[2..4].copy_from_slice(&self.end_port.to_le_bytes()); + // u32, align 4: 4..8. + img[4..8].copy_from_slice(&self.syn_hit_count.to_le_bytes()); + // u64, align 8: 8..48 (SYN throttle + idle timeouts + ban duration). + img[8..16].copy_from_slice(&self.syn_window_ns.to_le_bytes()); + img[16..24].copy_from_slice(&self.syn_ban_duration_ns.to_le_bytes()); + img[24..32].copy_from_slice(&self.conntrack_idle_ns.to_le_bytes()); + img[32..40].copy_from_slice(&self.player_idle_ns.to_le_bytes()); + img[40..48].copy_from_slice(&self.ban_duration_ns.to_le_bytes()); + // u8 ×3 подряд: 48, 49, 50 (max_out_of_order, udp_policy, —). + img[48] = self.max_out_of_order; + img[49] = self.udp_policy; + // u32 после u8: padding 50..52, значение 52..56. + img[52..56].copy_from_slice(&self.udp_hit_count.to_le_bytes()); + // u64, align 8: 56..64. + img[56..64].copy_from_slice(&self.udp_window_ns.to_le_bytes()); + // u8 ×3 (feature flags + challenge flag): 64, 65, 66. + img[64] = u8::from(self.feature_syn_throttle); + img[65] = u8::from(self.feature_events); + img[66] = u8::from(self.syn_challenge_enabled); + // u64 после u8: padding 67..72, значение 72..80. + img[72..80].copy_from_slice(&self.challenge_secret.to_le_bytes()); + // u32: 80..84. + img[80..84].copy_from_slice(&self.challenge_timeout_ms.to_le_bytes()); + // u64 после u32: padding 84..88, значение 88..96. + img[88..96].copy_from_slice(&self.challenge_verified_ttl_ns.to_le_bytes()); + img + } + + /// Строит глобалы из секции `[xdp]` конфига. + /// + /// # Errors + /// Неизвестное значение `udp_policy`; некорректный `challenge_secret_hex` + /// (не hex или не 8 байт); `protected_port_end < protected_port_start`. + pub fn from_config(cfg: &XdpConfig) -> Result { + let mut g = Self { + start_port: cfg.protected_port_start, + end_port: cfg.protected_port_end, + udp_hit_count: cfg.udp_rate_hit_count, + udp_window_ns: cfg.udp_rate_window_ms.saturating_mul(1_000_000), + feature_syn_throttle: cfg.throttle_enabled, + feature_events: cfg.events_enabled, + syn_challenge_enabled: cfg.syn_challenge_enabled, + challenge_timeout_ms: cfg.challenge_timeout_ms, + ..Self::default() + }; + if g.end_port < g.start_port { + bail!( + "xdp.protected_port_end ({}) < xdp.protected_port_start ({})", + g.end_port, + g.start_port + ); + } + g.udp_policy = match cfg.udp_policy.as_str() { + "pass" => UDP_POLICY_PASS, + "drop" => UDP_POLICY_DROP, + "rate-limit" => UDP_POLICY_RATE_LIMIT, + other => bail!("invalid xdp.udp_policy '{other}' (expected pass|drop|rate-limit)"), + }; + if let Some(secret) = cfg.challenge_secret_hex.as_deref() { + g.challenge_secret = parse_secret_hex(secret)?; + } + Ok(g) + } +} + +/// Парсит 128-битный hex (16 символов, опциональный префикс `0x`) в u64 (big-endian). +fn parse_secret_hex(s: &str) -> Result { + let cleaned = s.trim().trim_start_matches("0x").replace(['_', ':', '-'], ""); + let bytes = hex::decode(&cleaned).with_context(|| format!("invalid xdp.challenge_secret_hex '{s}'"))?; + let arr: [u8; 8] = bytes + .try_into() + .map_err(|v: Vec| anyhow::anyhow!("xdp.challenge_secret_hex must be exactly 8 bytes, got {}", v.len()))?; + Ok(u64::from_be_bytes(arr)) +} diff --git a/src/xdp/mod.rs b/src/xdp/mod.rs index 7d6ea98..0afe229 100644 --- a/src/xdp/mod.rs +++ b/src/xdp/mod.rs @@ -7,6 +7,11 @@ pub use diagnostics::{ AttachMode, EnvironmentReport, FilesystemProbe, KernelVersion, MIN_KERNEL, SystemProbe, driver_supports_native_xdp, }; +#[cfg(feature = "xdp")] +mod globals; +#[cfg(feature = "xdp")] +pub use globals::XdpGlobals; + #[cfg(feature = "xdp")] mod filter; #[cfg(feature = "xdp")] diff --git a/tests/config_parse.rs b/tests/config_parse.rs index df77bb9..c7bc16f 100644 --- a/tests/config_parse.rs +++ b/tests/config_parse.rs @@ -89,3 +89,54 @@ webhook_url = "https://hooks.example.test/rampart" Some("https://hooks.example.test/rampart") ); } + +#[test] +fn xdp_section_defaults() { + let config = Config::parse_str("").expect("empty config must parse"); + let xdp = &config.xdp; + assert_eq!(xdp.protected_port_start, 1); + assert_eq!(xdp.protected_port_end, 65535); + assert_eq!(xdp.udp_policy, "pass"); + assert_eq!(xdp.udp_rate_hit_count, 100); + assert_eq!(xdp.udp_rate_window_ms, 1000); + assert!(!xdp.syn_challenge_enabled); + assert!(xdp.challenge_secret_hex.is_none()); + assert_eq!(xdp.challenge_timeout_ms, 3000); + assert!(xdp.throttle_enabled); + assert!(xdp.events_enabled); +} + +#[test] +fn parses_xdp_section() { + let config = Config::parse_str( + r#" +[xdp] +enabled = true +interface = "ens3" +protected_port_start = 25560 +protected_port_end = 25600 +udp_policy = "drop" +udp_rate_hit_count = 250 +udp_rate_window_ms = 500 +syn_challenge_enabled = true +challenge_secret_hex = "dead_beef_dead_beef" +challenge_timeout_ms = 1500 +throttle_enabled = false +events_enabled = false +"#, + ) + .expect("[xdp] section must parse"); + let xdp = &config.xdp; + assert!(xdp.enabled); + assert_eq!(xdp.interface, "ens3"); + assert_eq!(xdp.protected_port_start, 25560); + assert_eq!(xdp.protected_port_end, 25600); + assert_eq!(xdp.udp_policy, "drop"); + assert_eq!(xdp.udp_rate_hit_count, 250); + assert_eq!(xdp.udp_rate_window_ms, 500); + assert!(xdp.syn_challenge_enabled); + assert_eq!(xdp.challenge_secret_hex.as_deref(), Some("dead_beef_dead_beef")); + assert_eq!(xdp.challenge_timeout_ms, 1500); + assert!(!xdp.throttle_enabled); + assert!(!xdp.events_enabled); +} diff --git a/tests/http_protocol.rs b/tests/http_protocol.rs new file mode 100644 index 0000000..c7f2b2b --- /dev/null +++ b/tests/http_protocol.rs @@ -0,0 +1,131 @@ +#![cfg(feature = "protocol-http")] + +use rampart::config::HttpProtocolConfig; +use rampart::protocol::http::HttpProtocolHandler; +use rampart::protocol::{ProtocolHandler, ProtocolRegistry}; +use tokio::io::AsyncWriteExt; +use tokio::net::{TcpListener, TcpStream}; + +const REQUEST: &[u8] = b"GET /index.html HTTP/1.1\r\nHost: example.com\r\n\r\n"; + +fn handler() -> HttpProtocolHandler { + HttpProtocolHandler::new(&HttpProtocolConfig::default(), &["127.0.0.1:9000".to_string()]) +} + +async fn socket_pair() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let addr = listener.local_addr().expect("local addr"); + let client = TcpStream::connect(addr).await.expect("connect"); + let (server, _) = listener.accept().await.expect("accept"); + (client, server) +} + +#[tokio::test] +async fn valid_request_is_allowed() { + let (mut client, mut server) = socket_pair().await; + client.write_all(REQUEST).await.expect("write"); + assert!(handler().handshake(&mut server).await.is_ok()); +} + +#[tokio::test] +async fn fragmented_head_is_accumulated_and_parsed() { + let (mut client, mut server) = socket_pair().await; + for chunk in REQUEST.chunks(7) { + client.write_all(chunk).await.expect("write chunk"); + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + assert!(handler().handshake(&mut server).await.is_ok()); +} + +#[tokio::test] +async fn oversized_head_is_rejected() { + let config = HttpProtocolConfig { + max_header_bytes: 32, + ..HttpProtocolConfig::default() + }; + let (mut client, mut server) = socket_pair().await; + client + .write_all(b"GET / HTTP/1.1\r\nHost: a-very-long-host-name-value\r\n") + .await + .expect("write"); + let err = handler_with(&config) + .handshake(&mut server) + .await + .expect_err("oversized head must be rejected"); + assert!(err.to_string().contains("exceeds")); +} + +#[tokio::test] +async fn unknown_method_is_rejected() { + let (mut client, mut server) = socket_pair().await; + client + .write_all(b"TRACE / HTTP/1.1\r\nHost: h\r\n\r\n") + .await + .expect("write"); + let err = handler() + .handshake(&mut server) + .await + .expect_err("unknown method must be rejected"); + assert!(err.to_string().contains("method")); +} + +#[tokio::test] +async fn missing_host_is_rejected() { + let (mut client, mut server) = socket_pair().await; + client.write_all(b"GET / HTTP/1.1\r\n\r\n").await.expect("write"); + let err = handler() + .handshake(&mut server) + .await + .expect_err("missing host must be rejected"); + assert!(err.to_string().contains("host")); +} + +#[tokio::test] +async fn blocked_path_is_rejected() { + let config = HttpProtocolConfig { + blocked_paths: vec!["/admin".to_string()], + ..HttpProtocolConfig::default() + }; + let (mut client, mut server) = socket_pair().await; + client + .write_all(b"GET /admin/secrets HTTP/1.1\r\nHost: h\r\n\r\n") + .await + .expect("write"); + let err = handler_with(&config) + .handshake(&mut server) + .await + .expect_err("blocked path must be rejected"); + assert!(err.to_string().contains("blocklisted")); +} + +#[tokio::test] +async fn required_user_agent_missing_is_rejected() { + let config = HttpProtocolConfig { + require_user_agent: true, + ..HttpProtocolConfig::default() + }; + let (mut client, mut server) = socket_pair().await; + client + .write_all(b"GET / HTTP/1.1\r\nHost: h\r\n\r\n") + .await + .expect("write"); + let err = handler_with(&config) + .handshake(&mut server) + .await + .expect_err("missing user-agent must be rejected"); + assert!(err.to_string().contains("user-agent")); +} + +#[tokio::test] +async fn registry_resolves_http_handler_when_feature_enabled() { + let mut registry = ProtocolRegistry::new(); + registry.register(Box::new(handler())); + assert_eq!(registry.names(), vec!["http"]); + let resolved = registry.get("http").expect("http handler must resolve"); + assert_eq!(resolved.name(), "http"); + assert!(registry.primary().is_ok()); +} + +fn handler_with(config: &HttpProtocolConfig) -> HttpProtocolHandler { + HttpProtocolHandler::new(config, &["127.0.0.1:9000".to_string()]) +} diff --git a/tests/xdp_globals.rs b/tests/xdp_globals.rs new file mode 100644 index 0000000..b8be66f --- /dev/null +++ b/tests/xdp_globals.rs @@ -0,0 +1,118 @@ +#![cfg(feature = "xdp")] + +use rampart::config::Config; +use rampart::xdp::XdpGlobals; + +#[test] +fn rodata_image_size_is_stable() { + let img = XdpGlobals::default().build_rodata_image(); + assert_eq!(img.len(), 96, ".rodata image size must match config.h layout"); +} + +#[test] +fn rodata_defaults_match_config_h() { + let img = XdpGlobals::default().build_rodata_image(); + let rd = |off: usize| u64::from_le_bytes(img[off..off + 8].try_into().expect("8 bytes")); + let rd32 = |off: usize| u32::from_le_bytes(img[off..off + 4].try_into().expect("4 bytes")); + assert_eq!(&img[0..2], 1u16.to_le_bytes()); + assert_eq!(&img[2..4], 65535u16.to_le_bytes()); + assert_eq!(rd32(4), 10); // G_SYN_HIT_COUNT + assert_eq!(rd(8), 3_000_000_000); // G_SYN_WINDOW_NS + assert_eq!(rd(16), 60_000_000_000); // G_SYN_BAN_DURATION_NS + assert_eq!(rd(24), 30_000_000_000); // G_CONNTRACK_IDLE_NS + assert_eq!(rd(32), 120_000_000_000); // G_PLAYER_IDLE_NS + assert_eq!(rd(40), 300_000_000_000); // G_BAN_DURATION_NS + assert_eq!(img[48], 4); // G_MAX_OUT_OF_ORDER + assert_eq!(img[49], 0); // G_UDP_POLICY + assert_eq!(rd32(52), 100); // G_UDP_HIT_COUNT (padding 50..52) + assert_eq!(rd(56), 1_000_000_000); // G_UDP_WINDOW_NS + assert_eq!(img[64], 1); // G_FEATURE_SYN_THROTTLE + assert_eq!(img[65], 1); // G_FEATURE_EVENTS + assert_eq!(img[66], 0); // G_SYN_CHALLENGE_ENABLED + assert_eq!(rd(72), 0xA5A5_A5A5_A5A5_A5A5); // G_CHALLENGE_SECRET (padding 67..72) + assert_eq!(rd32(80), 3000); // G_CHALLENGE_TIMEOUT_MS + assert_eq!(rd(88), 300_000_000_000); // G_CHALLENGE_VERIFIED_TTL_NS (padding 84..88) +} + +#[test] +fn rodata_padding_bytes_stay_zero() { + let g = XdpGlobals { + max_out_of_order: u8::MAX, + udp_policy: u8::MAX, + feature_syn_throttle: true, + feature_events: true, + syn_challenge_enabled: true, + ..XdpGlobals::default() + }; + let img = g.build_rodata_image(); + for off in [50, 51, 67, 68, 69, 70, 71, 84, 85, 86, 87] { + assert_eq!(img[off], 0, "alignment padding at {off} must stay zeroed"); + } +} + +#[test] +fn from_config_maps_fields() { + let cfg = Config::parse_str( + r#" +[xdp] +protected_port_start = 25560 +protected_port_end = 25600 +udp_policy = "rate-limit" +udp_rate_hit_count = 500 +udp_rate_window_ms = 2000 +syn_challenge_enabled = true +challenge_secret_hex = "0011223344556677" +challenge_timeout_ms = 7000 +throttle_enabled = false +events_enabled = false +"#, + ) + .expect("[xdp] section must parse"); + let g = XdpGlobals::from_config(&cfg.xdp).expect("globals must build"); + assert_eq!(g.start_port, 25560); + assert_eq!(g.end_port, 25600); + assert_eq!(g.udp_policy, 2); + assert_eq!(g.udp_hit_count, 500); + assert_eq!(g.udp_window_ns, 2_000_000_000); + assert!(!g.feature_syn_throttle); + assert!(!g.feature_events); + assert!(g.syn_challenge_enabled); + assert_eq!(g.challenge_secret, 0x0011_2233_4455_6677); + assert_eq!(g.challenge_timeout_ms, 7000); + + let img = g.build_rodata_image(); + assert_eq!(img[49], 2); + assert_eq!(img[64], 0); + assert_eq!(&img[0..2], 25560u16.to_le_bytes()); +} + +#[test] +fn from_config_rejects_bad_udp_policy() { + let cfg = Config::parse_str("[xdp]\nudp_policy = \"flood\"\n").expect("config parses"); + assert!(XdpGlobals::from_config(&cfg.xdp).is_err()); +} + +#[test] +fn from_config_rejects_inverted_port_range() { + let cfg = Config::parse_str("[xdp]\nprotected_port_start = 100\nprotected_port_end = 50\n").expect("config parses"); + assert!(XdpGlobals::from_config(&cfg.xdp).is_err()); +} + +#[test] +fn from_config_rejects_bad_secret_hex() { + for secret in ["nothex", "0011", "001122334455667788"] { + let toml = format!("[xdp]\nchallenge_secret_hex = \"{secret}\"\n"); + let cfg = Config::parse_str(&toml).expect("config parses"); + assert!( + XdpGlobals::from_config(&cfg.xdp).is_err(), + "secret '{secret}' must be rejected" + ); + } +} + +#[test] +fn xdp_section_defaults_match_config_h() { + let cfg = Config::parse_str("").expect("empty config parses"); + let g = XdpGlobals::from_config(&cfg.xdp).expect("globals must build"); + assert_eq!(g, XdpGlobals::default()); +}