feat: control API with session ports and persistent bans

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<u64>,
+}
+
+impl Ban {
+    fn active(&self, now: u64) -> bool {
+        self.until.is_none_or(|until| now < until)
+    }
+}
+
+pub struct Bans {
+    path: PathBuf,
+    list: HashMap<IpAddr, Ban>,
+}
+
+impl Bans {
+    /// Reads path; a missing file is no bans.
+    pub fn load(path: PathBuf) -> Result<Self> {
+        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<Vec<(IpAddr, Ban)>> {
+        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<bool> {
+        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<u64>) -> 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());
+    }
+}

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 <tunnel address>:<port of their session socket>), 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<Arc<String>>, 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<Shared>) -> Json<Vec<SessionView>> {
+    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<Shared>) -> 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::<Vec<_>>(),
+        )
+        .into_response(),
+        Err(e) => error(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}")),
+    }
+}
+
+#[derive(Deserialize)]
+struct BanRequest {
+    #[serde(default)]
+    reason: String,
+    /// None: until lifted.
+    minutes: Option<u64>,
+}
+
+/// 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<Shared>,
+    Path(ip): Path<IpAddr>,
+    Json(req): Json<BanRequest>,
+) -> 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<Shared>, Path(ip): Path<IpAddr>) -> 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()
+}

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<IpAddr, Bucket>,
+    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<u8> {
+        [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);
+    }
+}

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 <tunnel address>:<port of its session socket>: 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<String>,
 
+    /// 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<SocketAddr>,
+
+    /// 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<SocketAddr, Session>,
+    pub bans: Bans,
+}
+
+pub struct Session {
+    /// Closed once the session ended; dropping it ends the session.
+    pub tx: mpsc::Sender<Vec<u8>>,
+    /// Source port of the upstream socket: the port the ET server sees.
+    pub port: u16,
+    /// Unix seconds.
+    pub since: u64,
+}
+
+pub type Shared = Arc<Mutex<Relay>>;
+
 #[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<UdpSocket>, upstream: SocketAddr) -> Result<()> {
-    let mut sessions: HashMap<SocketAddr, mpsc::Sender<Vec<u8>>> = HashMap::new();
+async fn relay(public: Arc<UdpSocket>, 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<UdpSocket>, 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<UdpSocket>,
     upstream: SocketAddr,
     client: SocketAddr,
-) -> Result<mpsc::Sender<Vec<u8>>> {
+) -> Result<(mpsc::Sender<Vec<u8>>, u16)> {
     let sock = connect(upstream).await?;
+    let port = sock.local_addr()?.port();
     let (tx, mut rx) = mpsc::channel::<Vec<u8>>(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<Soc
     Ok(addr)
 }
 
-/// Decides which client packets reach the ET server.
-struct Filter {
-    queries: HashMap<IpAddr, Bucket>,
-    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<String> {
@@ -370,78 +353,3 @@ async fn connect(upstream: SocketAddr) -> Result<UdpSocket> {
     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<u8> {
-        [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);
-    }
-}