diff --git a/Cargo.lock b/Cargo.lock index 6ae4ebd..e7a2911 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -58,6 +58,68 @@ version = "1.0.104" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "330a5ed07fa54e4702c9d6c4174f74427fc0ef6e214bbd677ae50a5099946470" +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core", + "bytes", + "form_urlencoded", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + [[package]] name = "clap" version = "4.6.7" @@ -106,31 +168,180 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "etrelay" -version = "0.1.0" +version = "0.2.0" dependencies = [ "anyhow", + "axum", "clap", + "serde", + "serde_json", "tokio", ] +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "hyper" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b501faa50e7a26c3d3560ca625132f4078a17771f4810baf70475ae48cbe43" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", + "tower-service", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + [[package]] name = "libc" version = "0.2.189" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "mio" version = "1.2.3" @@ -148,6 +359,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -172,6 +389,90 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "smallvec" +version = "1.16.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9395f0f0eee849a9b707b2f06bb92a6a422090e2123bb2ef8e87a0e61892a8e" + [[package]] name = "socket2" version = "0.6.5" @@ -199,6 +500,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "tokio" version = "1.53.1" @@ -224,6 +531,33 @@ dependencies = [ "syn", ] +[[package]] +name = "tower" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" +dependencies = [ + "futures-core", + "futures-util", + "pin-project-lite", + "sync_wrapper", + "tokio", + "tower-layer", + "tower-service", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + [[package]] name = "unicode-ident" version = "1.0.26" @@ -256,3 +590,9 @@ checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" dependencies = [ "windows-link", ] + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index 9ecd35b..6a4869d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,11 +1,14 @@ [package] name = "etrelay" -version = "0.1.0" +version = "0.2.0" edition = "2024" [dependencies] anyhow = "1.0.100" +axum = { version = "0.8", default-features = false, features = ["http1", "json", "tokio", "query"] } clap = { version = "4.5.51", features = ["derive"] } +serde = { version = "1.0.229", features = ["derive"] } +serde_json = "1.0.151" tokio = { version = "1", features = ["macros", "net", "rt", "sync", "time"] } [profile.release] diff --git a/README.md b/README.md index 38fe2c6..d4d75c4 100644 --- a/README.md +++ b/README.md @@ -17,7 +17,7 @@ How it works in detail, with diagrams: [ENGINE.md](ENGINE.md). ## Usage ```bash -etrelay [--listen 0.0.0.0:27960] [--master host:port]... [--debug] +etrelay [--listen 0.0.0.0:27960] [--master host:port]... [--control ] [--state-dir /var/lib/etrelay] [--debug] ``` Default masters: `etmaster.idsoftware.com:27950` (id Software, the one ET 2.60b clients query) and `etmaster.net:27950` (ET: Legacy community). Passing `--master` replaces both. @@ -28,7 +28,7 @@ Default masters: `etmaster.idsoftware.com:27950` (id Software, the one ET 2.60b probe :27960: up ET server :27960 is up heartbeat EnemyTerritory-1 -> etmaster.net:27950 (198.51.100.10:27950) -session + 198.51.100.10:27950 (2 active) +session + 198.51.100.10:27950 as port 51234 (2 active) 198.51.100.10:27950 -> ET getChallenge (27 B) 198.51.100.10:27950 <- ET challengeResponse (32 B) 198.51.100.10:27950 -> ET getInfo (22 B) @@ -38,6 +38,23 @@ session - :27960: 5321 packets to ET, 4876 from ET As a service: add `--debug` to `ExecStart`, then `systemctl daemon-reload && systemctl restart etrelay`. +Without `--debug`, each player joining is logged with its real address and the port the ET server sees it on (`connect 203.0.113.7:27960 as port 51234`): browser queries aren't. + +## Control API + +`--control ` serves a small HTTP API for the admin tool (etadm), on the tunnel address only (e.g. `10.66.0.1:8090`: unreachable from the internet), with `Authorization: Bearer $ETRELAY_CONTROL_TOKEN` (32 characters at least). Binding is retried every 10 s: the tunnel may come up after etrelay. + +The ET server sees every player as `:`, the port of the player's session socket; the API maps it back to the real address. + +| Method | URL | | +|---|---|---| +| GET | `/sessions` | `[{"client": "203.0.113.7:27960", "port": 51234, "since": }]`: open sessions | +| GET | `/bans` | `[{"ip", "reason", "at", "until"}]` (`until` null: until lifted) | +| PUT | `/bans/{ip}` | `{"reason": "...", "minutes": N}` (`minutes` left out: until lifted); cuts the address's sessions at once | +| DELETE | `/bans/{ip}` | 204, 404 when not banned | + +A banned address's packets are dropped before reaching the server: the player times out, and can't come back. Bans are kept in `/bans.json` (written atomically), expired ones forgotten. + ## Build ```bash diff --git a/etrelay.service b/etrelay.service index 9bf2c88..c04f193 100644 --- a/etrelay.service +++ b/etrelay.service @@ -9,6 +9,10 @@ Restart=always RestartSec=2 # One socket per client session, at most 2048 (MAX_SESSIONS): room above it. LimitNOFILE=4096 +# ETRELAY_CONTROL_TOKEN=... for --control; read by systemd, root only (0600). +EnvironmentFile=-/etc/etrelay/env +# /var/lib/etrelay: the bans file, kept across restarts and upgrades. +StateDirectory=etrelay # Port 27960 > 1024: no privilege needed. DynamicUser=yes diff --git a/src/bans.rs b/src/bans.rs new file mode 100644 index 0000000..f7d45fd --- /dev/null +++ b/src/bans.rs @@ -0,0 +1,164 @@ +//! Banned source IPs: their packets are dropped before reaching the ET server. Saved to a +//! JSON file on each change, so bans outlive restarts. +use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; +use std::collections::HashMap; +use std::net::IpAddr; +use std::path::PathBuf; +use std::time::{SystemTime, UNIX_EPOCH}; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct Ban { + pub reason: String, + /// Unix seconds. + pub at: u64, + /// Unix seconds; None: until lifted. + pub until: Option, +} + +impl Ban { + fn active(&self, now: u64) -> bool { + self.until.is_none_or(|until| now < until) + } +} + +pub struct Bans { + path: PathBuf, + list: HashMap, +} + +impl Bans { + /// Reads path; a missing file is no bans. + pub fn load(path: PathBuf) -> Result { + let list = match std::fs::read(&path) { + Ok(data) => serde_json::from_slice(&data) + .with_context(|| format!("{}: not a bans file", path.display()))?, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => HashMap::new(), + Err(e) => return Err(e).with_context(|| format!("read {}", path.display())), + }; + + Ok(Self { path, list }) + } + + /// Checked for every packet: a map lookup, expired bans are dropped by list and set. + pub fn banned(&self, ip: IpAddr, now: u64) -> bool { + self.list.get(&ip).is_some_and(|b| b.active(now)) + } + + /// Active bans; expired ones are forgotten. + pub fn list(&mut self, now: u64) -> Result> { + self.prune(now)?; + + let mut list: Vec<_> = self.list.iter().map(|(ip, b)| (*ip, b.clone())).collect(); + list.sort_by_key(|(_, b)| b.at); + + Ok(list) + } + + pub fn set(&mut self, ip: IpAddr, ban: Ban) -> Result<()> { + let now = ban.at; + self.list.insert(ip, ban); + self.prune(now)?; + + self.save() + } + + /// False when ip wasn't banned. + pub fn remove(&mut self, ip: IpAddr) -> Result { + if self.list.remove(&ip).is_none() { + return Ok(false); + } + + self.save()?; + Ok(true) + } + + fn prune(&mut self, now: u64) -> Result<()> { + let before = self.list.len(); + self.list.retain(|_, b| b.active(now)); + + if self.list.len() != before { + self.save()?; + } + + Ok(()) + } + + /// Written next to the file then renamed: a crash never leaves half a list. + fn save(&self) -> Result<()> { + let tmp = self.path.with_extension("tmp"); + let data = serde_json::to_vec_pretty(&self.list)?; + + std::fs::write(&tmp, data).with_context(|| format!("write {}", tmp.display()))?; + std::fs::rename(&tmp, &self.path) + .with_context(|| format!("rename to {}", self.path.display())) + } +} + +/// Unix seconds. +pub fn now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn ip(s: &str) -> IpAddr { + s.parse().unwrap() + } + + fn ban(at: u64, until: Option) -> Ban { + Ban { + reason: "cheat".into(), + at, + until, + } + } + + fn temp_path(name: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("etrelay-test-{}-{name}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + dir.join("bans.json") + } + + #[test] + fn bans_expire_and_persist() { + let path = temp_path("persist"); + let mut bans = Bans::load(path.clone()).unwrap(); + + bans.set(ip("203.0.113.7"), ban(100, None)).unwrap(); + bans.set(ip("198.51.100.1"), ban(100, Some(200))).unwrap(); + + assert!(bans.banned(ip("203.0.113.7"), 1_000)); + assert!(bans.banned(ip("198.51.100.1"), 150)); + assert!(!bans.banned(ip("198.51.100.1"), 200)); + assert!(!bans.banned(ip("192.0.2.1"), 150)); + + // Reloaded from the file. + let mut again = Bans::load(path.clone()).unwrap(); + assert_eq!(again.list(150).unwrap().len(), 2); + + // Expired: dropped from the list and the file. + assert_eq!(again.list(300).unwrap().len(), 1); + assert_eq!( + Bans::load(path.clone()).unwrap().list(300).unwrap().len(), + 1 + ); + + assert!(again.remove(ip("203.0.113.7")).unwrap()); + assert!(!again.remove(ip("203.0.113.7")).unwrap()); + assert!(Bans::load(path).unwrap().list(300).unwrap().is_empty()); + } + + #[test] + fn refuses_a_broken_file() { + let path = temp_path("broken"); + std::fs::write(&path, "not json").unwrap(); + + assert!(Bans::load(path).is_err()); + } +} diff --git a/src/control.rs b/src/control.rs new file mode 100644 index 0000000..3d75745 --- /dev/null +++ b/src/control.rs @@ -0,0 +1,163 @@ +//! Control API for etadm: which real address is behind a session port (the ET server +//! sees players as :), and bans. Listens on +//! the tunnel address only, bearer token on top. +use crate::Shared; +use crate::bans::{self, Ban}; +use axum::extract::{Path, Request, State}; +use axum::http::{StatusCode, header}; +use axum::middleware::{self, Next}; +use axum::response::{IntoResponse, Response}; +use axum::routing::get; +use axum::{Json, Router}; +use serde::{Deserialize, Serialize}; +use std::net::{IpAddr, SocketAddr}; +use std::sync::Arc; +use std::time::Duration; +use tokio::net::TcpListener; + +/// The tunnel may come up after etrelay: binding is retried rather than given up. +const BIND_RETRY: Duration = Duration::from_secs(10); + +pub async fn serve(addr: SocketAddr, token: String, shared: Shared) { + let app = Router::new() + .route("/sessions", get(sessions)) + .route("/bans", get(list_bans)) + .route("/bans/{ip}", axum::routing::put(set_ban).delete(lift_ban)) + .route_layer(middleware::from_fn_with_state(Arc::new(token), auth)) + .with_state(shared); + + loop { + match TcpListener::bind(addr).await { + Ok(listener) => { + println!("control API on {addr}"); + if let Err(e) = axum::serve(listener, app.clone()).await { + eprintln!("control API {addr}: {e}"); + } + } + Err(e) => eprintln!("control API {addr}: {e}, retrying in {BIND_RETRY:?}"), + } + + tokio::time::sleep(BIND_RETRY).await; + } +} + +async fn auth(State(token): State>, req: Request, next: Next) -> Response { + let given = req + .headers() + .get(header::AUTHORIZATION) + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.strip_prefix("Bearer ")) + .unwrap_or(""); + + if !same(given.as_bytes(), token.as_bytes()) { + return error(StatusCode::UNAUTHORIZED, "authentication required"); + } + + next.run(req).await +} + +/// Constant time: the time taken doesn't tell how much of the token was right. +fn same(a: &[u8], b: &[u8]) -> bool { + a.len() == b.len() && a.iter().zip(b).fold(0u8, |acc, (x, y)| acc | (x ^ y)) == 0 +} + +#[derive(Serialize)] +struct SessionView { + /// The player's real address. + client: SocketAddr, + /// Source port of the session socket: the port the ET server sees. + port: u16, + /// Unix seconds. + since: u64, +} + +async fn sessions(State(shared): State) -> Json> { + let relay = shared.lock().expect("relay state"); + + let mut list: Vec<_> = relay + .sessions + .iter() + .filter(|(_, s)| !s.tx.is_closed()) + .map(|(client, s)| SessionView { + client: *client, + port: s.port, + since: s.since, + }) + .collect(); + list.sort_by_key(|s| s.since); + + Json(list) +} + +#[derive(Serialize)] +struct BanView { + ip: IpAddr, + #[serde(flatten)] + ban: Ban, +} + +async fn list_bans(State(shared): State) -> Response { + let mut relay = shared.lock().expect("relay state"); + + match relay.bans.list(bans::now()) { + Ok(list) => Json( + list.into_iter() + .map(|(ip, ban)| BanView { ip, ban }) + .collect::>(), + ) + .into_response(), + Err(e) => error(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}")), + } +} + +#[derive(Deserialize)] +struct BanRequest { + #[serde(default)] + reason: String, + /// None: until lifted. + minutes: Option, +} + +/// Bans ip and cuts its sessions at once: its packets stop here, the server times the +/// player out. +async fn set_ban( + State(shared): State, + Path(ip): Path, + Json(req): Json, +) -> Response { + let now = bans::now(); + let ban = Ban { + reason: req.reason, + at: now, + until: req + .minutes + .map(|m| now.saturating_add(m.saturating_mul(60))), + }; + + let mut relay = shared.lock().expect("relay state"); + if let Err(e) = relay.bans.set(ip, ban.clone()) { + return error(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}")); + } + + relay.sessions.retain(|client, _| client.ip() != ip); + println!("ban {ip} until {:?}: {}", ban.until, ban.reason); + + Json(BanView { ip, ban }).into_response() +} + +async fn lift_ban(State(shared): State, Path(ip): Path) -> Response { + let mut relay = shared.lock().expect("relay state"); + + match relay.bans.remove(ip) { + Ok(true) => { + println!("ban {ip} lifted"); + StatusCode::NO_CONTENT.into_response() + } + Ok(false) => error(StatusCode::NOT_FOUND, "not banned"), + Err(e) => error(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}")), + } +} + +fn error(code: StatusCode, message: &str) -> Response { + (code, Json(serde_json::json!({ "error": message }))).into_response() +} diff --git a/src/filter.rs b/src/filter.rs new file mode 100644 index 0000000..a2eaabc --- /dev/null +++ b/src/filter.rs @@ -0,0 +1,186 @@ +//! Which client packets reach the ET server: in-game packets always, connectionless ones +//! only when a client or a master would send them, queries rate limited per source IP. +use crate::OOB; +use std::collections::HashMap; +use std::net::IpAddr; +use std::time::Duration; +use tokio::time::Instant; + +/// 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); + +pub struct Filter { + queries: HashMap, + forgotten: Instant, +} + +/// Token bucket of one source IP's queries. +struct Bucket { + tokens: f64, + at: Instant, +} + +impl Filter { + pub 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. + pub 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] +} + +#[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); + } +} diff --git a/src/main.rs b/src/main.rs index 2d47ae5..a7d9c17 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,16 +1,26 @@ //! UDP relay for a Wolfenstein: Enemy Territory server reachable only through a tunnel. //! //! - Proxies every client (players, server browsers, masters) to the ET server, one -//! upstream socket per client address, like nginx `stream`. +//! upstream socket per client address, like nginx `stream`. The ET server sees each +//! client as :: the control API maps that +//! port back to the real address. //! - Sends master heartbeats from the public socket, so masters challenge (`getinfo`) the //! public address: the challenge goes through the relay like any client packet, the ET //! server answers it, and the master lists the public address. -use anyhow::{Context, Result}; +//! - Drops what a client has no business sending (see filter) and banned addresses. +mod bans; +mod control; +mod filter; + +use anyhow::{Context, Result, bail}; +use bans::Bans; use clap::Parser; +use filter::Filter; use std::collections::HashMap; -use std::net::{IpAddr, SocketAddr}; -use std::sync::Arc; +use std::net::SocketAddr; +use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::net::{UdpSocket, lookup_host}; use tokio::sync::mpsc::{self, error::TrySendError}; @@ -29,26 +39,11 @@ 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; +/// Shortest control token accepted. +const MIN_TOKEN: usize = 32; static DEBUG: AtomicBool = AtomicBool::new(false); @@ -79,16 +74,57 @@ struct Cli { )] masters: Vec, + /// Control API address (session ports, bans), on the tunnel only, e.g. 10.66.0.1:8090; + /// bearer token in $ETRELAY_CONTROL_TOKEN + #[arg(long)] + control: Option, + + /// Folder of the bans file (bans.json) + #[arg(long, default_value = "/var/lib/etrelay")] + state_dir: PathBuf, + /// Log sessions, connectionless packets (getinfo, getstatus, connect...), probes and heartbeats #[arg(long)] debug: bool, } +/// What the relay loop and the control API share. +pub struct Relay { + pub sessions: HashMap, + pub bans: Bans, +} + +pub struct Session { + /// Closed once the session ended; dropping it ends the session. + pub tx: mpsc::Sender>, + /// Source port of the upstream socket: the port the ET server sees. + pub port: u16, + /// Unix seconds. + pub since: u64, +} + +pub type Shared = Arc>; + #[tokio::main(flavor = "current_thread")] async fn main() -> Result<()> { let cli = Cli::parse(); DEBUG.store(cli.debug, Ordering::Relaxed); + let bans = Bans::load(cli.state_dir.join("bans.json"))?; + let shared: Shared = Arc::new(Mutex::new(Relay { + sessions: HashMap::new(), + bans, + })); + + if let Some(addr) = cli.control { + let token = std::env::var("ETRELAY_CONTROL_TOKEN").unwrap_or_default(); + if token.len() < MIN_TOKEN { + bail!("--control: $ETRELAY_CONTROL_TOKEN must hold at least {MIN_TOKEN} characters"); + } + + tokio::spawn(control::serve(addr, token, shared.clone())); + } + let public = UdpSocket::bind(cli.listen) .await .with_context(|| format!("bind {}", cli.listen))?; @@ -102,12 +138,11 @@ async fn main() -> Result<()> { ); tokio::spawn(heartbeat(public.clone(), cli.upstream, cli.masters)); - relay(public, cli.upstream).await + relay(public, cli.upstream, shared).await } /// 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(); +async fn relay(public: Arc, upstream: SocketAddr, shared: Shared) -> Result<()> { let mut filter = Filter::new(Instant::now()); let mut buf = vec![0u8; MAX_PACKET]; @@ -121,52 +156,87 @@ async fn relay(public: Arc, upstream: SocketAddr) -> Result<()> { let mut packet = buf[..n].to_vec(); - if let Some(tx) = sessions.get(&client) { - match tx.try_send(packet) { - // Full: drop it, like any congested UDP hop. - Ok(()) | Err(TrySendError::Full(_)) => continue, - // Session expired: open a new one below. - Err(TrySendError::Closed(p)) => packet = p, + // Never held across an await: the control API takes it too. + { + let mut relay = shared.lock().expect("relay state"); + + if relay.bans.banned(client.ip(), bans::now()) { + debug!("{client} dropped: banned"); + continue; + } + + if let Some(session) = relay.sessions.get(&client) { + match session.tx.try_send(packet) { + // Full: drop it, like any congested UDP hop. + Ok(()) | Err(TrySendError::Full(_)) => continue, + // Session expired: open a new one below. + Err(TrySendError::Closed(p)) => packet = p, + } + } + + relay.sessions.retain(|_, s| !s.tx.is_closed()); + + if relay.sessions.len() >= MAX_SESSIONS { + debug!("{client} dropped: {MAX_SESSIONS} sessions open"); + continue; } } - 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) => { + Ok((tx, port)) => { let _ = tx.try_send(packet); - sessions.insert(client, tx); - debug!("session + {client} ({} active)", sessions.len()); + + let mut relay = shared.lock().expect("relay state"); + relay.sessions.insert( + client, + Session { + tx, + port, + since: bans::now(), + }, + ); + debug!( + "session + {client} as port {port} ({} active)", + relay.sessions.len() + ); } Err(e) => eprintln!("session {client}: {e:#}"), } } } -/// Forwards packets between one client and the ET server until SESSION_TIMEOUT of silence. +/// Forwards packets between one client and the ET server until SESSION_TIMEOUT of +/// silence, or until its sender is dropped (ban). Returns the sender and the port the ET +/// server sees. async fn open_session( public: Arc, upstream: SocketAddr, client: SocketAddr, -) -> Result>> { +) -> Result<(mpsc::Sender>, u16)> { let sock = connect(upstream).await?; + let port = sock.local_addr()?.port(); let (tx, mut rx) = mpsc::channel::>(SESSION_QUEUE); tokio::spawn(async move { let mut buf = vec![0u8; MAX_PACKET]; let mut deadline = Instant::now() + SESSION_TIMEOUT; let (mut sent, mut received) = (0u64, 0u64); + let mut connected = false; loop { tokio::select! { - Some(packet) = rx.recv() => { + packet = rx.recv() => { + let Some(packet) = packet else { break }; + if let Some(command) = command(&packet) { debug!("{client} -> ET {command} ({} B)", packet.len()); + + // A player joining, not a browser query: kept in the journal, + // the ET logs only know the port. + if !connected && command.eq_ignore_ascii_case("connect") { + println!("connect {client} as port {port}"); + connected = true; + } } if let Err(e) = sock.send(&packet).await { debug!("{client} -> ET send failed: {e}"); @@ -193,7 +263,7 @@ async fn open_session( debug!("session - {client}: {sent} packets to ET, {received} from ET"); }); - Ok(tx) + Ok((tx, port)) } /// Heartbeats masters while the ET server answers, sends one flatline when it stops. @@ -258,93 +328,6 @@ 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 { @@ -370,78 +353,3 @@ 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); - } -}