diff --git a/README.md b/README.md index dddfeec..38fe2c6 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,7 @@ players / masters ──> front :27960 (etrelay) ══wg══> :27960 ``` - **Proxy**: one upstream socket per client address, packets forwarded as-is both ways, session dropped after 2 min of silence. +- **Filter**: connectionless packets pass only for what clients and masters send a server (`getstatus`, `getinfo`, `getchallenge`, `connect`, PunkBuster's `PB_`). `rcon` never reaches the server from the internet: it travels in clear, and a stream of it blocks the local admin tool (2.60b runs one rcon per 500 ms, whoever sends it). `getstatus`/`getinfo` are limited per source IP (burst of 5, then 1/s): their answers are much bigger than the query, a spoofed source would make the server a DDoS reflector. At most 2048 sessions (one socket each). - **Heartbeat**: every 5 min, `getinfo` to the ET server; if it answers, `heartbeat EnemyTerritory-1` to the masters **from the public socket**. When it stops answering, one `heartbeat ETFlatline-1`. Masters check (`getChallenge`, `getInfo`) the address the heartbeat came from: the public `:27960`. Those packets go through the relay like any client, the ET server answers them, and the master lists the front public address. diff --git a/etrelay.service b/etrelay.service index b2ad1a8..9bf2c88 100644 --- a/etrelay.service +++ b/etrelay.service @@ -7,6 +7,8 @@ Wants=network-online.target ExecStart=/usr/local/bin/etrelay @UPSTREAM@ @ARGS@ Restart=always RestartSec=2 +# One socket per client session, at most 2048 (MAX_SESSIONS): room above it. +LimitNOFILE=4096 # Port 27960 > 1024: no privilege needed. DynamicUser=yes diff --git a/src/main.rs b/src/main.rs index 7baba71..2d47ae5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,7 +8,7 @@ use anyhow::{Context, Result}; use clap::Parser; use std::collections::HashMap; -use std::net::SocketAddr; +use std::net::{IpAddr, SocketAddr}; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; @@ -29,6 +29,26 @@ const SESSION_QUEUE: usize = 64; const MAX_PACKET: usize = 65_535; /// Longest connectionless command name shown in debug logs. const MAX_COMMAND: usize = 32; +/// Connectionless commands clients and masters send a server. Anything else is dropped, +/// rcon above all: only etadm uses it, from the server itself, and a stream of rcon +/// packets would block it (2.60b runs one rcon per 500 ms, whoever sends it). +const ALLOWED: [&str; 4] = ["getstatus", "getinfo", "getchallenge", "connect"]; +/// PunkBuster's connectionless packets, matched as is by the server (no tokenizing). +const PUNKBUSTER: &[u8] = b"PB_"; +/// Queries answered with a much bigger packet: from a spoofed source they would make the +/// server flood a victim (UDP amplification). Limited per source IP. +const QUERIES: [&str; 2] = ["getstatus", "getinfo"]; +/// A burst (a browser refresh, a master's challenge), then one query a second. +const QUERY_BURST: f64 = 5.0; +const QUERY_RATE: f64 = 1.0; +/// Source IPs the limiter remembers: past this, queries from new IPs are dropped, so a +/// flood of spoofed sources can't grow memory. +const MAX_QUERY_SOURCES: usize = 65_536; +/// Idle IPs are forgotten after this (their bucket is full again long before). +const QUERY_FORGET: Duration = Duration::from_secs(60); +/// One upstream socket per session: spoofed sources must not use up the file +/// descriptors (etrelay.service raises LimitNOFILE above it). +const MAX_SESSIONS: usize = 2_048; static DEBUG: AtomicBool = AtomicBool::new(false); @@ -88,10 +108,17 @@ async fn main() -> Result<()> { /// Dispatches client packets to their session, opening one on first packet. async fn relay(public: Arc, upstream: SocketAddr) -> Result<()> { let mut sessions: HashMap>> = HashMap::new(); + let mut filter = Filter::new(Instant::now()); let mut buf = vec![0u8; MAX_PACKET]; loop { let (n, client) = public.recv_from(&mut buf).await.context("recv")?; + + if let Some(reason) = filter.refuse(&buf[..n], client.ip(), Instant::now()) { + debug!("{client} dropped: {reason}"); + continue; + } + let mut packet = buf[..n].to_vec(); if let Some(tx) = sessions.get(&client) { @@ -105,6 +132,11 @@ async fn relay(public: Arc, upstream: SocketAddr) -> Result<()> { sessions.retain(|_, tx| !tx.is_closed()); + if sessions.len() >= MAX_SESSIONS { + debug!("{client} dropped: {MAX_SESSIONS} sessions open"); + continue; + } + match open_session(public.clone(), upstream, client).await { Ok(tx) => { let _ = tx.try_send(packet); @@ -226,6 +258,93 @@ async fn send_to_host(sock: &UdpSocket, packet: &[u8], host: &str) -> Result, + forgotten: Instant, +} + +/// Token bucket of one source IP's queries. +struct Bucket { + tokens: f64, + at: Instant, +} + +impl Filter { + fn new(now: Instant) -> Self { + Self { + queries: HashMap::new(), + forgotten: now, + } + } + + /// Why the packet is dropped, None to forward it. In-game packets always pass. + fn refuse(&mut self, packet: &[u8], ip: IpAddr, now: Instant) -> Option<&'static str> { + let body = packet.strip_prefix(OOB)?; + if body.starts_with(PUNKBUSTER) { + return None; + } + + let name = first_token(body); + if !ALLOWED + .iter() + .any(|c| c.as_bytes().eq_ignore_ascii_case(name)) + { + return Some("connectionless command not allowed"); + } + + if QUERIES + .iter() + .any(|q| q.as_bytes().eq_ignore_ascii_case(name)) + && !self.query(ip, now) + { + return Some("query rate"); + } + + None + } + + /// Takes a token from ip's bucket. + fn query(&mut self, ip: IpAddr, now: Instant) -> bool { + if now.duration_since(self.forgotten) >= QUERY_FORGET { + self.queries + .retain(|_, b| now.duration_since(b.at) < QUERY_FORGET); + self.forgotten = now; + } + + if !self.queries.contains_key(&ip) && self.queries.len() >= MAX_QUERY_SOURCES { + return false; + } + + let bucket = self.queries.entry(ip).or_insert(Bucket { + tokens: QUERY_BURST, + at: now, + }); + let refill = now.duration_since(bucket.at).as_secs_f64() * QUERY_RATE; + bucket.tokens = (bucket.tokens + refill).min(QUERY_BURST); + bucket.at = now; + + if bucket.tokens < 1.0 { + return false; + } + + bucket.tokens -= 1.0; + true + } +} + +/// First word of a connectionless packet as the server's tokenizer reads it: leading +/// bytes up to space skipped (control characters included). Only compared against +/// ALLOWED: a word the tokenizer would read differently (comments, quotes) matches +/// nothing and is dropped. +fn first_token(body: &[u8]) -> &[u8] { + let start = body.iter().position(|b| *b > b' ').unwrap_or(body.len()); + let body = &body[start..]; + let end = body.iter().position(|b| *b <= b' ').unwrap_or(body.len()); + + &body[..end] +} + /// Command name of a connectionless packet (`getinfo`, `infoResponse`, `connect`...), /// None for in-game packets. Arguments are left out: `connect` carries the userinfo. fn command(packet: &[u8]) -> Option { @@ -251,3 +370,78 @@ async fn connect(upstream: SocketAddr) -> Result { sock.connect(upstream).await?; Ok(sock) } + +#[cfg(test)] +mod tests { + use super::*; + + const IP: IpAddr = IpAddr::V4(std::net::Ipv4Addr::new(203, 0, 113, 7)); + + fn oob(body: &[u8]) -> Vec { + [OOB, body].concat() + } + + #[test] + fn forwards_game_traffic() { + let mut f = Filter::new(Instant::now()); + let now = Instant::now(); + + for body in [ + &b"getchallenge"[..], + b"connect \x12\x34", + b"getinfo xxx", + b"PB_whatever", + ] { + assert_eq!(f.refuse(&oob(body), IP, now), None, "{body:?}"); + } + + // In-game packets (no OOB prefix). + assert_eq!(f.refuse(b"\x01\x02\x03", IP, now), None); + } + + #[test] + fn drops_rcon_however_written() { + let mut f = Filter::new(Instant::now()); + let now = Instant::now(); + + for body in [ + &b"rcon pass status"[..], + b"RCON pass status", + b" \t rcon pass status", + b"\x01rcon pass status", + b"/**/rcon pass status", + b"\"rcon\" pass status", + b"", + ] { + assert!(f.refuse(&oob(body), IP, now).is_some(), "{body:?}"); + } + } + + #[test] + fn limits_queries_per_ip() { + let start = Instant::now(); + let mut f = Filter::new(start); + + let passed = (0..20) + .filter(|_| f.refuse(&oob(b"getstatus"), IP, start).is_none()) + .count(); + assert_eq!(passed, QUERY_BURST as usize); + + // Another IP has its own bucket; connect isn't a query. + let other = IpAddr::V4(std::net::Ipv4Addr::new(198, 51, 100, 1)); + assert_eq!(f.refuse(&oob(b"getinfo"), other, start), None); + assert_eq!(f.refuse(&oob(b"connect x"), IP, start), None); + + // One token a second. + let later = start + Duration::from_secs(1); + assert_eq!(f.refuse(&oob(b"getinfo"), IP, later), None); + assert!(f.refuse(&oob(b"getinfo"), IP, later).is_some()); + + // Forgotten after QUERY_FORGET: full again. + let much_later = start + QUERY_FORGET + Duration::from_secs(1); + let passed = (0..20) + .filter(|_| f.refuse(&oob(b"getinfo"), IP, much_later).is_none()) + .count(); + assert_eq!(passed, QUERY_BURST as usize); + } +}