feat: relay several servers, registered through the control API

2026-10-01 13:32:45 +02:00
parent af2dc3af53
commit 6b2b137c1d
18 changed files with 1496 additions and 376 deletions
@@ -114,12 +114,27 @@ dependencies = [
 "tower-service",
 ]
 [[package]]
 name = "block-buffer"
 version = "0.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
 dependencies = [
 "generic-array",
 ]
 [[package]]
 name = "bytes"
 version = "1.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
 [[package]]
 name = "cfg-if"
 version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600"
 [[package]]
 name = "clap"
 version = "4.6.7"
@@ -166,15 +181,45 @@ version = "1.0.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 [[package]]
 name = "cpufeatures"
 version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280"
 dependencies = [
 "libc",
 ]
 [[package]]
 name = "crypto-common"
 version = "0.1.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
 dependencies = [
 "generic-array",
 "typenum",
 ]
 [[package]]
 name = "digest"
 version = "0.10.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
 "block-buffer",
 "crypto-common",
 ]
 [[package]]
 name = "etrelay"
-version = "0.2.0"
+version = "0.3.0"
 dependencies = [
 "anyhow",
 "axum",
 "clap",
 "serde",
 "serde_json",
 "sha2",
 "tokio",
 ]
@@ -220,6 +265,16 @@ dependencies = [
 "slab",
 ]
 [[package]]
 name = "generic-array"
 version = "0.14.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
 dependencies = [
 "typenum",
 "version_check",
 ]
 [[package]]
 name = "heck"
 version = "0.5.0"
@@ -461,6 +516,17 @@ dependencies = [
 "serde",
 ]
 [[package]]
 name = "sha2"
 version = "0.10.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
 dependencies = [
 "cfg-if",
 "cpufeatures",
 "digest",
 ]
 [[package]]
 name = "slab"
 version = "0.4.12"
@@ -558,6 +624,12 @@ version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3"
 [[package]]
 name = "typenum"
 version = "1.20.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
 [[package]]
 name = "unicode-ident"
 version = "1.0.26"
@@ -570,6 +642,12 @@ version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 [[package]]
 name = "version_check"
 version = "0.9.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
 [[package]]
 name = "wasi"
 version = "0.11.1+wasi-snapshot-preview1"
@@ -1,6 +1,6 @@
 [package]
 name = "etrelay"
-version = "0.2.0"
+version = "0.3.0"
 edition = "2024"
 [dependencies]
