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)
This commit is contained in:
loki5512344 2026-08-24 12:18:22 +02:00
parent 8b35ac693c
commit 6863249ad4
Signed by: boba
GPG key ID: 253067914055423B
14 changed files with 972 additions and 22 deletions

View file

@ -37,8 +37,8 @@ store-redis = ["dep:redis"]
geoip = ["dep:maxminddb"] geoip = ["dep:maxminddb"]
xdp = ["dep:libbpf-rs", "dep:libc"] xdp = ["dep:libbpf-rs", "dep:libc"]
io-uring = ["dep:tokio-splice"] io-uring = ["dep:tokio-splice"]
# Резерв под реализацию HTTP-протокольного обработчика (модель plugin-by-feature). # HTTP/1.1 protocol handler: парсинг request-head, verdict Allow/Reject,
# Реализаций пока нет: реестр протоколов пуст и edge-нода требует явного флага. # туннелирование на upstream (см. src/protocol/http/ и [protocol.http] в конфиге).
protocol-http = [] protocol-http = []
[dependencies] [dependencies]

View file

@ -14,6 +14,8 @@ use rampart::traffic::detector::{AttackDetector, AttackStatus};
use rampart::traffic::hook::TrafficHook; use rampart::traffic::hook::TrafficHook;
use rampart::traffic::reputation::IpReputation; use rampart::traffic::reputation::IpReputation;
use rampart::xdp::XdpFilter; use rampart::xdp::XdpFilter;
#[cfg(feature = "xdp")]
use rampart::xdp::XdpGlobals;
use std::collections::HashSet; use std::collections::HashSet;
use std::net::IpAddr; use std::net::IpAddr;
use std::sync::Arc; use std::sync::Arc;
@ -33,19 +35,30 @@ fn attack_status_value(status: AttackStatus) -> i64 {
/// Собирает реестр протоколов из скомпилированных реализаций. /// Собирает реестр протоколов из скомпилированных реализаций.
/// ///
/// В ядре реализаций нет: `protocol-http` — зарезервированная пустая feature, /// Обработчики поставляются фичами: `protocol-http` регистрирует
/// показывающая модель plugin-by-feature. Внешние крейты регистрируют свои /// HTTP/1.1-обработчик с политикой из секции `[protocol.http]`.
/// обработчики в этом месте. /// Внешние крейты регистрируют свои обработчики в этом же месте.
#[allow(unused_mut)] fn build_registry(config: &Config) -> ProtocolRegistry {
fn build_registry() -> ProtocolRegistry {
let mut registry = ProtocolRegistry::new();
#[cfg(feature = "protocol-http")] #[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")] #[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] #[tokio::main]
async fn main() -> anyhow::Result<()> { 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()?)) .with_env_filter(EnvFilter::from_default_env().add_directive("rampart=info".parse()?))
.init(); .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() { if registry.is_empty() {
anyhow::bail!( 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" 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 whitelist = build_whitelist(&config)?;
let rate_limiter = Arc::new(RateLimiter::new( let rate_limiter = Arc::new(RateLimiter::new(
@ -239,7 +253,12 @@ fn start_xdp(
let filter = XdpFilter::new(&config.xdp.interface); let filter = XdpFilter::new(&config.xdp.interface);
let shared = Arc::new(Mutex::new(filter)); 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 xdp_metrics = XdpMetrics::register()?;
let sd = shutdown_rx.clone(); let sd = shutdown_rx.clone();

View file

@ -5,6 +5,8 @@ pub use sections::{
BackendConfig, BanConfig, BindConfig, DetectAlertConfig, DetectAutobanConfig, DetectConfig, DetectPrefixConfig, BackendConfig, BanConfig, BindConfig, DetectAlertConfig, DetectAutobanConfig, DetectConfig, DetectPrefixConfig,
LimitsConfig, LoggingConfig, MetricsConfig, PowConfig, StoreConfig, WorkerConfig, XdpConfig, LimitsConfig, LoggingConfig, MetricsConfig, PowConfig, StoreConfig, WorkerConfig, XdpConfig,
}; };
#[cfg(feature = "protocol-http")]
pub use sections::{HttpProtocolConfig, ProtocolConfig};
use serde::Deserialize; use serde::Deserialize;
use std::fs; use std::fs;
@ -33,6 +35,9 @@ pub struct Config {
pub pow: PowConfig, pub pow: PowConfig,
#[serde(default)] #[serde(default)]
pub detect: DetectConfig, pub detect: DetectConfig,
#[cfg(feature = "protocol-http")]
#[serde(default)]
pub protocol: ProtocolConfig,
#[serde(default)] #[serde(default)]
pub whitelist: Vec<String>, pub whitelist: Vec<String>,
} }
@ -75,6 +80,15 @@ impl Config {
if self.detect.prefix.enabled && self.detect.prefix.syn_threshold == 0 { if self.detect.prefix.enabled && self.detect.prefix.syn_threshold == 0 {
anyhow::bail!("detect.prefix.syn_threshold must be positive when enabled"); 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(()) Ok(())
} }
} }

View file

@ -143,6 +143,28 @@ pub struct XdpConfig {
pub enabled: bool, pub enabled: bool,
#[serde(default = "default_xdp_interface")] #[serde(default = "default_xdp_interface")]
pub interface: String, 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<String>,
#[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 { impl Default for XdpConfig {
@ -150,6 +172,16 @@ impl Default for XdpConfig {
Self { Self {
enabled: false, enabled: false,
interface: default_xdp_interface(), 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 { fn default_xdp_interface() -> String {
"eth0".to_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)] #[derive(Debug, Clone, Deserialize)]
pub struct LoggingConfig { pub struct LoggingConfig {
@ -303,3 +359,46 @@ fn default_prefix_window_secs() -> u64 {
fn default_prefix_min_unique_sources() -> u64 { fn default_prefix_min_unique_sources() -> u64 {
16 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<String>,
#[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
}

112
src/protocol/http/mod.rs Normal file
View file

@ -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<Vec<u8>> {
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<Box<dyn Future<Output = anyhow::Result<Upstream>> + 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);
}
}

196
src/protocol/http/parser.rs Normal file
View file

@ -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<String>,
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<usize> {
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 { .. }));
}
}

View file

@ -1,3 +1,6 @@
#[cfg(feature = "protocol-http")]
pub mod http;
use std::future::Future; use std::future::Future;
use std::pin::Pin; use std::pin::Pin;
use tokio::net::TcpStream; use tokio::net::TcpStream;

View file

@ -262,6 +262,7 @@ impl EnvironmentReport {
/// ///
/// # Errors /// # Errors
/// См. [`EnvironmentReport::validate`]. /// См. [`EnvironmentReport::validate`].
#[cfg(feature = "xdp")]
pub(crate) fn preflight(interface: &str) -> Result<()> { pub(crate) fn preflight(interface: &str) -> Result<()> {
let report = EnvironmentReport::collect(&FilesystemProbe, interface); let report = EnvironmentReport::collect(&FilesystemProbe, interface);
let kernel = report let kernel = report

View file

@ -1,10 +1,11 @@
use anyhow::{Context, Result, bail}; 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::IpAddr;
use std::net::Ipv4Addr; use std::net::Ipv4Addr;
use std::os::unix::io::AsFd; use std::os::unix::io::AsFd;
use super::XdpStats; use super::XdpStats;
use super::globals::{RODATA_MAP_NAME, XdpGlobals};
use crate::traffic::prefix::{PrefixKey, PrefixStatsVal}; use crate::traffic::prefix::{PrefixKey, PrefixStatsVal};
pub struct XdpFilter { pub struct XdpFilter {
@ -12,6 +13,7 @@ pub struct XdpFilter {
ringbuf: Option<RingBuffer<'static>>, ringbuf: Option<RingBuffer<'static>>,
ifindex: i32, ifindex: i32,
interface: String, interface: String,
globals: XdpGlobals,
} }
impl XdpFilter { impl XdpFilter {
@ -21,19 +23,25 @@ impl XdpFilter {
ringbuf: None, ringbuf: None,
ifindex: 0, ifindex: 0,
interface: interface.to_string(), 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<()> { pub fn load(&mut self) -> Result<()> {
// Диагностика ДО загрузки: fail-fast на неподдерживаемом ядре/драйвере. // Диагностика ДО загрузки: fail-fast на неподдерживаемом ядре/драйвере.
super::diagnostics::preflight(&self.interface)?; super::diagnostics::preflight(&self.interface)?;
let bpf_obj = include_bytes!(concat!(env!("OUT_DIR"), "/universal_filter.o")); 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) .open_memory(bpf_obj)
.context("Failed to open XDP object")? .context("Failed to open XDP object")?;
.load() patch_rodata(&mut open_obj, &self.globals)?;
.context("Failed to load XDP object (verifier error?)")?; 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) }; let ifindex = unsafe { libc::if_nametoindex(self.interface.as_ptr() as *const libc::c_char) };
if ifindex == 0 { if ifindex == 0 {
@ -182,6 +190,17 @@ fn unix_ns() -> u64 {
.as_nanos() as 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<RingBuffer<'static>> { fn build_ringbuf(obj: &Object) -> Result<RingBuffer<'static>> {
let map = obj let map = obj
.maps() .maps()

182
src/xdp/globals.rs Normal file
View file

@ -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<u8> {
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<Self> {
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<u64> {
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<u8>| anyhow::anyhow!("xdp.challenge_secret_hex must be exactly 8 bytes, got {}", v.len()))?;
Ok(u64::from_be_bytes(arr))
}

View file

@ -7,6 +7,11 @@ pub use diagnostics::{
AttachMode, EnvironmentReport, FilesystemProbe, KernelVersion, MIN_KERNEL, SystemProbe, driver_supports_native_xdp, 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")] #[cfg(feature = "xdp")]
mod filter; mod filter;
#[cfg(feature = "xdp")] #[cfg(feature = "xdp")]

View file

@ -89,3 +89,54 @@ webhook_url = "https://hooks.example.test/rampart"
Some("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);
}

131
tests/http_protocol.rs Normal file
View file

@ -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()])
}

118
tests/xdp_globals.rs Normal file
View file

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