@@ -9,6 +9,7 @@ axum = { version = "0.8", default-features = false, features = ["http1", "json",
 clap = { version = "4.5.51", features = ["derive"] }
 serde = { version = "1.0.229", features = ["derive"] }
 serde_json = "1.0.151"
 sha2 = "0.10"
 tokio = { version = "1", features = ["macros", "net", "rt", "sync", "time"] }
 [profile.release]
@@ -16,7 +16,7 @@ Our ET server sits at home, reachable from the Internet only through the front s
 ## The fix
-`etrelay` runs on the front and owns the public UDP port 27960. It does two things on **the same socket**:
+`etrelay` runs on the front and owns the public UDP ports, one per ET server (27960 for the first). For each server it does two things on **the same socket**:
 - **Proxy**: forwards every client (players, server browsers, masters) to the ET server, one upstream socket per client address. Packets a client has no business sending (`rcon`...) are dropped on the way.
 - **Heartbeat**: sends the heartbeats itself, so the master challenges the **front** address. That challenge is just another client packet: it goes through the proxy, the ET server answers it, the answer leaves from the front address, and the master lists the front.
@@ -29,7 +29,7 @@ Masters are independent: a server is only listed on the masters it heartbeats, a
 ![After: everything goes through the front](diagrams/architecture.png)
-## Registration sequence
+## Master registration sequence
 ![Master registration (every 5 min)](diagrams/registration.png)
@@ -55,16 +55,42 @@ The last two lines are a server tracker that got the address from the master.
 The ET server sees every client as `<front-wg>:<port>`: one source IP (the front's tunnel address), one port per session, the port of the session's upstream socket. Its own IP bans can't tell players apart: bans are etrelay's (below). A player coming back after the session expired gets a new port.
 ## Several servers: registering
 The relay holds several ET servers. The ET protocol has no host name to tell them apart: each one gets its own public port (`--ports`), with its own socket, sessions, heartbeats and bans. Masters list servers by address and port, so several on the front's address are fine.
 A server is relayed once its admin tool registered it, with `PUT /server` on the control API (`--control <addr>`, on the front's tunnel address, unreachable from the internet):
 ![Registering a server](diagrams/pairing.png)
 - **The relay's token** only lets a tool register: it keeps strangers of the tunnel off the public ports.
 - **The key** is picked by the tool, and is what every later call carries: it names the server, so a tool only reaches its own. etrelay keeps its SHA-256 (`servers.json`), never the key.
 - **The proof**: before taking a key, etrelay checks the tool runs the ET server it registers. The tool makes the server show the key's fingerprint (the first 128 bits of its SHA-256) in the serverinfo cvar `etrelay_proof`, etrelay reads it there with a `getstatus`. Only who holds that server's console can do it, wherever the request comes from (a NAT between the two changes nothing), and the fingerprint, public for a moment, doesn't give the key away. An empty cvar is left out of serverinfo: the tool sets it to `""` afterwards (`unset` isn't in 2.60b).
 What can't happen:
 - **two servers on one port**: the second one gets a 409, the first one is untouched;
 - **another tool on a registered server**: it has neither the key, nor the server's console to prove another one.
 One limit: a server that is gone for good without `DELETE /server` keeps its port (the masters drop it, nothing answers there), until its entry is removed from `servers.json` and etrelay restarted.
 An ET server registered again is taken over by who proves to run it (a tool reinstalled, with a new key): the former key is refused from then on. On the same port the relay goes on (sessions, bans); on another port the former one is freed. A relay that lost its state answers 401 to every key: tools register again by themselves.
 The registered servers are relayed again after a restart. `DELETE /server` leaves the relay: the masters get a flatline, the port is freed, the bans dropped.
 `etrelay <UPSTREAM>` still relays one server without any of this, on `--listen`: it has no key, so no control API over it.
 ## Control API and bans
-With `--control <addr>`, an HTTP API listens on the front's tunnel address (unreachable from the internet), bearer token on top. It serves an admin tool on the tunnel side:
+Once registered, the admin tool asks about its server, with its key:
 - `GET /server`: the registration, and whether the ET server answered the last probe.
 - `GET /sessions`: the open sessions, real client address and port the ET server sees, so the tool can put a real address on each player.
 - `GET /bans`, `PUT /bans/{ip}`, `DELETE /bans/{ip}`: bans by source IP, with a reason and an optional duration.
 ![Control API: real address and ban of a player](diagrams/ban.png)
-A ban cuts the address's sessions at once, and the relay loop drops its packets before any session lookup: the ET server gets nothing more from the player and times them out, and they can't come back. Bans are saved to `<state-dir>/bans.json` on each change (written next to it, then renamed), expired ones forgotten.
+A ban cuts the address's sessions at once, and the relay loop drops its packets before any session lookup: the ET server gets nothing more from the player and times them out, and they can't come back. Each server has its own bans, saved to `<state-dir>/bans-<port>.json` on each change (written next to it, then renamed), expired ones forgotten.
 ## Filter
@@ -78,14 +104,16 @@ Only what clients and masters send a server gets through, checked on the first w
 ## Internals
-Single-threaded tokio runtime, four kinds of tasks. The sessions and the bans are shared by the relay loop and the control API behind a mutex, never held across an `await`:
+Single-threaded tokio runtime, four kinds of tasks. Each server has its relay loop, its heartbeat task and its state (sessions, bans), shared with the control API behind a mutex. The registry (the servers by public port, the hash of their key) is behind another one, which the packets never take. Neither is held across an `await`:
 ![Tasks and data flow](diagrams/tasks.png)
- **relay loop**: reads the public socket, runs the filter and the ban check, finds the client's session and pushes the packet into its queue. No session, or an expired one: opens a new one (and purges expired entries), unless `MAX_SESSIONS` are open. A full queue drops the packet, like any congested UDP hop.
+- **relay loop** (one per server): reads the server's public socket, runs the filter and the ban check, finds the client's session and pushes the packet into its queue. No session, or an expired one: opens a new one (and purges expired entries), unless `MAX_SESSIONS` are open, all servers together. A full queue drops the packet, like any congested UDP hop.
 - **session task**: owns one ephemeral socket connected to the ET server. Forwards the queue to the ET server and the ET server answers back to the client **through the public socket** (so they come from `:27960`). Ends after `SESSION_TIMEOUT` without a packet either way, or at once when a ban drops its queue's sender.
- **heartbeat task**: probes the ET server with `getinfo` on its own socket (up to 3 attempts, 3 s each), then heartbeats the masters through the public socket. Masters are resolved on every heartbeat (DNS may change).
+- **heartbeat task** (one per server): probes the ET server with `getinfo` on its own socket (up to 3 attempts, 3 s each), then heartbeats the masters through the public socket. Masters are resolved on every heartbeat (DNS may change).
- **control API task** (with `--control`): axum on the tunnel address. Binding is retried every 10 s: the tunnel may come up after etrelay.
+- **control API task** (with `--control`): axum on the tunnel address. Binding is retried every 10 s: the tunnel may come up after etrelay. Registering starts a server's two tasks, leaving aborts them and drops its sessions.
 The code follows that split: `server` (one relayed server), `registry` (which servers, `servers.json`), `proof` (keys and their proof), `control` (the API), `et` (the probes: `getinfo`, `getstatus`), `filter`, `bans`.
 ### Heartbeat states
@@ -102,10 +130,11 @@ The master drops a server that stops heartbeating after a while anyway; the flat
 | `SESSION_TIMEOUT` | 2 min | Idle clients (server browsers) freed quickly; players send packets constantly |
 | `SESSION_QUEUE` | 64 packets | Burst absorption per client before dropping |
 | `MAX_PACKET` | 65 535 B | Any UDP datagram: `statusResponse` easily exceeds nginx's old 512 B buffer |
-| `MAX_SESSIONS` | 2048 | One socket each: spoofed sources must not use up the file descriptors (`LimitNOFILE=4096` in the unit) |
+| `MAX_SESSIONS` | 2048 | One socket each, all servers together: spoofed sources must not use up the file descriptors (`LimitNOFILE=4096` in the unit) |
 | `QUERY_BURST`, `QUERY_RATE` | 5, then 1/s | `getstatus`/`getinfo` per source IP: a browser refresh or a master's check passes, a reflection attack doesn't |
 | `MAX_QUERY_SOURCES`, `QUERY_FORGET` | 65 536 IPs, 60 s | The limiter's memory stays bounded under a flood of spoofed sources |
-| `MIN_TOKEN` | 32 characters | Shortest control API token accepted |
+| `MIN_TOKEN`, `MIN_KEY` | 32 characters | Shortest relay token and server key accepted |
 | `PROOF_BYTES` | 16 | Bytes of the key's SHA-256 shown as its fingerprint: serverinfo holds 1024 characters for every cvar |
 ### Protocol cheat sheet
@@ -1,26 +1,34 @@
 # etrelay
-UDP relay + master heartbeat for a Wolfenstein: Enemy Territory server reachable only through a tunnel: runs on the front server (public IP) and makes the server show up in the in-game browser.
+UDP relay + master heartbeat for Wolfenstein: Enemy Territory servers reachable only through a tunnel: runs on the front server (public IP) and makes the servers show up in the in-game browser. Each server gets its own public port.
 ```
-players / masters ──> front :27960 (etrelay) ══wg══> ET <et-server>:27960
+players / masters ──> front :27960 (etrelay) ══wg══> ET <et-server-1>:27960
-                      front <front-wg>:8090 (control API) <══wg══ admin tool
+                      front :27961 (etrelay) ══wg══> ET <et-server-2>:27960
                      front <front-wg>:8090 (control API) <══wg══ admin tools
 ```
 A server is relayed because its admin tool (etadm) **registered** it through the control API, or because it was given on the command line (one server, no control API over it).
 - **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`.
+- **Heartbeat**: every 5 min, `getinfo` to the ET server; if it answers, `heartbeat EnemyTerritory-1` to the masters **from the server's public socket**. When it stops answering, or when the server leaves the relay, 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.
+Masters list servers by address and port, several on one address are fine. They check (`getChallenge`, `getInfo`) the address the heartbeat came from: the server's public port. Those packets go through the relay like any client, the ET server answers them, and the master lists the front public address.
 How it works in detail, with diagrams: [ENGINE.md](ENGINE.md).
 ## Usage
 ```bash
-etrelay <UPSTREAM> [--listen 0.0.0.0:27960] [--master host:port]... [--control <addr>] [--state-dir /var/lib/etrelay] [--debug]
+etrelay [UPSTREAM] [--listen 0.0.0.0:27960] [--ports 27960-27969] [--master host:port]... [--control <addr>] [--state-dir /var/lib/etrelay] [--debug]
 ```
 - `--control <addr>`: servers register through the control API, on a port of `--ports` bound on `--listen`'s address. They are kept in `<state-dir>/servers.json` and relayed again after a restart.
 - `UPSTREAM` (an ET server's `ip:port`): relayed on `--listen` from the start, without registering. Nothing reaches it through the control API: no real addresses, no bans. It can't be registered, nor its port taken.
 One of the two at least. The ports of `--ports` must be open (UDP) on the front's firewall.
 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.
 `--debug` logs sessions (open/close, packet counts), connectionless packets by command name only (`getinfo`, `getstatus`, `connect`...: no arguments, `connect` carries the userinfo), probes and heartbeats. In-game packets are only counted.
@@ -29,7 +37,7 @@ Default masters: `etmaster.idsoftware.com:27950` (id Software, the one ET 2.60b
 probe <wg-peer>:27960: up
 ET server <wg-peer>:27960 is up
 heartbeat EnemyTerritory-1 -> etmaster.net:27950 (198.51.100.10:27950)
-session + 198.51.100.10:27950 as port 51234 (2 active)
+session + 198.51.100.10:27950 as port 51234 (2 active on port 27960)
 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)
@@ -43,18 +51,44 @@ Without `--debug`, each player joining is logged with its real address and the p
 ## Control API
-`--control <addr>` 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.
+`--control <addr>` serves a small HTTP API for admin tools (etadm), on the tunnel address only (e.g. `10.66.0.1:8090`: unreachable from the internet). Binding is retried every 10 s: the tunnel may come up after etrelay.
-The ET server sees every player as `<tunnel address>:<port>`, the port of the player's session socket; the API maps it back to the real address.
+### Registering a server
-| Method | URL | |
+An admin tool registers the ET server it runs, with two things:
 |---|---|---|
 | GET | `/sessions` | `[{"client": "203.0.113.7:27960", "port": 51234, "since": <unix>}]`: 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 `<state-dir>/bans.json` (written atomically), expired ones forgotten.
+- **the relay's token** (`$ETRELAY_CONTROL_TOKEN`, 32 characters at least): it only lets a server register;
 - **a key it picks** (32 characters at least), and the proof it runs that server: it makes the server show the key's fingerprint in the serverinfo cvar `etrelay_proof` (`sets etrelay_proof <fingerprint>` on its console), where etrelay reads it with a `getstatus` before taking the key. The fingerprint is the first 32 hex digits of the key's SHA-256: public while registering, it doesn't give the key away. The tool removes the cvar afterwards (`set etrelay_proof ""`).
 Then every other call carries the key (`Authorization: Bearer <key>`), which names the server: a tool only sees and changes its own server. etrelay keeps the key's SHA-256, never the key.
 | Method | URL | Bearer | |
 |---|---|---|---|
 | PUT | `/server` | token | `{"name": "...", "port": 27960, "upstream_ip": "10.66.0.20", "upstream_port": 27960, "key": "..."}`: register the ET server `upstream_ip:upstream_port` (as seen from the relay; `upstream_ip` left out: the caller's address) on the public `port`. Answers the registration (below) |
 | GET | `/server` | key | `{"port", "upstream", "name", "since", "alive"}`: `alive`, the ET server answered the last probe, the masters are told |
 | DELETE | `/server` | key | leave the relay: the port is freed, the masters get a flatline, the server's bans are dropped (204) |
 | GET | `/sessions` | key | `[{"client": "203.0.113.7:27960", "port": 51234, "since": <unix>}]`: open sessions |
 | GET | `/bans` | key | `[{"ip", "reason", "at", "until"}]` (`until` null: until lifted) |
 | PUT | `/bans/{ip}` | key | `{"reason": "...", "minutes": N}` (`minutes` left out: until lifted); cuts the address's sessions at once |
 | DELETE | `/bans/{ip}` | key | 204, 404 when not banned |
 `PUT /server` answers:
 | Status | |
 |---|---|
 | 200 | registered |
 | 400 | port outside `--ports`, key too short |
 | 401 | wrong token |
 | 403 | the ET server doesn't answer, or doesn't show the key's fingerprint |
 | 409 | another server holds the port, or the ET server is the command line's |
 One server per public port, one port per server. Registering a server again (same ET server, another key: an admin tool reinstalled) takes it over: who proves to run it holds it, the former key is refused from then on. On the same port its sessions and bans go on; on another port the former one is freed. Unknown keys get 401: a tool whose relay lost its state registers again.
 ### Sessions and bans
 The ET server sees every player as `<tunnel address>:<port>`, the port of the player's session socket; `/sessions` maps it back to the real address.
 A banned address's packets are dropped before reaching the server: the player times out, and can't come back. Each server has its own bans, kept in `<state-dir>/bans-<port>.json` (written atomically), expired ones forgotten (`bans.json`: the command line's server).
 ## Build
@@ -67,20 +101,26 @@ make release   # both
 Artifacts land in `dist/`, named `<name>-v<version>-<target triple>` with a `.sha256` checksum:
 ```
-dist/etrelay-v0.1.0-x86_64-unknown-linux-gnu    # dynamic
+dist/etrelay-v0.3.0-x86_64-unknown-linux-gnu    # dynamic
-dist/etrelay-v0.1.0-x86_64-unknown-linux-musl   # static
+dist/etrelay-v0.3.0-x86_64-unknown-linux-musl   # static
 ```
 ## Run as a service
-1. Binary and unit on the front server ([etrelay.service](etrelay.service) is a template: replace `@UPSTREAM@` with the ET server `ip:port` as seen from the front, `@ARGS@` with extra flags or nothing):
+1. The token servers register with, on the front server (root only, systemd hands it to the service):
   ```bash
   install -d -m 700 /etc/etrelay
   echo "ETRELAY_CONTROL_TOKEN=$(head -c 32 /dev/urandom | od -An -tx1 | tr -d ' \n')" > /etc/etrelay/env
   ```
 2. Binary and unit ([etrelay.service](etrelay.service) is a template: `@UPSTREAM@` is nothing when servers register, or one ET server's `ip:port` as seen from the front; `@ARGS@` the flags):
   ```bash
   install -m 755 dist/etrelay-v<version>-x86_64-unknown-linux-musl /usr/local/bin/etrelay
-   sed -e 's#@UPSTREAM@#<ip:port>#' -e 's#@ARGS@##' etrelay.service > /etc/systemd/system/etrelay.service
+   sed -e 's#@UPSTREAM@##' -e 's#@ARGS@#--control <tunnel address>:8090 --ports 27960-27969#' etrelay.service > /etc/systemd/system/etrelay.service
   systemctl daemon-reload && systemctl enable --now etrelay
   ```
-2. Nothing else on the front may capture UDP 27960 (DNAT rule, nginx `stream`...), or players bypass the relay: `iptables -t nat -S | grep 27960` must print nothing.
+3. The UDP ports of `--ports` open on the front. Nothing else may capture them (DNAT rule, nginx `stream`...), or players bypass the relay: `iptables -t nat -S | grep 27960` must print nothing.
-3. ET server `server.cfg`: stop its own heartbeats, they leave through the home connection with the wrong IP:
+4. Each admin tool gets the relay's address, the token and the public port it asks for (etadm: `--relay-url`, `--relay-token-file`, `--relay-port`, or its Relay page).
 5. ET server `server.cfg`: stop its own heartbeats, they leave through the home connection with the wrong IP:
   ```
   set sv_master1 ""
   set sv_master2 ""
@@ -92,7 +132,7 @@ dist/etrelay-v0.1.0-x86_64-unknown-linux-musl   # static
 ## Check
 ```bash
-journalctl -u etrelay -n 50                      # "ET server ... is up"
+journalctl -u etrelay -n 50                      # "registered ... on port 27960", "ET server ... is up"
 tcpdump -ni any udp port 27950                   # heartbeat out, getChallenge/getInfo in, answers out
 ```
@@ -9,13 +9,14 @@ participant "ET server" as et
 actor "Player\n203.0.113.7" as player
 note over admin, et : the ET server shows the player as <front-wg>:51234
 note over admin, control : every call carries the key the server registered with
 admin -> control : GET /sessions
 control --> admin : [{client: 203.0.113.7:27960, port: 51234}]
 note right of admin : port 51234 = 203.0.113.7
 == Ban ==
 admin -> control : PUT /bans/203.0.113.7\n{reason, minutes}
-control -> control : bans.json (atomic write)\ncut 203.0.113.7's sessions
+control -> control : bans-<port>.json (atomic write)\ncut 203.0.113.7's sessions
 player -> relay : game packets
 relay -> relay : banned: dropped
 note over et : no packet: times the player out
@@ -0,0 +1,29 @@
@startuml
 skinparam shadowing false
 title Registering a server (pairing)
 participant "Admin tool\n(etadm)" as admin
 participant "ET server\n<et-server>:27960" as et
 participant "etrelay control API\n<front-wg>:8090" as control
 participant "Masters" as master
 note over admin : picks a key once\n(kept on its side)
 admin -> et : rcon: sets etrelay_proof <fingerprint of the key>
 admin -> control : PUT /server {port: 27960, upstream_port: 27960, key}\nBearer: the relay's token
 control -> control : port offered and free?\n(409: held by another server)
 control -> et : getstatus
 et --> control : statusResponse\n...\\etrelay_proof\\<fingerprint>...
 control -> control : fingerprint == sha256(key)?\n(403: not shown)
 control -> control : bind :27960, keep sha256(key)\nin servers.json
 control --> admin : 200 {port, upstream, since, alive}
 admin -> et : rcon: set etrelay_proof ""
 control -> master : heartbeat EnemyTerritory-1\nfrom :27960
 == Later ==
 admin -> control : GET /sessions, /bans...\nBearer: the key (names the server)
 control --> admin : this server's only
 note over admin, control : 401 (the relay lost its state):\nthe admin tool registers again
 admin -> control : DELETE /server
 control -> master : heartbeat ETFlatline-1
 note over control : port freed, bans dropped
@enduml
@@ -2,14 +2,18 @@
 skinparam shadowing false
 title Tasks and data flow
-rectangle "relay loop (main task)" as loop {
+rectangle "relay loop (one per server)" as loop {
  card "public socket\nrecv_from" as recv
  card "filter\n(commands, query rate)" as filter
 }
-rectangle "shared state (Mutex)" as shared {
+rectangle "server state (Mutex, one per server)" as shared {
  card "sessions:\nHashMap<client, Session>" as map
-  card "bans\n(bans.json)" as bans
+  card "bans\n(bans-<port>.json)" as bans
 }
 rectangle "registry (Mutex)" as registry {
  card "servers by public port,\ntheir key's hash\n(servers.json)" as servers
 }
 rectangle "session task (one per client)" as st {
@@ -17,12 +21,13 @@ rectangle "session task (one per client)" as st {
  card "upstream socket\n(connected to ET)" as up
 }
-rectangle "heartbeat task" as hb {
+rectangle "heartbeat task (one per server)" as hb {
  card "interval 5 min" as tick
  card "probe socket" as probe
 }
 rectangle "control API task (axum)" as ctl {
  card "PUT/GET/DELETE /server" as pairing
  card "GET /sessions\nGET/PUT/DELETE /bans" as routes
 }
@@ -36,4 +41,7 @@ tick --> probe : getinfo
 tick --> recv : heartbeat via public socket
 routes --> map : list ports,\ndrop a banned IP's sessions
 routes --> bans : set / lift
 routes --> servers : key -> server
 pairing --> servers : register / leave
 servers --> loop : start / stop\n(with its heartbeat)
@enduml
@@ -9,12 +9,14 @@ 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).
+# ETRELAY_CONTROL_TOKEN=... for --control (what servers register with); read by systemd,
 # root only (0600).
 EnvironmentFile=-/etc/etrelay/env
-# /var/lib/etrelay: the bans file, kept across restarts and upgrades.
+# /var/lib/etrelay: the registered servers and their bans, kept across restarts and
 # upgrades.
 StateDirectory=etrelay
-# Port 27960 > 1024: no privilege needed.
+# Ports 27960 and up > 1024: no privilege needed.
 DynamicUser=yes
 NoNewPrivileges=yes
 ProtectSystem=strict
@@ -1,14 +1,20 @@
-//! Control API for etadm: which real address is behind a session port (the ET server
+//! Control API for admin tools (etadm), on the tunnel address only.
-//! sees players as <tunnel address>:<port of their session socket>), and bans. Listens on
+//!
-//! the tunnel address only, bearer token on top.
+//! An admin tool registers the ET server it runs (PUT /server, with the relay's token and
-use crate::Shared;
+//! the proof it runs that server, see proof). Everything else carries the key it
 //! registered with, which names its server: which real address is behind a session port
 //! (the ET server sees players as <tunnel address>:<port of their session socket>), the
 //! bans, leaving the relay.
 use crate::bans::{self, Ban};
-use axum::extract::{Path, Request, State};
+use crate::proof::{self, KeyHash, MIN_KEY};
 use crate::registry::{Refused, Registration, Shared};
 use crate::server::Server;
 use axum::extract::{ConnectInfo, Path, Request, State};
 use axum::http::{StatusCode, header};
 use axum::middleware::{self, Next};
 use axum::response::{IntoResponse, Response};
-use axum::routing::get;
+use axum::routing::{get, put};
-use axum::{Json, Router};
+use axum::{Extension, Json, Router};
 use serde::{Deserialize, Serialize};
 use std::net::{IpAddr, SocketAddr};
 use std::sync::Arc;
@@ -18,19 +24,28 @@ 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) {
+pub async fn serve(addr: SocketAddr, token: String, registry: Shared) {
-    let app = Router::new()
+    let registered = Router::new()
        .route("/server", get(server).delete(unregister))
        .route("/sessions", get(sessions))
        .route("/bans", get(list_bans))
-        .route("/bans/{ip}", axum::routing::put(set_ban).delete(lift_ban))
+        .route("/bans/{ip}", put(set_ban).delete(lift_ban))
-        .route_layer(middleware::from_fn_with_state(Arc::new(token), auth))
+        .route_layer(middleware::from_fn_with_state(registry.clone(), by_key));
-        .with_state(shared);
+
    let app = Router::new()
        .route("/server", put(register))
        .route_layer(middleware::from_fn_with_state(Arc::new(token), by_token))
        .merge(registered)
        .with_state(registry);
    loop {
        match TcpListener::bind(addr).await {
            Ok(listener) => {
                println!("control API on {addr}");
-                if let Err(e) = axum::serve(listener, app.clone()).await {
+                let service = app
                    .clone()
                    .into_make_service_with_connect_info::<SocketAddr>();
                if let Err(e) = axum::serve(listener, service).await {
                    eprintln!("control API {addr}: {e}");
                }
            }
@@ -41,26 +56,134 @@ pub async fn serve(addr: SocketAddr, token: String, shared: Shared) {
    }
 }
-async fn auth(State(token): State<Arc<String>>, req: Request, next: Next) -> Response {
+fn bearer(req: &Request) -> &str {
-    let given = req
+    req.headers()
        .headers()
        .get(header::AUTHORIZATION)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.strip_prefix("Bearer "))
-        .unwrap_or("");
+        .unwrap_or("")
 }
-    if !same(given.as_bytes(), token.as_bytes()) {
+/// Registering: the relay's token.
-        return error(StatusCode::UNAUTHORIZED, "authentication required");
+async fn by_token(State(token): State<Arc<String>>, req: Request, next: Next) -> Response {
    if !same(bearer(&req).as_bytes(), token.as_bytes()) {
        return error(StatusCode::UNAUTHORIZED, "the relay's token is required");
    }
    next.run(req).await
 }
 /// Everything else: the key of a registered server, handed to the handler.
 async fn by_key(State(registry): State<Shared>, mut req: Request, next: Next) -> Response {
    let key = KeyHash::of(bearer(&req));
    let server = registry.lock().expect("registry").find(&key);
    let Some(server) = server else {
        return error(
            StatusCode::UNAUTHORIZED,
            "no server registered with this key",
        );
    };
    req.extensions_mut().insert(server);
    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(Deserialize)]
 struct RegisterRequest {
    /// Shown in the journal, to tell servers apart.
    #[serde(default)]
    name: String,
    /// Public port asked for.
    port: u16,
    /// The ET server as seen from here; its address defaults to the caller's (the admin
    /// tool runs next to its server).
    upstream_ip: Option<IpAddr>,
    upstream_port: u16,
    /// The key every later call carries, picked by the caller.
    key: String,
 }
 /// Registers the caller's ET server on a public port, once the server shows the key's
 /// fingerprint: 409 when another server holds the port, 403 without the proof.
 async fn register(
    State(registry): State<Shared>,
    ConnectInfo(peer): ConnectInfo<SocketAddr>,
    Json(req): Json<RegisterRequest>,
 ) -> Response {
    if req.key.len() < MIN_KEY {
        return error(
            StatusCode::BAD_REQUEST,
            &format!("the key must hold at least {MIN_KEY} characters"),
        );
    }
    let upstream = SocketAddr::new(req.upstream_ip.unwrap_or(peer.ip()), req.upstream_port);
    let key = KeyHash::of(&req.key);
    let allowed = registry.lock().expect("registry").check(req.port, upstream);
    if let Err(e) = allowed {
        return refused(e);
    }
    if let Err(e) = proof::shown(upstream, &key).await {
        return error(StatusCode::FORBIDDEN, &e.to_string());
    }
    let registration = Registration {
        port: req.port,
        upstream,
        name: req.name,
        key,
    };
    let registered = registry.lock().expect("registry").register(registration);
    match registered {
        Ok(view) => Json(view).into_response(),
        Err(e) => refused(e),
    }
 }
 fn refused(e: Refused) -> Response {
    let code = match e {
        Refused::Port(..) => StatusCode::BAD_REQUEST,
        Refused::Taken(_) | Refused::Static(_) => StatusCode::CONFLICT,
        Refused::Failed(_) => StatusCode::INTERNAL_SERVER_ERROR,
    };
    error(code, &e.to_string())
 }
 async fn server(
    State(registry): State<Shared>,
    Extension(server): Extension<Arc<Server>>,
 ) -> Response {
    let view = registry.lock().expect("registry").view(server.port);
    match view {
        Some(view) => Json(view).into_response(),
        None => error(StatusCode::NOT_FOUND, "not registered anymore"),
    }
 }
 /// Leaves the relay: the port is free, the masters are told, the bans forgotten.
 async fn unregister(
    State(registry): State<Shared>,
    Extension(server): Extension<Arc<Server>>,
 ) -> Response {
    let left = registry.lock().expect("registry").unregister(server.port);
    match left {
        Ok(()) => StatusCode::NO_CONTENT.into_response(),
        Err(e) => error(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}")),
    }
 }
 #[derive(Serialize)]
 struct SessionView {
    /// The player's real address.
@@ -71,13 +194,13 @@ struct SessionView {
    since: u64,
 }
-async fn sessions(State(shared): State<Shared>) -> Json<Vec<SessionView>> {
+async fn sessions(Extension(server): Extension<Arc<Server>>) -> Json<Vec<SessionView>> {
-    let relay = shared.lock().expect("relay state");
+    let state = server.state();
-    let mut list: Vec<_> = relay
+    let mut list: Vec<_> = state
        .sessions
        .iter()
-        .filter(|(_, s)| !s.tx.is_closed())
+        .filter(|(_, s)| s.open())
        .map(|(client, s)| SessionView {
            client: *client,
            port: s.port,
@@ -96,10 +219,10 @@ struct BanView {
    ban: Ban,
 }
-async fn list_bans(State(shared): State<Shared>) -> Response {
+async fn list_bans(Extension(server): Extension<Arc<Server>>) -> Response {
-    let mut relay = shared.lock().expect("relay state");
+    let mut state = server.state();
-    match relay.bans.list(bans::now()) {
+    match state.bans.list(bans::now()) {
        Ok(list) => Json(
            list.into_iter()
                .map(|(ip, ban)| BanView { ip, ban })
@@ -121,7 +244,7 @@ struct BanRequest {
 /// 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>,
+    Extension(server): Extension<Arc<Server>>,
    Path(ip): Path<IpAddr>,
    Json(req): Json<BanRequest>,
 ) -> Response {
@@ -134,23 +257,26 @@ async fn set_ban(
            .map(|m| now.saturating_add(m.saturating_mul(60))),
    };
-    let mut relay = shared.lock().expect("relay state");
+    let mut state = server.state();
-    if let Err(e) = relay.bans.set(ip, ban.clone()) {
+    if let Err(e) = state.bans.set(ip, ban.clone()) {
        return error(StatusCode::INTERNAL_SERVER_ERROR, &format!("{e:#}"));
    }
-    relay.sessions.retain(|client, _| client.ip() != ip);
+    state.sessions.retain(|client, _| client.ip() != ip);
-    println!("ban {ip} until {:?}: {}", ban.until, ban.reason);
+    println!(
        "port {}: ban {ip} until {:?}: {}",
        server.port, ban.until, ban.reason
    );
    Json(BanView { ip, ban }).into_response()
 }
-async fn lift_ban(State(shared): State<Shared>, Path(ip): Path<IpAddr>) -> Response {
+async fn lift_ban(Extension(server): Extension<Arc<Server>>, Path(ip): Path<IpAddr>) -> Response {
-    let mut relay = shared.lock().expect("relay state");
+    let mut state = server.state();
-    match relay.bans.remove(ip) {
+    match state.bans.remove(ip) {
        Ok(true) => {
-            println!("ban {ip} lifted");
+            println!("port {}: ban {ip} lifted", server.port);
            StatusCode::NO_CONTENT.into_response()
        }
        Ok(false) => error(StatusCode::NOT_FOUND, "not banned"),
@@ -0,0 +1,119 @@
 //! What etrelay itself says to an ET server: is it up (`getinfo`), what does it show
 //! (`getstatus`). Everything else is forwarded as is.
 use anyhow::Result;
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::time::Duration;
 use tokio::net::UdpSocket;
 use tokio::time::timeout;
 /// Connectionless packet prefix (Quake 3 / ET protocol).
 pub const OOB: &[u8] = b"\xff\xff\xff\xff";
 pub const MAX_PACKET: usize = 65_535;
 const PROBE_TIMEOUT: Duration = Duration::from_secs(3);
 const PROBE_ATTEMPTS: usize = 3;
 /// Longest connectionless command name shown in debug logs.
 const MAX_COMMAND: usize = 32;
 /// Ephemeral socket connected to `upstream`.
 pub async fn connect(upstream: SocketAddr) -> Result<UdpSocket> {
    let any: SocketAddr = if upstream.is_ipv4() {
        "0.0.0.0:0".parse()?
    } else {
        "[::]:0".parse()?
    };
    let sock = UdpSocket::bind(any).await?;
    sock.connect(upstream).await?;
    Ok(sock)
 }
 /// True if the ET server answers `getinfo`.
 pub async fn probe(upstream: SocketAddr) -> bool {
    ask(upstream, b"getinfo etrelay", b"infoResponse")
        .await
        .is_some()
 }
 /// The cvars the ET server shows everyone (`getstatus`'s serverinfo), names lowercased;
 /// None when it doesn't answer.
 pub async fn serverinfo(upstream: SocketAddr) -> Option<HashMap<String, String>> {
    let answer = ask(upstream, b"getstatus", b"statusResponse").await?;
    Some(parse_serverinfo(&answer))
 }
 /// Sends `query`, returns the answer when it is a `response` packet. A lost UDP packet
 /// must not count as a server down: PROBE_ATTEMPTS tries.
 async fn ask(upstream: SocketAddr, query: &[u8], response: &[u8]) -> Option<Vec<u8>> {
    for _ in 0..PROBE_ATTEMPTS {
        let attempt = async {
            let sock = connect(upstream).await?;
            sock.send(&[OOB, query].concat()).await?;
            let mut buf = vec![0u8; MAX_PACKET];
            let n = sock.recv(&mut buf).await?;
            buf.truncate(n);
            Ok::<_, anyhow::Error>(buf)
        };
        if let Ok(Ok(answer)) = timeout(PROBE_TIMEOUT, attempt).await
            && answer.starts_with(&[OOB, response].concat())
        {
            return Some(answer);
        }
    }
    None
 }
 /// `statusResponse\n\key\value\key\value...\n<one line per player>`.
 fn parse_serverinfo(answer: &[u8]) -> HashMap<String, String> {
    let text = String::from_utf8_lossy(answer);
    let line = text.lines().nth(1).unwrap_or("");
    let mut parts = line.split('\\').skip(1);
    let mut cvars = HashMap::new();
    while let (Some(key), Some(value)) = (parts.next(), parts.next()) {
        cvars.insert(key.to_ascii_lowercase(), value.to_string());
    }
    cvars
 }
 /// Command name of a connectionless packet (`getinfo`, `infoResponse`, `connect`...),
 /// None for in-game packets. Arguments are left out: `connect` carries the userinfo.
 pub fn command(packet: &[u8]) -> Option<String> {
    let body = packet.strip_prefix(OOB)?;
    let end = body
        .iter()
        .position(|b| b.is_ascii_whitespace() || *b == b'\\')
        .unwrap_or(body.len())
        .min(MAX_COMMAND);
    Some(String::from_utf8_lossy(&body[..end]).into_owned())
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn reads_the_serverinfo_line() {
        let answer = b"\xff\xff\xff\xffstatusResponse\n\\sv_hostname\\^1Big\\mapname\\oasis\\ETRelay_Proof\\abc\n12 48 \"joe\"\n";
        let cvars = parse_serverinfo(answer);
        assert_eq!(cvars["mapname"], "oasis");
        assert_eq!(cvars["etrelay_proof"], "abc");
        assert_eq!(cvars.len(), 3);
    }
    #[test]
    fn names_connectionless_commands_only() {
        assert_eq!(
            command(b"\xff\xff\xff\xffconnect \"\\name\\joe\"").as_deref(),
            Some("connect")
        );
        assert_eq!(command(b"\x01\x02ingame"), None);
    }
 }
@@ -1,6 +1,6 @@
 //! 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 crate::et::OOB;
 use std::collections::HashMap;
 use std::net::IpAddr;
 use std::time::Duration;
@@ -1,72 +1,59 @@
-//! UDP relay for a Wolfenstein: Enemy Territory server reachable only through a tunnel.
+//! UDP relay for Wolfenstein: Enemy Territory servers reachable only through a tunnel:
-//!
+//! each server gets a public port here (see server), registered by its admin tool through
-//! - Proxies every client (players, server browsers, masters) to the ET server, one
+//! the control API (see control, registry) or given on the command line.
 //!   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.
 //! - 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::SocketAddr;
 use std::ops::RangeInclusive;
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
+use tokio::sync::Semaphore;
 use tokio::net::{UdpSocket, lookup_host};
 use tokio::sync::mpsc::{self, error::TrySendError};
 use tokio::time::{Instant, interval, sleep_until, timeout};
 /// Connectionless packet prefix (Quake 3 / ET protocol).
 const OOB: &[u8] = b"\xff\xff\xff\xff";
 /// Same heartbeat period as the ET server itself.
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(300);
 const HEARTBEAT_ALIVE: &str = "EnemyTerritory-1";
 const HEARTBEAT_DEAD: &str = "ETFlatline-1";
 const PROBE_TIMEOUT: Duration = Duration::from_secs(3);
 const PROBE_ATTEMPTS: usize = 3;
 const SESSION_TIMEOUT: Duration = Duration::from_secs(120);
 const SESSION_QUEUE: usize = 64;
 const MAX_PACKET: usize = 65_535;
 /// Longest connectionless command name shown in debug logs.
 const MAX_COMMAND: usize = 32;
 /// 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);
-/// println! only with --debug.
+/// println! only with --debug. Before the modules: they use it.
 macro_rules! debug {
    ($($arg:tt)*) => {
-        if DEBUG.load(Ordering::Relaxed) {
+        if $crate::DEBUG.load(std::sync::atomic::Ordering::Relaxed) {
            println!($($arg)*);
        }
    };
 }
-/// UDP relay + master heartbeat for a Wolfenstein: Enemy Territory server.
+mod bans;
 mod control;
 mod et;
 mod filter;
 mod proof;
 mod registry;
 mod server;
 use registry::Registry;
 use server::Common;
 /// One upstream socket per session, all servers together: 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;
 /// UDP relay + master heartbeat for Wolfenstein: Enemy Territory servers.
 #[derive(Parser)]
 #[command(version, name = "etrelay")]
 struct Cli {
-    /// ET server address, e.g. 10.0.0.2:27960
+    /// An ET server relayed on --listen from the start, e.g. 10.0.0.2:27960: no control
-    upstream: SocketAddr,
+    /// API reaches it. Without it, servers register through --control
    upstream: Option<SocketAddr>,
-    /// Public address to listen on
+    /// Public address: UPSTREAM is relayed on it, registered servers on its IP
    #[arg(long, default_value = "0.0.0.0:27960")]
    listen: SocketAddr,
    /// Public ports a server can register on, e.g. 27960-27969 (or one port)
    #[arg(long, default_value = "27960-27969", value_parser = parse_ports)]
    ports: RangeInclusive<u16>,
    /// Master server host:port (repeatable)
    #[arg(
        long = "master",
@@ -74,12 +61,12 @@ struct Cli {
    )]
    masters: Vec<String>,
-    /// Control API address (session ports, bans), on the tunnel only, e.g. 10.66.0.1:8090;
+    /// Control API address (registering, session ports, bans), on the tunnel only, e.g.
-    /// bearer token in $ETRELAY_CONTROL_TOKEN
+    /// 10.66.0.1:8090; the token to register with in $ETRELAY_CONTROL_TOKEN
    #[arg(long)]
    control: Option<SocketAddr>,
-    /// Folder of the bans file (bans.json)
+    /// Folder of the registered servers (servers.json) and of the bans
    #[arg(long, default_value = "/var/lib/etrelay")]
    state_dir: PathBuf,
@@ -88,268 +75,76 @@ struct Cli {
    debug: bool,
 }
-/// What the relay loop and the control API share.
+fn parse_ports(text: &str) -> Result<RangeInclusive<u16>, String> {
-pub struct Relay {
+    let (first, last) = text.split_once('-').unwrap_or((text, text));
-    pub sessions: HashMap<SocketAddr, Session>,
+    let port = |p: &str| p.trim().parse::<u16>().map_err(|e| format!("{p}: {e}"));
-    pub bans: Bans,
+    let (first, last) = (port(first)?, port(last)?);
 }
-pub struct Session {
+    if first == 0 || first > last {
-    /// Closed once the session ended; dropping it ends the session.
+        return Err(format!("{text}: want <first>-<last>"));
-    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>>;
+    Ok(first..=last)
 }
 #[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"))?;
+    if cli.upstream.is_none() && cli.control.is_none() {
-    let shared: Shared = Arc::new(Mutex::new(Relay {
+        bail!(
-        sessions: HashMap::new(),
+            "nothing to relay: give an ET server (UPSTREAM) or --control for servers to register"
-        bans,
+        );
-    }));
+    }
    let common = Common {
        masters: Arc::new(cli.masters),
        sessions: Arc::new(Semaphore::new(MAX_SESSIONS)),
    };
    println!("etrelay: masters: {}", common.masters.join(", "));
    let mut registry = Registry::new(cli.state_dir, cli.listen.ip(), cli.ports.clone(), common);
    if let Some(upstream) = cli.upstream {
        registry.add_static(cli.listen, upstream)?;
    }
    let Some(control) = cli.control else {
        // Only the command line's server: its tasks do everything.
        return std::future::pending().await;
    };
    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()));
+    registry
-    }
+        .restore()
-
+        .context("servers registered before the restart")?;
    let public = UdpSocket::bind(cli.listen)
        .await
        .with_context(|| format!("bind {}", cli.listen))?;
    let public = Arc::new(public);
    println!(
-        "etrelay: {} -> {}, masters: {}",
+        "servers register on {} ports {}-{}",
-        cli.listen,
+        cli.listen.ip(),
-        cli.upstream,
+        cli.ports.start(),
-        cli.masters.join(", ")
+        cli.ports.end()
    );
-    tokio::spawn(heartbeat(public.clone(), cli.upstream, cli.masters));
+    control::serve(control, token, Arc::new(Mutex::new(registry))).await;
-    relay(public, cli.upstream, shared).await
+
    Ok(())
 }
-/// Dispatches client packets to their session, opening one on first packet.
+#[cfg(test)]
-async fn relay(public: Arc<UdpSocket>, upstream: SocketAddr, shared: Shared) -> Result<()> {
+mod tests {
-    let mut filter = Filter::new(Instant::now());
+    use super::*;
    let mut buf = vec![0u8; MAX_PACKET];
-    loop {
+    #[test]
-        let (n, client) = public.recv_from(&mut buf).await.context("recv")?;
+    fn ports_are_a_range_or_one_port() {
-
+        assert_eq!(parse_ports("27960-27969"), Ok(27960..=27969));
-        if let Some(reason) = filter.refuse(&buf[..n], client.ip(), Instant::now()) {
+        assert_eq!(parse_ports("27960"), Ok(27960..=27960));
-            debug!("{client} dropped: {reason}");
+        assert!(parse_ports("27969-27960").is_err());
-            continue;
+        assert!(parse_ports("0-10").is_err());
-        }
+        assert!(parse_ports("a-b").is_err());
        let mut packet = buf[..n].to_vec();
        // 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;
            }
        }
        match open_session(public.clone(), upstream, client).await {
            Ok((tx, port)) => {
                let _ = tx.try_send(packet);
                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, 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>>, 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! {
                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}");
                    }
                    sent += 1;
                    deadline = Instant::now() + SESSION_TIMEOUT;
                }
                res = sock.recv(&mut buf) => match res {
                    Ok(n) => {
                        if let Some(command) = command(&buf[..n]) {
                            debug!("{client} <- ET  {command} ({n} B)");
                        }
                        let _ = public.send_to(&buf[..n], client).await;
                        received += 1;
                        deadline = Instant::now() + SESSION_TIMEOUT;
                    }
                    // ICMP unreachable while the ET server is down: keep waiting.
                    Err(e) => debug!("{client} <- ET  recv failed: {e}"),
                },
                _ = sleep_until(deadline) => break,
            }
        }
        debug!("session - {client}: {sent} packets to ET, {received} from ET");
    });
    Ok((tx, port))
 }
 /// Heartbeats masters while the ET server answers, sends one flatline when it stops.
 async fn heartbeat(public: Arc<UdpSocket>, upstream: SocketAddr, masters: Vec<String>) {
    let mut ticker = interval(HEARTBEAT_INTERVAL);
    let mut alive = false;
    loop {
        ticker.tick().await;
        let up = probe(upstream).await;
        debug!("probe {upstream}: {}", if up { "up" } else { "no answer" });
        if up != alive {
            println!("ET server {upstream} is {}", if up { "up" } else { "down" });
        }
        let message = match (up, alive) {
            (true, _) => HEARTBEAT_ALIVE,
            (false, true) => HEARTBEAT_DEAD,
            (false, false) => continue,
        };
        alive = up;
        let packet = [OOB, format!("heartbeat {message}\n").as_bytes()].concat();
        for master in &masters {
            match send_to_host(&public, &packet, master).await {
                Ok(addr) => debug!("heartbeat {message} -> {master} ({addr})"),
                Err(e) => eprintln!("heartbeat {master}: {e:#}"),
            }
        }
    }
 }
 /// True if the ET server answers `getinfo`.
 async fn probe(upstream: SocketAddr) -> bool {
    for _ in 0..PROBE_ATTEMPTS {
        let attempt = async {
            let sock = connect(upstream).await?;
            sock.send(&[OOB, b"getinfo etrelay"].concat()).await?;
            let mut buf = vec![0u8; MAX_PACKET];
            let n = sock.recv(&mut buf).await?;
            Ok::<_, anyhow::Error>(buf[..n].starts_with(&[OOB, b"infoResponse"].concat()))
        };
        if let Ok(Ok(true)) = timeout(PROBE_TIMEOUT, attempt).await {
            return true;
        }
    }
    false
 }
 /// Resolves `host` (DNS may change between heartbeats) and sends from `sock`.
 async fn send_to_host(sock: &UdpSocket, packet: &[u8], host: &str) -> Result<SocketAddr> {
    let ipv4 = sock.local_addr()?.is_ipv4();
    let addr = lookup_host(host)
        .await?
        .find(|a| a.is_ipv4() == ipv4)
        .context("no address in the listen socket family")?;
    sock.send_to(packet, addr).await?;
    Ok(addr)
 }
 /// 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> {
    let body = packet.strip_prefix(OOB)?;
    let end = body
        .iter()
        .position(|b| b.is_ascii_whitespace() || *b == b'\\')
        .unwrap_or(body.len())
        .min(MAX_COMMAND);
    Some(String::from_utf8_lossy(&body[..end]).into_owned())
 }
 /// Ephemeral socket connected to `upstream`.
 async fn connect(upstream: SocketAddr) -> Result<UdpSocket> {
    let any: SocketAddr = if upstream.is_ipv4() {
        "0.0.0.0:0".parse()?
    } else {
        "[::]:0".parse()?
    };
    let sock = UdpSocket::bind(any).await?;
    sock.connect(upstream).await?;
    Ok(sock)
 }
@@ -0,0 +1,121 @@
 //! Pairing keys, and the proof an admin tool gives to register an ET server: it picks a
 //! key, makes the server show the key's fingerprint in a cvar anyone can read
 //! (`getstatus`), then registers with the key. Only who runs that server's console can do
 //! it, and the fingerprint tells nothing about the key.
 use crate::et;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::fmt;
 use std::net::SocketAddr;
 /// The serverinfo cvar holding the fingerprint (`sets etrelay_proof <fingerprint>`).
 pub const PROOF_CVAR: &str = "etrelay_proof";
 /// Shortest key accepted.
 pub const MIN_KEY: usize = 32;
 /// Bytes of the hash shown as the fingerprint: the serverinfo string is short (1024
 /// characters for every cvar), 128 bits are plenty.
 const PROOF_BYTES: usize = 16;
 /// SHA-256 of a pairing key: what the relay keeps and compares, never the key itself.
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(into = "String", try_from = "String")]
 pub struct KeyHash([u8; 32]);
 impl KeyHash {
    pub fn of(key: &str) -> Self {
        Self(Sha256::digest(key.as_bytes()).into())
    }
    /// What the ET server must show in PROOF_CVAR.
    pub fn proof(&self) -> String {
        hex(&self.0[..PROOF_BYTES])
    }
 }
 impl From<KeyHash> for String {
    fn from(hash: KeyHash) -> Self {
        hex(&hash.0)
    }
 }
 impl TryFrom<String> for KeyHash {
    type Error = String;
    fn try_from(text: String) -> Result<Self, Self::Error> {
        let bad = || format!("not a key hash: {text}");
        if text.len() != 64 || !text.is_ascii() {
            return Err(bad());
        }
        let mut hash = [0u8; 32];
        for (i, byte) in hash.iter_mut().enumerate() {
            *byte = u8::from_str_radix(&text[2 * i..2 * i + 2], 16).map_err(|_| bad())?;
        }
        Ok(Self(hash))
    }
 }
 fn hex(bytes: &[u8]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
 }
 #[derive(Debug, PartialEq)]
 pub enum Missing {
    /// The ET server didn't answer `getstatus`.
    NoAnswer(SocketAddr),
    /// It answered without the fingerprint of the key.
    NotShown(SocketAddr),
 }
 impl fmt::Display for Missing {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::NoAnswer(upstream) => write!(f, "the ET server {upstream} doesn't answer"),
            Self::NotShown(upstream) => write!(
                f,
                "the ET server {upstream} doesn't show the key's fingerprint in {PROOF_CVAR}"
            ),
        }
    }
 }
 /// Checks the ET server at `upstream` shows `key`'s fingerprint.
 pub async fn shown(upstream: SocketAddr, key: &KeyHash) -> Result<(), Missing> {
    let cvars = et::serverinfo(upstream)
        .await
        .ok_or(Missing::NoAnswer(upstream))?;
    match cvars.get(PROOF_CVAR) {
        Some(value) if value.eq_ignore_ascii_case(&key.proof()) => Ok(()),
        _ => Err(Missing::NotShown(upstream)),
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn fingerprint_is_the_start_of_the_hash() {
        // sha256("abc")
        let hash = KeyHash::of("abc");
        assert_eq!(hash.proof(), "ba7816bf8f01cfea414140de5dae2223");
        assert_eq!(
            String::from(hash.clone()),
            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"
        );
    }
    #[test]
    fn hash_survives_its_text_form() {
        let hash = KeyHash::of("some key");
        let text = String::from(hash.clone());
        assert!(KeyHash::try_from(text) == Ok(hash));
        assert!(KeyHash::try_from("zz".repeat(32)).is_err());
        assert!(KeyHash::try_from("abcd".to_string()).is_err());
    }
 }
@@ -0,0 +1,454 @@
 //! The ET servers the relay holds, one per public port: registered through the control
 //! API (kept in servers.json, back after a restart) or given on the command line.
 use crate::bans::{self, Bans};
 use crate::proof::KeyHash;
 use crate::server::{Common, Server};
 use anyhow::{Context, Result};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::fmt;
 use std::net::{IpAddr, SocketAddr};
 use std::ops::RangeInclusive;
 use std::path::PathBuf;
 use std::sync::{Arc, Mutex};
 use tokio::net::UdpSocket;
 pub type Shared = Arc<Mutex<Registry>>;
 pub struct Registry {
    /// Folder of servers.json and of each server's bans.
    dir: PathBuf,
    /// Public address the servers' ports are bound on.
    listen: IpAddr,
    /// Public ports a server can register on.
    ports: RangeInclusive<u16>,
    common: Common,
    entries: HashMap<u16, Entry>,
 }
 struct Entry {
    server: Arc<Server>,
    /// None for the command line's server: it has no key, nothing reaches it through the
    /// control API.
    pairing: Option<Pairing>,
 }
 /// Who registered a server.
 #[derive(Clone)]
 struct Pairing {
    name: String,
    key: KeyHash,
    /// Unix seconds.
    since: u64,
 }
 /// A registered server as servers.json keeps it.
 #[derive(Serialize, Deserialize)]
 struct Record {
    port: u16,
    upstream: SocketAddr,
    name: String,
    key: KeyHash,
    since: u64,
 }
 /// What an admin tool asks for; the key's proof is checked before (see proof).
 pub struct Registration {
    pub port: u16,
    pub upstream: SocketAddr,
    pub name: String,
    pub key: KeyHash,
 }
 /// A registered server as the control API shows it.
 #[derive(Serialize)]
 pub struct View {
    pub port: u16,
    pub upstream: SocketAddr,
    pub name: String,
    /// Unix seconds.
    pub since: u64,
    /// The ET server answered the last probe: the masters are told about it.
    pub alive: bool,
 }
 #[derive(Debug)]
 pub enum Refused {
    /// Not a port servers can register on.
    Port(u16, RangeInclusive<u16>),
    /// Another ET server holds the port.
    Taken(u16),
    /// The ET server is the command line's: it isn't registered.
    Static(SocketAddr),
    Failed(anyhow::Error),
 }
 impl fmt::Display for Refused {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Port(port, ports) => write!(
                f,
                "port {port} is not offered: {}-{}",
                ports.start(),
                ports.end()
            ),
            Self::Taken(port) => write!(f, "port {port} is taken by another server"),
            Self::Static(upstream) => write!(
                f,
                "{upstream} is the relay's own server (command line), it can't be registered"
            ),
            Self::Failed(e) => write!(f, "{e:#}"),
        }
    }
 }
 impl Registry {
    pub fn new(dir: PathBuf, listen: IpAddr, ports: RangeInclusive<u16>, common: Common) -> Self {
        Self {
            dir,
            listen,
            ports,
            common,
            entries: HashMap::new(),
        }
    }
    /// Starts the servers registered before the restart.
    pub fn restore(&mut self) -> Result<()> {
        let path = self.file();
        let records: Vec<Record> = match std::fs::read(&path) {
            Ok(data) => serde_json::from_slice(&data)
                .with_context(|| format!("{}: not a servers file", path.display()))?,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Vec::new(),
            Err(e) => return Err(e).with_context(|| format!("read {}", path.display())),
        };
        for r in records {
            let pairing = Pairing {
                name: r.name,
                key: r.key,
                since: r.since,
            };
            self.open(
                SocketAddr::new(self.listen, r.port),
                r.upstream,
                Some(pairing),
            )?;
        }
        Ok(())
    }
    /// The command line's server, relayed on `listen`: never kept, never reachable through
    /// the control API. Its bans (bans.json) are the ones set before servers registered.
    pub fn add_static(&mut self, listen: SocketAddr, upstream: SocketAddr) -> Result<()> {
        self.open(listen, upstream, None)
    }
    /// Whether `upstream` could register on `port`: checked before asking the ET server
    /// for the proof, and again by register.
    pub fn check(&self, port: u16, upstream: SocketAddr) -> Result<(), Refused> {
        if !self.ports.contains(&port) {
            return Err(Refused::Port(port, self.ports.clone()));
        }
        if self
            .entries
            .values()
            .any(|e| e.pairing.is_none() && e.server.upstream == upstream)
        {
            return Err(Refused::Static(upstream));
        }
        match self.entries.get(&port) {
            Some(e) if e.pairing.is_none() || e.server.upstream != upstream => {
                Err(Refused::Taken(port))
            }
            _ => Ok(()),
        }
    }
    /// Registers an ET server on a public port. An ET server registered already is taken
    /// over: who proved to run it holds it now, the former key is gone. On the same port
    /// its sessions and bans go on, on another one its former port is freed.
    pub fn register(&mut self, reg: Registration) -> Result<View, Refused> {
        self.check(reg.port, reg.upstream)?;
        let pairing = Pairing {
            name: reg.name,
            key: reg.key,
            since: bans::now(),
        };
        let moved: Vec<u16> = self
            .entries
            .iter()
            .filter(|(port, e)| e.server.upstream == reg.upstream && **port != reg.port)
            .map(|(port, _)| *port)
            .collect();
        for port in moved {
            self.close(port).map_err(Refused::Failed)?;
        }
        match self.entries.get_mut(&reg.port) {
            Some(entry) => entry.pairing = Some(pairing),
            None => self
                .open(
                    SocketAddr::new(self.listen, reg.port),
                    reg.upstream,
                    Some(pairing),
                )
                .map_err(Refused::Failed)?,
        }
        self.save().map_err(Refused::Failed)?;
        self.entries[&reg.port].server.answered();
        println!("registered {} on port {}", reg.upstream, reg.port);
        Ok(self.view(reg.port).expect("just registered"))
    }
    /// Stops relaying the server of `port` and forgets it, its bans too.
    pub fn unregister(&mut self, port: u16) -> Result<()> {
        self.close(port)?;
        self.save()?;
        println!("unregistered port {port}");
        Ok(())
    }
    /// The server registered with `key`.
    pub fn find(&self, key: &KeyHash) -> Option<Arc<Server>> {
        self.entries
            .values()
            .find(|e| e.pairing.as_ref().is_some_and(|p| p.key == *key))
            .map(|e| e.server.clone())
    }
    /// None for a free port and for the command line's server.
    pub fn view(&self, port: u16) -> Option<View> {
        let entry = self.entries.get(&port)?;
        let pairing = entry.pairing.as_ref()?;
        Some(View {
            port,
            upstream: entry.server.upstream,
            name: pairing.name.clone(),
            since: pairing.since,
            alive: entry.server.alive(),
        })
    }
    /// Binds the public port and starts relaying it.
    fn open(
        &mut self,
        listen: SocketAddr,
        upstream: SocketAddr,
        pairing: Option<Pairing>,
    ) -> Result<()> {
        let bans = Bans::load(self.bans_file(listen.port(), pairing.is_some()))?;
        // std's bind: no await, the caller holds the registry's lock.
        let socket = std::net::UdpSocket::bind(listen).with_context(|| format!("bind {listen}"))?;
        socket.set_nonblocking(true)?;
        let socket = UdpSocket::from_std(socket)?;
        let server = Server::start(socket, upstream, bans, self.common.clone())?;
        println!("relaying {listen} -> {upstream}");
        self.entries.insert(server.port, Entry { server, pairing });
        Ok(())
    }
    fn close(&mut self, port: u16) -> Result<()> {
        let Some(entry) = self.entries.remove(&port) else {
            return Ok(());
        };
        entry.server.stop();
        let bans = self.bans_file(port, entry.pairing.is_some());
        match std::fs::remove_file(&bans) {
            Err(e) if e.kind() != std::io::ErrorKind::NotFound => {
                Err(e).with_context(|| format!("remove {}", bans.display()))
            }
            _ => Ok(()),
        }
    }
    fn file(&self) -> PathBuf {
        self.dir.join("servers.json")
    }
    /// Each registered server has its own bans; bans.json is the command line's server's.
    fn bans_file(&self, port: u16, registered: bool) -> PathBuf {
        if !registered {
            return self.dir.join("bans.json");
        }
        self.dir.join(format!("bans-{port}.json"))
    }
    /// Written next to the file then renamed: a crash never leaves half a list.
    fn save(&self) -> Result<()> {
        let mut records: Vec<Record> = self
            .entries
            .iter()
            .filter_map(|(port, e)| {
                let p = e.pairing.as_ref()?;
                Some(Record {
                    port: *port,
                    upstream: e.server.upstream,
                    name: p.name.clone(),
                    key: p.key.clone(),
                    since: p.since,
                })
            })
            .collect();
        records.sort_by_key(|r| r.port);
        let path = self.file();
        let tmp = path.with_extension("tmp");
        let data = serde_json::to_vec_pretty(&records)?;
        std::fs::write(&tmp, data).with_context(|| format!("write {}", tmp.display()))?;
        std::fs::rename(&tmp, &path).with_context(|| format!("rename to {}", path.display()))
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    use std::net::Ipv4Addr;
    use tokio::sync::Semaphore;
    const LOCAL: IpAddr = IpAddr::V4(Ipv4Addr::LOCALHOST);
    /// A registry on its own folder and ports (tests run side by side).
    fn registry(name: &str, base: u16) -> Registry {
        let dir = std::env::temp_dir().join(format!("etrelay-test-{}-{name}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).unwrap();
        again(dir, base)
    }
    fn again(dir: PathBuf, base: u16) -> Registry {
        let common = Common {
            masters: Arc::new(Vec::new()),
            sessions: Arc::new(Semaphore::new(8)),
        };
        Registry::new(dir, LOCAL, base..=base + 1, common)
    }
    fn upstream(port: u16) -> SocketAddr {
        SocketAddr::new(LOCAL, port)
    }
    fn reg(port: u16, upstream: SocketAddr, key: &str) -> Registration {
        Registration {
            port,
            upstream,
            name: "test".into(),
            key: KeyHash::of(key),
        }
    }
    #[tokio::test]
    async fn one_server_per_port() {
        let mut r = registry("ports", 47100);
        let (a, b) = (upstream(47110), upstream(47111));
        assert!(matches!(
            r.register(reg(47105, a, "key a")),
            Err(Refused::Port(47105, _))
        ));
        let view = r.register(reg(47100, a, "key a")).unwrap();
        assert_eq!((view.port, view.upstream), (47100, a));
        // Another server on a taken port: refused, the first one stays.
        assert!(matches!(
            r.register(reg(47100, b, "key b")),
            Err(Refused::Taken(47100))
        ));
        assert!(r.find(&KeyHash::of("key a")).is_some());
        assert!(r.find(&KeyHash::of("key b")).is_none());
        r.register(reg(47101, b, "key b")).unwrap();
        assert_eq!(r.find(&KeyHash::of("key b")).unwrap().port, 47101);
    }
    #[tokio::test]
    async fn a_server_registered_again_is_taken_over() {
        let mut r = registry("takeover", 47120);
        let a = upstream(47130);
        r.register(reg(47120, a, "old key")).unwrap();
        let server = r.find(&KeyHash::of("old key")).unwrap();
        // Same port: the same relay goes on, under the new key only.
        r.register(reg(47120, a, "new key")).unwrap();
        assert!(r.find(&KeyHash::of("old key")).is_none());
        assert!(Arc::ptr_eq(
            &server,
            &r.find(&KeyHash::of("new key")).unwrap()
        ));
        // Another port: the former one is freed.
        r.register(reg(47121, a, "new key")).unwrap();
        assert!(r.view(47120).is_none());
        assert_eq!(r.find(&KeyHash::of("new key")).unwrap().port, 47121);
    }
    #[tokio::test]
    async fn registered_servers_come_back_and_leave() {
        let mut r = registry("restore", 47140);
        let dir = r.dir.clone();
        r.register(reg(47140, upstream(47150), "key")).unwrap();
        r.unregister(47140).unwrap();
        r.register(reg(47141, upstream(47151), "key")).unwrap();
        // A restart: the port is free once the tasks holding its socket ended.
        for entry in r.entries.values() {
            entry.server.stop();
        }
        drop(r);
        tokio::task::yield_now().await;
        let mut back = again(dir.clone(), 47140);
        back.restore().unwrap_or_else(|e| panic!("{e:#}"));
        let view = back.view(47141).unwrap();
        assert_eq!(
            (view.upstream, view.name.as_str()),
            (upstream(47151), "test")
        );
        assert!(back.view(47140).is_none());
        assert!(back.find(&KeyHash::of("key")).is_some());
        back.unregister(47141).unwrap();
        assert!(again(dir, 47140).restore().is_ok());
    }
    #[tokio::test]
    async fn the_command_line_server_is_out_of_reach() {
        let mut r = registry("static", 47160);
        let own = upstream(47170);
        r.add_static(SocketAddr::new(LOCAL, 47160), own).unwrap();
        assert!(r.view(47160).is_none());
        assert!(matches!(
            r.register(reg(47161, own, "key")),
            Err(Refused::Static(_))
        ));
        assert!(matches!(
            r.register(reg(47160, upstream(47171), "key")),
            Err(Refused::Taken(47160))
        ));
    }
 }
@@ -0,0 +1,317 @@
 //! One relayed ET server: its public socket, its clients' sessions, its bans and its
 //! master heartbeats.
 //!
 //! - Proxies every client (players, server browsers, masters) to the ET server, one
 //!   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.
 //! - Drops what a client has no business sending (see filter) and banned addresses.
 use crate::bans::{self, Bans};
 use crate::et::{self, MAX_PACKET, OOB};
 use crate::filter::Filter;
 use anyhow::{Context, Result};
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex, MutexGuard};
 use std::time::Duration;
 use tokio::net::{UdpSocket, lookup_host};
 use tokio::sync::mpsc::{self, error::TrySendError};
 use tokio::sync::{OwnedSemaphorePermit, Semaphore};
 use tokio::task::AbortHandle;
 use tokio::time::{Instant, interval, sleep, sleep_until};
 /// Same heartbeat period as the ET server itself.
 const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(300);
 const HEARTBEAT_ALIVE: &str = "EnemyTerritory-1";
 const HEARTBEAT_DEAD: &str = "ETFlatline-1";
 const SESSION_TIMEOUT: Duration = Duration::from_secs(120);
 const SESSION_QUEUE: usize = 64;
 /// A public socket that fails is read again after this, not in a tight loop.
 const RECV_RETRY: Duration = Duration::from_millis(100);
 /// What every server of the relay shares.
 #[derive(Clone)]
 pub struct Common {
    pub masters: Arc<Vec<String>>,
    /// One permit per open session, whatever its server: each holds a socket, spoofed
    /// sources must not use up the file descriptors (see MAX_SESSIONS).
    pub sessions: Arc<Semaphore>,
 }
 pub struct Server {
    /// Public port: where clients and masters reach this server.
    pub port: u16,
    /// The ET server, as seen from here.
    pub upstream: SocketAddr,
    socket: Arc<UdpSocket>,
    common: Common,
    state: Mutex<State>,
    /// The ET server answered the last probe.
    alive: AtomicBool,
    tasks: Mutex<Vec<AbortHandle>>,
 }
 /// What the relay loop and the control API share. Never held across an await.
 pub struct State {
    pub sessions: HashMap<SocketAddr, Session>,
    pub bans: Bans,
 }
 pub struct Session {
    /// Closed once the session ended; dropping it ends the session.
    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,
 }
 impl Session {
    pub fn open(&self) -> bool {
        !self.tx.is_closed()
    }
 }
 impl Server {
    /// Relays `socket` (bound to the public port) to `upstream` and heartbeats the
    /// masters, until stop.
    pub fn start(
        socket: UdpSocket,
        upstream: SocketAddr,
        bans: Bans,
        common: Common,
    ) -> Result<Arc<Self>> {
        let server = Arc::new(Self {
            port: socket.local_addr().context("public socket")?.port(),
            upstream,
            socket: Arc::new(socket),
            common,
            state: Mutex::new(State {
                sessions: HashMap::new(),
                bans,
            }),
            alive: AtomicBool::new(false),
            tasks: Mutex::new(Vec::new()),
        });
        let relay = tokio::spawn(server.clone().relay());
        let heartbeat = tokio::spawn(server.clone().heartbeat());
        *server.tasks.lock().expect("server tasks") =
            vec![relay.abort_handle(), heartbeat.abort_handle()];
        Ok(server)
    }
    /// Stops relaying: sessions are cut, the masters told when they listed the server.
    /// The public port is free once the flatline left.
    pub fn stop(&self) {
        for task in self.tasks.lock().expect("server tasks").drain(..) {
            task.abort();
        }
        self.state().sessions.clear();
        if self.alive.swap(false, Ordering::Relaxed) {
            let (socket, masters) = (self.socket.clone(), self.common.masters.clone());
            tokio::spawn(async move { announce(&socket, &masters, HEARTBEAT_DEAD).await });
        }
    }
    pub fn state(&self) -> MutexGuard<'_, State> {
        self.state.lock().expect("server state")
    }
    pub fn alive(&self) -> bool {
        self.alive.load(Ordering::Relaxed)
    }
    /// The ET server just answered (the proof of a registration): no need to wait for
    /// the next probe to say so.
    pub fn answered(&self) {
        self.alive.store(true, Ordering::Relaxed);
    }
    /// Dispatches client packets to their session, opening one on first packet.
    async fn relay(self: Arc<Self>) {
        let mut filter = Filter::new(Instant::now());
        let mut buf = vec![0u8; MAX_PACKET];
        loop {
            let (n, client) = match self.socket.recv_from(&mut buf).await {
                Ok(received) => received,
                Err(e) => {
                    eprintln!("port {}: recv: {e}", self.port);
                    sleep(RECV_RETRY).await;
                    continue;
                }
            };
            if let Some(reason) = filter.refuse(&buf[..n], client.ip(), Instant::now()) {
                debug!("{client} dropped: {reason}");
                continue;
            }
            let mut packet = buf[..n].to_vec();
            {
                let mut state = self.state();
                if state.bans.banned(client.ip(), bans::now()) {
                    debug!("{client} dropped: banned");
                    continue;
                }
                if let Some(session) = state.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,
                    }
                }
                state.sessions.retain(|_, s| s.open());
            }
            let Ok(permit) = self.common.sessions.clone().try_acquire_owned() else {
                debug!("{client} dropped: too many sessions open");
                continue;
            };
            match self.open_session(client, permit).await {
                Ok(session) => {
                    let _ = session.tx.try_send(packet);
                    let port = session.port;
                    let mut state = self.state();
                    state.sessions.insert(client, session);
                    debug!(
                        "session + {client} as port {port} ({} active on port {})",
                        state.sessions.len(),
                        self.port
                    );
                }
                Err(e) => eprintln!("session {client}: {e:#}"),
            }
        }
    }
    /// Forwards packets between one client and the ET server until SESSION_TIMEOUT of
    /// silence, or until its sender is dropped (ban, stop). `permit` is given back then.
    async fn open_session(
        &self,
        client: SocketAddr,
        permit: OwnedSemaphorePermit,
    ) -> Result<Session> {
        let sock = et::connect(self.upstream).await?;
        let port = sock.local_addr()?.port();
        let (tx, mut rx) = mpsc::channel::<Vec<u8>>(SESSION_QUEUE);
        let public = self.socket.clone();
        tokio::spawn(async move {
            let _permit = permit;
            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! {
                    packet = rx.recv() => {
                        let Some(packet) = packet else { break };
                        if let Some(command) = et::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}");
                        }
                        sent += 1;
                        deadline = Instant::now() + SESSION_TIMEOUT;
                    }
                    res = sock.recv(&mut buf) => match res {
                        Ok(n) => {
                            if let Some(command) = et::command(&buf[..n]) {
                                debug!("{client} <- ET  {command} ({n} B)");
                            }
                            let _ = public.send_to(&buf[..n], client).await;
                            received += 1;
                            deadline = Instant::now() + SESSION_TIMEOUT;
                        }
                        // ICMP unreachable while the ET server is down: keep waiting.
                        Err(e) => debug!("{client} <- ET  recv failed: {e}"),
                    },
                    _ = sleep_until(deadline) => break,
                }
            }
            debug!("session - {client}: {sent} packets to ET, {received} from ET");
        });
        Ok(Session {
            tx,
            port,
            since: bans::now(),
        })
    }
    /// Heartbeats masters while the ET server answers, sends one flatline when it stops.
    async fn heartbeat(self: Arc<Self>) {
        let mut ticker = interval(HEARTBEAT_INTERVAL);
        let upstream = self.upstream;
        loop {
            ticker.tick().await;
            let up = et::probe(upstream).await;
            let alive = self.alive.swap(up, Ordering::Relaxed);
            debug!("probe {upstream}: {}", if up { "up" } else { "no answer" });
            if up != alive {
                println!("ET server {upstream} is {}", if up { "up" } else { "down" });
            }
            match (up, alive) {
                (true, _) => announce(&self.socket, &self.common.masters, HEARTBEAT_ALIVE).await,
                (false, true) => announce(&self.socket, &self.common.masters, HEARTBEAT_DEAD).await,
                (false, false) => {}
            }
        }
    }
 }
 /// Sends `heartbeat <message>` to every master from `socket`: the masters check the
 /// address it came from.
 async fn announce(socket: &UdpSocket, masters: &[String], message: &str) {
    let packet = [OOB, format!("heartbeat {message}\n").as_bytes()].concat();
    for master in masters {
        match send_to_host(socket, &packet, master).await {
            Ok(addr) => debug!("heartbeat {message} -> {master} ({addr})"),
            Err(e) => eprintln!("heartbeat {master}: {e:#}"),
        }
    }
 }
 /// Resolves `host` (DNS may change between heartbeats) and sends from `sock`.
 async fn send_to_host(sock: &UdpSocket, packet: &[u8], host: &str) -> Result<SocketAddr> {
    let ipv4 = sock.local_addr()?.is_ipv4();
    let addr = lookup_host(host)
        .await?
        .find(|a| a.is_ipv4() == ipv4)
        .context("no address in the listen socket family")?;
    sock.send_to(packet, addr).await?;
    Ok(addr)
 }