diff --git a/AGENTS.md b/AGENTS.md index ed3170b..ae32137 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -140,6 +140,10 @@ Client address transparency is preserved via `transparent::connect_as_udp` using What UDP keeps: per-flow filter state keyed by client address and released after `IDLE` — UDP has no close to observe, so a timeout is the only thing that ends a flow — and both directions inspected. `Verdict::Reject` drops the datagram rather than closing anything, because there is nothing to close. Replies leave through the **listener** socket so conntrack rewrites them to appear from the address the client dialled; sending from the upstream socket would reach a client not expecting that source. The relay's own dial carries `SELF_MARK`, or the redirect rule would catch it and loop. A flow outlives its service restarting: a datagram sent while the service was down comes back as ICMP "port unreachable", which the flow's connected socket reports once as `ConnectionRefused`, and `replies` carries on past it. It used to end the task reading the service's answers while the flow stayed in the table, and a client that kept talking kept the flow from ever going idle — so it never heard from the service again, while a new client did (`tests/udp.rs::a_flow_outlives_its_service_restarting`). +**Each flow's datagrams are judged by a task of the flow's own** (`forward` in `udp.rs`, fed through a channel of `FLOW_QUEUE`). The receive loop used to run the chain itself, one datagram after another for every client of the address, so a filter taking a second over one client's datagram was a second in which no other client's was read, and the socket buffer dropped theirs behind it. The loop only dispatches now; within one flow the order is kept, and a flow sending faster than it is judged has its own datagrams dropped past the queue, as the socket buffer would. Pinned by `tests/udp.rs::a_slow_verdict_for_one_client_does_not_hold_up_another` and `one_flows_datagrams_keep_their_order`. + +**UDP flows are forgotten in conntrack whenever the rules that steer them change** (`modules/services/conntrack.py`, ctnetlink over `NETLINK_NETFILTER` — no `conntrack` binary, no dependency). Conntrack translates a flow at its first packet and a UDP flow lasts for as long as datagrams keep arriving, so on this layer the entry decided where each datagram went: a flow older than the service went on straight to it, **unfiltered, for as long as it kept talking** (anybody could open one before the protection went up); one carried across a restart stayed with the engine being retired, under its old filters, keeping that engine alive until the drain gave up; and once that engine was gone the entry sent every datagram to a relay port nobody held, refreshing itself each time, until the client happened to fall silent for two minutes. `ServiceManager._claim_udp` runs after the redirect is installed (start, address added, rules put back), `_release_udp` after it is taken off on a stop (not on a restart, whose start half claims them a moment later), and `_engine_gone` once an engine has exited, for the flows still pointed at its relay ports. A datagram address forgets every flow headed for it (`moved`): changing engines between two datagrams costs a flow nothing but the filter's own state. A **QUIC** address forgets only what nothing is carrying (`unsteered`: never redirected, or redirected to a port nobody is bound to, read from `/proc/net/udp{,6}`), because a QUIC connection moved to another engine ends. The engine's own dials (`SELF_MARK` on the conntrack entry) are never touched, nor is anything TCP. A failure is logged and changes nothing else. `tests/integration/test_udp.py` holds a flow open across a start, a restart and a killed engine. + `ConnectionMeta.l4` carries the transport to the filter, so the library's stream and HTTP models key off what the engine says rather than a hardcoded `True` in the worker. It is an enum and not a boolean because QUIC broke the question in two: a QUIC stream is ordered and reliable, so every model built on a stream applies to it, and it rides on UDP, so every rule that matches the wire matches UDP — one flag answering both got one of them wrong whichever way it was set. The worker is sent both words (`is_tcp` for the wire, `l4` for what can be built on it) and `RawPacket` exposes `l4` and `is_stream` beside `is_tcp`; `is_stream` is what `models/tcp.py` and `models/http.py` gate on. Absent, `l4` derives from `is_tcp`, because the NFQUEUE binaries carry TCP and UDP and have nothing else to say. ### A service is a list of addresses @@ -149,7 +153,7 @@ Consequences worth keeping: - **Adding an address to a running service drops nothing.** `ServiceManager.address_added` installs only that address's rules against the datapath already running (`self._steer` holds what to point at). On TCP and TLS the proxy listener is dual-stack from the start, and on UDP new relays are bound dynamically on the fly, so adding any address (IPv4, IPv6, UDP, TCP, TLS) never drops connections or restarts the service. - **`address_removed` drops the address from `self.srv.addresses` itself** rather than re-reading the table, because the caller takes the rules back *before* deleting the row. Re-reading would put it straight back, and the next restart would re-protect an address that no longer exists — with TLS, by claiming a port for a stream nobody owns, which then refuses the next service that wants it. - **`proto` is denormalised onto each address** and rewritten when the service's changes. That is what makes `(ip, port, proto)` a usable uniqueness key, so a TCP service and a UDP one can share an address exactly as the kernel allows. -- **The `external` proxy endpoint belongs to the address, not the service**, and a partial unique index forbids two addresses sharing one: the return rule recognises the operator's proxy by address and port to put the original port back, so two addresses behind one endpoint could not be told apart on the way out. +- **The `external` proxy endpoint belongs to the address, not the service**, and a partial unique index forbids two addresses sharing one: the return rule recognises the operator's proxy by address and port to put the original port back, so two addresses behind one endpoint could not be told apart on the way out. The endpoint is the hand-off layer's alone: `edit_service` clears it from every address of a service moving to another layer, or it stayed in the index read by nothing and refused another service's hand-off. `_hijack_ip` refuses one that is not a single address in the family of the address it stands in for, at the moment the row is written. - **The proxy listener is dual-stack whenever the host supports IPv6** (`[::]:0`), and `transparent::unmap` turns the v4-mapped peer and local addresses back into IPv4 before anything else in the engine touches them — conntrack, the source-preserving dial and the metadata a filter reads all want the address the packet actually had. - **Addresses accept network interface names as well as IPs.** An entry in `service_addresses` can be an IP/CIDR or a network interface name (1-15 chars, e.g. `eth0`, `wg0`, `tun0`, `lo`), checked via `utils.is_interface_name` and marked by `Address.is_interface`. The interface is not a second spelling of an address, and the interface is what the operator is choosing when the address is somebody else's to hand out. The frontend says which of the two it understood *while it is being typed* — `addressKind` in `components/InterfaceInput.tsx` puts it on a badge in the field, because `eth-0` is a perfectly good interface name and a perfectly wrong address, and the two are matched by completely different rules. That rule is written once on the frontend too, as `isInterfaceName`/`isIpAddress` in `js/utils.tsx`, reading the same charset the backend does — it used to be pasted into `AddEditService.tsx` and `AddressList.tsx`, two copies of a rule the picker beside them did not consult at all, which is how the hand-off layer came to offer interface names in the list and refuse them on submit. `InterfaceInput` lists interfaces and addresses as two groups, each interface carrying what it holds right now (one question decides between them: "and what is on it"), and stays a combobox rather than a select because an address a service will answer on need not exist on this host yet. **Link-local is not offered** — every interface carries an `fe80::` it was never configured with, which doubled the list with addresses that would put a service where no client looks for it, the same reason `udp_relay_host()` takes them last; one typed by hand is still accepted, and an interface holding nothing else says "link-local only" rather than listing it. A row is **the badge as a column, and everything else in the one beside it** — value, then the note, wrapping inside that column and never under the badge. Neither of those two is given a fixed share, because which one is long changes with the group (the addresses a link carries, or the address itself): fixing it took the whole width from a full-length IPv6 and left the other wrapping a character per line, a row a screen tall beside an address cut off mid-digit; and three items in one wrapping row put that same address at the container's left edge, under the badge, three lines tall. The list's own width is **computed from the longest value in it**, between a floor and a ceiling, and three simpler rules were wrong first: inherited from the content it collapsed to the field's width, because content allowed to wrap asks for less than it needs to stay on one line; pinned to the widest row any host could produce, every list was that wide beside fields half the size; and a floor alone put the one long address back over three lines. A host answering on `10.60.3.1` gets the floor, one carrying a full-length IPv6 gets the room to show it with its note beside it, and monospace is what makes the arithmetic honest. **The row beside them is laid out by the same arithmetic, and got it wrong the same way.** Its note (`now 10.10.0.3`, `on eth0`) is a flex item with a basis wide enough for a full-length IPv6, because a basis is what decides whether an item shares a line or drops to one of its own — at 80px it shared, and the first row with a couple of tags on it squeezed the note into the gap that was left and broke an IPv6 across three lines mid-digit. A min-width beside that basis is not the fix and was tried: on a phone the column is narrower than any floor worth setting, so the floor pushed the address past the card's edge, and it buys nothing anyway — an item only shares a line when its basis fitted, and on a line of its own it is already as wide as it can be. The row's own container grows to what the buttons leave rather than to its contents, or each row is as wide as its own longest line and two rows of one list wrap opposite ways: the one carrying a full IPv6 kept its note inline while the one saying `now 127.0.0.1` pushed it down. `overflowWrap: anywhere` stays as the last resort, which is what a 375px screen needs and nothing wider should reach. **`AddressList` reads in the same words it was chosen in** — the same badge at the same `BADGE_WIDTH`, then the value, then `on eth0` for an address and `now 10.10.0.3` for an interface — because a row an operator comes back to check, described differently from the list it came out of, is a reason to wonder whether it is the same thing. What is *shown* is `bareAddress`: the backend stores what it parsed, so `127.0.0.1` comes back as `127.0.0.1/32`, a spelling nobody types that reads like a range beside real ones; a host-length prefix is dropped for display and a real one is kept, which is also what tells `addressKind` apart. A real range survives an edit for the same reason — the modal seeded itself by cutting at the first slash, so opening "move this address" on a `/16` and saving narrowed a service to one host of it. Where interfaces are not an answer — the `external` proxy endpoint, and the service address on that layer — they are not listed at all and a name typed there is badged red, rather than offered and then refused on submit. Nftables matches use `meta iifname ` for inbound/prerouting and `meta oifname ` for outbound/postrouting/route. On `proxy` (TCP/TLS), redirect rules match `iifname`; the return path needs no per-interface rule, because it follows the conntrack mark (below). **The output hook cannot match either of those, and matches the interface's addresses instead.** `iifname` means nothing for a packet this host generates, and `oifname` is the wrong question twice over: `oifname eth0 dport 80` says *we* are calling somebody else's port 80, which is the host's own outbound traffic and none of firegex's business — and a connection from this host to one of its **own** addresses is routed through `lo` anyway, so it would never match. So `_add_proxy` resolves what the interface carries (`interface_addresses`, link-local dropped for the reason `udp_relay_host` takes it last) and writes one `daddr` rule per address. For a while it wrote nothing there at all, on the correct observation that `oifname` would hijack the wrong traffic, and the consequence was a service intercepted for the outside world and silently not for anything on the box — reported by an operator curling their own service and watching a blocked payload sail through. The cost is that an address the interface gains later is not covered *for traffic this host generates* until the service restarts; inbound keeps following the interface, and `udp_relay_host` pays the same price for the same reason. Each of those rules carries the interface's name as a **comment**, because nothing else in it does: `delete()` finds rules by what they match, and without it every one would outlive the service that installed it as a redirect to a port nobody is listening on. On UDP `proxy`, the primary IP assigned to the interface is resolved dynamically via `psutil.net_if_addrs()` to bind the relay. On `nfqueue`, interface traffic is queued natively in prerouting/postrouting without needing an IP. On `external`, interfaces are disallowed because return rewrites require a concrete IP. @@ -222,12 +226,16 @@ The firewall module is the one that had to move to get there, and the reason is - **`iptables` stopped working on the host.** `iptables-nft` refuses an entire table as soon as one rule in it cannot be expressed in iptables' own format, and it stores conntrack matches as an `xt` blob where nft writes a native `ct state`. Bisecting the rules firegex writes, `ct state` was the only one that tripped it — which is to say `allow_established` and `drop_invalid`, on in any sane configuration. From then on `iptables -L` and `iptables-save` answer ``table `filter' is incompatible, use 'nft' tool``, on a box where Docker, ufw or the organisers' scripts are the ones asking. - **Stopping firegex opened the host.** `reset()` re-added `filter INPUT` with `policy accept`, and in nftables `add chain` on an existing chain *updates* its policy. That chain was the administrator's: a host carrying default-deny had it silently removed when the firegex firewall was switched off. -A base chain carries its own policy wherever it lives, so default-deny is expressed exactly as before and a `drop` is terminal whichever table's chain reached it. **One thing about what gets blocked did change, and `allow_dnat` puts it back**: in iptables' own `FORWARD` chain, Docker's or podman's accept for a published port came before or after firegex's jump in the same chain, and either way traffic to a published container port never met firegex's policy; a base chain of our own is evaluated on its own, so a drop policy dropped every connection to a published container port — how every service on a CTF box is reached. `dnat_rules()` appends `ct status dnat accept` to `fgex_forward` after the jump (on by default, a switch in the firewall settings): the operator's forward rules still apply first, and the runtime's own chain still decides for itself. **And what the containers send**, which the first version of it left out: with the policy at drop, a web container could not reach its database on the same bridge and no container reached the network — measured with two containers on a Docker network — because Docker accepts that in `FORWARD` too, beside where firegex's rules used to be. So traffic arriving from a container runtime's bridge (`CONTAINER_BRIDGES`: `docker0`, `br-*`, `docker_gwbridge`, `podman*`, `cni-podman*`; br_netfilter shows bridged packets with the bridge as their interface) is left to the runtime the same way. `get_rules` also had the family of every rule narrowed by the first one naming an address (`families` set once, outside the loop): an "accept port 80" listed below an IPv4 rule did not exist for IPv6. What changed is that firegex now adds a layer instead of taking one over: an `accept` in `fgex_rules_in` means "firegex does not block this", never "nothing else will", which is all it ever honestly meant. `reset()` is now deleting two tables per family (`add` before `delete`, because a batch is atomic and deleting a table that is not there fails all of it). +A base chain carries its own policy wherever it lives, so default-deny is expressed exactly as before and a `drop` is terminal whichever table's chain reached it. **One thing about what gets blocked did change, and `allow_dnat` puts it back**: in iptables' own `FORWARD` chain, Docker's or podman's accept for a published port came before or after firegex's jump in the same chain, and either way traffic to a published container port never met firegex's policy; a base chain of our own is evaluated on its own, so a drop policy dropped every connection to a published container port — how every service on a CTF box is reached. `dnat_rules()` appends `ct status dnat accept` to `fgex_forward` after the jump (on by default, a switch in the firewall settings): the operator's forward rules still apply first, and the runtime's own chain still decides for itself. **And what the containers send**, which the first version of it left out: with the policy at drop, a web container could not reach its database on the same bridge and no container reached the network — measured with two containers on a Docker network — because Docker accepts that in `FORWARD` too, beside where firegex's rules used to be. So traffic arriving from a container runtime's bridge (`CONTAINER_BRIDGES`: `docker0`, `br-*`, `docker_gwbridge`, `podman*`, `cni-podman*`; br_netfilter shows bridged packets with the bridge as their interface) is left to the runtime the same way. `get_rules` also had the family of every rule narrowed by the first one naming an address (`families` set once, outside the loop): an "accept port 80" listed below an IPv4 rule did not exist for IPv6. What changed is that firegex now adds a layer instead of taking one over: an `accept` in `fgex_rules_in` means "firegex does not block this", never "nothing else will", which is all it ever honestly meant. `reset()` is now deleting two tables per family (`add` before `delete`, because a batch is atomic and deleting a table that is not there fails all of it). **`set()` sends that teardown in the same batch as the firewall replacing it.** It was a batch of its own sent first, so when nft refused a rule — an interface name longer than the kernel takes was enough, and the router accepted anything without a `/` — the old tables were gone and the new ones never arrived: a default-deny firewall became none, and the watcher failed the same way every three seconds. Now a refusal leaves the previous firewall in force, the router puts the database back (`apply_changes(undo)`) and answers 400, and an address field is an IP or an interface name the kernel takes (`INTERFACE`, a trailing `*` allowed). `Rule.table` is still `filter` or `mangle` in the database, the API and the interface — it names the hook a rule is evaluated at. `NFT_TABLES` maps it to the table it lands in; the two were one string only because that string used to be somebody else's table.<> A refused edit must change nothing: `_apply_chain(service_id, undo)` in `routers/services.py` rolls the write back when the transport rejects the resulting chain. Without it an operator is told their filter was rejected and then finds it in the list, and the next start fails for the reason they thought they had avoided. +**And a refused rebuild does not leave the service stopped.** On NFQUEUE a new shape is rebuilt (`ChainShapeChanged` → `restart()`), and the rebuild stops the service before it finds out the new chain cannot start — a filter that loads in the checker's interpreter and not in `cpproxy`'s subinterpreter, a push that times out. Undoing the database afterwards answered "refused", which reads as "nothing happened", while the service was down. `update_chain(undo)` and `address_added(address_id, undo)` now take the undo, run it, and start the service again on what it restored (`_back_as_it_was`); `_once` makes the undo safe to run from both the manager and the router's own `except`. + +**The status the API reports is what is running** (`_running_status`), not the stored one, which is what the operator asked for and what the next boot starts. A service that could not come back at boot kept `active` in the database and the list showed it active with nothing filtering it. Stopping such a service still writes `stop`, or the next boot would try it again. + `edit_service` holds itself to the same rule, and did not. It now checks the service it would become **before writing** — `Transport.check()` against the chain, whether or not the service is running (a stopped one used to take any combination and fail on its next start), a published port refused on a layer that does not dial, and the addresses' `edge`/`proto`/`upstream` recomputed for the new protocol (`_address_under`) — then writes the service and its addresses in **one transaction** (`db.queries`), because a protocol change that collided with another service's address used to leave the service rewritten and its addresses not. A running service whose new definition will not start has the old one written back and is started again on it, rather than being left stopped on the definition that failed. What an address can still be given afterwards — the port of the operator's own proxy on a service moved to the hand-off layer — is not checked there, exactly as at creation: `ExternalTransport.check` names it at start. **Only what differs from the row counts as an edit.** The form sends every setting on every save, and each save used to stop and start a running service — resetting every connection it carried for opening the dialog and pressing save. A change of name alone is written and nothing is restarted, because nothing running reads it. Order is the load-bearing detail in `services/firewall.py`, because getting it wrong is not an error anybody sees — it is a window during which traffic goes unfiltered. Starting: the capture interface if the service decrypts (the engine opens its socket to it at startup, so one that appears afterwards is one this process will not find), then the datapath, then the nft rules that point at it. Stopping is the exact reverse. @@ -235,7 +243,7 @@ Order is the load-bearing detail in `services/firewall.py`, because getting it w ### Block statistics (`modules/services/stats.py`) A cumulative counter per filter, per pattern and per `@pyfilter` function lives in the database; the bucketed history answers the other question — *when* did this start. Blocks are counted per filter into one-minute buckets, kept for `HISTORY_BUCKETS`, and flushed at most every `FLUSH_INTERVAL` seconds. Both properties are load-bearing for the same reason the live log's are: a service refusing thousands of connections a minute must not be able to take the backend down through its own bookkeeping, which is what one row per block would do. There is deliberately **no background task** — flushing happens on the next write once the interval has passed, and always on a read — so nothing has to be started, cancelled or waited for at shutdown. The frontend draws it as hand-written SVG (`StatsPanel.tsx`) rather than pulling in a charting library several times the size of the page it sits on. -**Per-block counters are written once a second, in one transaction** (`ServiceManager.flush_blocks`, `BLOCK_FLUSH`), and the names a block is logged under are looked up once per rule until the chain changes. They were three commits per refused connection, about 17 ms a block, so the backend fell behind any flood of refusals — and that was not only its own problem: the engine wrote `BLOCKED` to a pipe the backend was not reading fast enough, the write blocked the runtime thread that made it, and the flood slowed every other client of the service (measured: 5000 refusals took 74 s and a legitimate client waited a median 4.5 s, failing half the time; now 4 s and 1 ms). Every services API request writes what is waiting first (the router's `_counters_written` dependency), so a read is exact and an edit resetting a pattern's counter cannot have the old pattern's last blocks written onto it afterwards. On the engine's side **nothing on the datapath writes to stdout** (`report.rs`): lines go through a bounded queue and a thread of their own; `BLOCKED` and `STATS` are dropped and counted when the backend is that far behind, and replies the backend waits for (`PORT`, `UDP`, `ACK`) wait for room, in order. +**Per-block counters are written once a second, in one transaction** (`ServiceManager.flush_blocks`, `BLOCK_FLUSH`), and the names a block is logged under are looked up once per rule until the chain changes. They were three commits per refused connection, about 17 ms a block, so the backend fell behind any flood of refusals — and that was not only its own problem: the engine wrote `BLOCKED` to a pipe the backend was not reading fast enough, the write blocked the runtime thread that made it, and the flood slowed every other client of the service (measured: 5000 refusals took 74 s and a legitimate client waited a median 4.5 s, failing half the time; now 4 s and 1 ms). Every services API request writes what is waiting first (the router's `_counters_written` dependency), so a read is exact and an edit resetting a pattern's counter cannot have the old pattern's last blocks written onto it afterwards. On the engine's side **nothing on the datapath writes to stdout** (`report.rs`): lines go through a bounded queue and a thread of their own; `BLOCKED` and `STATS` are dropped and counted when the backend is that far behind, and replies the backend waits for (`PORT`, `UDP`, `ACK`) wait for room, in order. **Nor does anything on the datapath write to stderr directly**: `crate::diag!` goes through a queue and a thread of its own and is dropped and counted when full, and `crate::diag_throttled!` says at most a few lines per window from one call site and counts the rest into one line (`report::Throttle`). Every line a client can make the engine write at will — a connection ending with an error, a cleartext client at a TLS address, a malformed request — was an `eprintln!` from the thread carrying traffic, which takes one lock for the whole process and blocks once the pipe fills, and at `[warn]` it flooded the 500-line log and the problem badge. `main.rs` and `control.rs` keep `eprintln!`: a `[fatal]` line has to be written before the `exit` after it. `share` is computed in the backend, not the browser, so a chart and the table beside it cannot disagree about what a rule accounts for. @@ -353,7 +361,7 @@ Note for tests: a client now dials the service's own address, because there is n It is UDP to the kernel, and that is the whole of what the backend had to learn: `L4.l4_of(quic) == udp`, so the addresses, the uniqueness key, the nft rules, `udp_relay_key`/`udp_relay_host`, `ProxyTransport.udp_ports` and `ADD_UDP` are the code that was already there. What changes is what binds the port — `FGEX_PROXY_QUIC=1` beside the certificate makes the engine build `quic::QuicManager` instead of `udp::UdpManager`, and `relays::Relays` is the two-armed enum the control channel and the startup list both go through. `Service.decrypts` (once `terminates_tls`) is what gates the certificate, on both protocols. -**The ALPN is mirrored, as on the TLS path, and nothing is configured.** The ClientHello travels inside an Initial packet, and that was taken for a reason it could not be read before answering — so the engine offered the service a candidate list instead, `FGEX_PROXY_QUIC_ALPN` for the whole instance and then a per-service field, and a QUIC service speaking anything but HTTP/3 was unreachable until somebody found the setting. But Initial keys are derived from the Destination Connection ID the client sends in the clear (RFC 9001 §5.2), exactly so a middlebox can read this. `quic_hello.rs` does: `SniffingSocket` wraps the endpoint's socket, every long-header Initial is decrypted with `rustls::quic::Suite::keys`, CRYPTO frames are reassembled by offset across datagrams (a ClientHello with post-quantum key shares does not fit one packet), and the ALPN list is parked per client address for `QuicRelay::offered`, which waits `HELLO_WAIT` for the rest of a split hello. `dial_upstream` then offers the service exactly that list and `QuicSetup::answering_with` tells the client the one protocol the service chose — both built per connection, as the TLS path does. The map is bounded (`MAX_PENDING`, `KEPT_FOR`) because forged Initials cost the sender nothing. A hello that cannot be read falls back to `h3`, said once per relay. The parser is tested against packets protected with rustls's own client-side keys, and `tests/quic.rs` drives real quinn clients through it, including one whose hello spans packets. +**The ALPN is mirrored, as on the TLS path, and nothing is configured.** The ClientHello travels inside an Initial packet, and that was taken for a reason it could not be read before answering — so the engine offered the service a candidate list instead, `FGEX_PROXY_QUIC_ALPN` for the whole instance and then a per-service field, and a QUIC service speaking anything but HTTP/3 was unreachable until somebody found the setting. But Initial keys are derived from the Destination Connection ID the client sends in the clear (RFC 9001 §5.2), exactly so a middlebox can read this. `quic_hello.rs` does: `SniffingSocket` wraps the endpoint's socket, every long-header Initial is decrypted with `rustls::quic::Suite::keys`, CRYPTO frames are reassembled by offset across datagrams (a ClientHello with post-quantum key shares does not fit one packet), and the ALPN list is parked per client address for `QuicRelay::offered`, which waits `HELLO_WAIT` for the rest of a split hello. `dial_upstream` then offers the service exactly that list and `QuicSetup::answering_with` tells the client the one protocol the service chose — both built per connection, as the TLS path does. The map is bounded (`MAX_PENDING`, `KEPT_FOR`, `MAX_FRAGMENTS`, `MAX_HELLO`) because forged Initials cost the sender nothing — and it was not bounded enough: empty CRYPTO frames counted towards nothing, so a peer grew one hello without end while every packet sorted all its pieces again inside the endpoint's receive path. Empty pieces and pieces past `MAX_HELLO` are not kept, a hello holds at most `MAX_FRAGMENTS`, the table is swept by age every `SWEEP_EVERY` rather than only when full, and a full table evicts its oldest entry instead of refusing newcomers — refusing meant a thousand forged Initials every ten seconds stopped every real client's hello from being read. A hello that cannot be read falls back to `h3`, said once per relay. The parser is tested against packets protected with rustls's own client-side keys, and `tests/quic.rs` drives real quinn clients through it, including one whose hello spans packets. **One stream is one connection to the chain** (`Carrier::open`), with its own `ChainSessions` and its own module globals, released when the stream ends. A filter's state follows a stream of bytes from start to end, which is what a QUIC stream is and what a QUIC connection is not; on HTTP/3 it also lands where an operator expects, one request to a stream. The raw path reuses `proxy.rs`'s pumps unchanged — `spawn_pumps`, `pump` and `join_pumps` are `pub(crate)` for exactly this, and `spawn_pumps` takes the `spoken` flag rather than making one, because the first-byte deadline is the *connection's* question and a flag per stream would never answer it for a connection that opened a hundred and spoke on none. diff --git a/backend/modules/firewall/nftables.py b/backend/modules/firewall/nftables.py index 5cc8d63..69654a5 100644 --- a/backend/modules/firewall/nftables.py +++ b/backend/modules/firewall/nftables.py @@ -254,7 +254,7 @@ class FiregexTables(NFTableManager): # one because deleting the table around them was not an option. `add` before # `delete` is there because a batch is atomic and deleting a table that is not # there fails the whole thing; adding one that already exists does nothing. - super().__init__(self.init_comands(), [ + self._teardown = [ command for family in ("ip", "ip6") for table in (self.filter_table, self.mangle_table) @@ -262,7 +262,8 @@ class FiregexTables(NFTableManager): {"add": {"table": {"name": table, "family": family}}}, {"delete": {"table": {"name": table, "family": family}}}, ) - ]) + ] + super().__init__(self.init_comands(), self._teardown) def intact(self) -> bool: """Whether the firewall firegex installed is still in the kernel. @@ -276,8 +277,16 @@ class FiregexTables(NFTableManager): return code == 0 and any("rule" in item for item in listed["nftables"]) def set(self, srvs:list[Rule], policy:str=Action.ACCEPT, opt:FirewallSettings = None): + """Replace the whole firewall with this one, or leave the one in force alone. + + **One batch, teardown included.** nft applies a batch whole or not at all, and the + teardown used to be a batch of its own sent first: when nft then refused a rule — + an interface name longer than the kernel allows was enough — the old tables were + already gone and the new ones never arrived. A default-deny firewall became no + firewall at all, and the watcher putting it back failed the same way every few + seconds. Sent together, a refusal leaves the previous firewall exactly as it was. + """ srvs = list(srvs) - self.reset() if policy == Action.REJECT: policy = Action.DROP srvs.append(Rule( @@ -296,7 +305,7 @@ class FiregexTables(NFTableManager): # No hooking step any more: the base chains are firegex's own and `_skeleton` # already pointed each of them at its rules chain. rules = self.init_comands(policy, opt) + self.get_rules(*srvs) + self.dnat_rules(opt) - self.cmd(*rules) + self.cmd(*self._teardown, *rules) def dnat_rules(self, opt: FirewallSettings | None) -> list[dict]: """Leave port-forwarded traffic to the rules that forwarded it, when no rule of diff --git a/backend/modules/services/conntrack.py b/backend/modules/services/conntrack.py new file mode 100644 index 0000000..fbcf5ca --- /dev/null +++ b/backend/modules/services/conntrack.py @@ -0,0 +1,358 @@ +"""Forgetting UDP flows, so the next datagram of each is steered by the rules in force now. + +Conntrack translates a flow once, at its first packet, and the proxy layer's `redirect` +is a translation. TCP lives with that: a connection is one thing from its handshake to +its close, and one opened before the service started is simply carried to the end. UDP +has no close — a flow lasts for as long as datagrams keep arriving inside the timeout — +and that turned the same rule into three failures: + +* **a flow older than the service was never filtered**, for as long as it kept talking: + nothing redirected it, and nothing ever would. A client could open one before the + protection went up and keep it open indefinitely; +* **a flow carried across a restart stayed with the engine being retired**, under the + filters it started with, and kept that engine alive until the drain gave up on it; +* **and once that engine was gone, the flow went nowhere.** Conntrack went on sending it + to a relay port nobody listened on, every datagram refreshing the entry that did so, + until the client happened to fall silent for two minutes. + +A UDP flow can simply be forgotten: its next datagram is a new flow, judged by whatever +rules are there now, and neither end had any state that depended on the old entry. So +this deletes the entries, over ctnetlink — the kernel's own interface, the one the +`conntrack` tool speaks, with nothing to install and nothing to spawn. + +TCP is never touched here: forgetting a connection breaks it. +""" + +import ipaddress +import socket +import struct +from dataclasses import dataclass +from typing import Callable, Iterable + +NETLINK_NETFILTER = 12 + +NLM_F_REQUEST = 0x1 +NLM_F_ACK = 0x4 +NLM_F_DUMP = 0x300 +NLMSG_ERROR = 2 +NLMSG_DONE = 3 + +NFNL_SUBSYS_CTNETLINK = 1 +IPCTNL_MSG_CT_GET = 1 +IPCTNL_MSG_CT_DELETE = 2 + +NLA_TYPE_MASK = 0x3FFF + +CTA_TUPLE_ORIG = 1 +CTA_TUPLE_REPLY = 2 +CTA_STATUS = 3 +CTA_MARK = 8 +CTA_ZONE = 18 + +CTA_TUPLE_IP = 1 +CTA_TUPLE_PROTO = 2 + +CTA_IP_V4_SRC = 1 +CTA_IP_V4_DST = 2 +CTA_IP_V6_SRC = 3 +CTA_IP_V6_DST = 4 + +CTA_PROTO_NUM = 1 +CTA_PROTO_SRC_PORT = 2 +CTA_PROTO_DST_PORT = 3 + +#: `IPS_DST_NAT`: the flow's destination was rewritten — by a `redirect`, among others. +IPS_DST_NAT = 1 << 5 + +IPPROTO_UDP = 17 + +_HEADER = struct.Struct("=IHHII") +_NFGEN = struct.Struct("=BBH") +_ATTR = struct.Struct("=HH") + + +@dataclass +class Flow: + """One conntrack entry, as much of it as deciding and deleting need.""" + + family: int + l4proto: int + orig_src: ipaddress._BaseAddress | None + orig_dst: ipaddress._BaseAddress | None + orig_sport: int | None + orig_dport: int | None + reply_src: ipaddress._BaseAddress | None + reply_sport: int | None + status: int + mark: int + #: The original tuple and the zone exactly as the kernel sent them, which is what a + #: delete names the entry by. + raw_orig: bytes + raw_zone: bytes = b"" + + @property + def redirected(self) -> bool: + return bool(self.status & IPS_DST_NAT) + + +# --- the wire ------------------------------------------------------------------- + + +def _align(n: int) -> int: + return (n + 3) & ~3 + + +def attr(kind: int, payload: bytes) -> bytes: + """One netlink attribute, padded.""" + raw = _ATTR.pack(_ATTR.size + len(payload), kind) + payload + return raw + b"\0" * (_align(len(raw)) - len(raw)) + + +def attrs(data: bytes) -> Iterable[tuple[int, bytes, bytes]]: + """`(type, payload, the whole attribute)` for each attribute in `data`.""" + pos = 0 + while pos + _ATTR.size <= len(data): + length, kind = _ATTR.unpack_from(data, pos) + if length < _ATTR.size or pos + length > len(data): + return + yield kind & NLA_TYPE_MASK, data[pos + _ATTR.size:pos + length], data[pos:pos + length] + pos += _align(length) + + +def message(kind: int, flags: int, seq: int, family: int, payload: bytes = b"") -> bytes: + body = _NFGEN.pack(family, 0, 0) + payload + return _HEADER.pack(_HEADER.size + len(body), (NFNL_SUBSYS_CTNETLINK << 8) | kind, + flags, seq, 0) + body + + +def messages(data: bytes) -> Iterable[tuple[int, bytes]]: + """`(type, payload)` for each netlink message in one read.""" + pos = 0 + while pos + _HEADER.size <= len(data): + length, kind, _flags, _seq, _pid = _HEADER.unpack_from(data, pos) + if length < _HEADER.size or pos + length > len(data): + return + yield kind, data[pos + _HEADER.size:pos + length] + pos += _align(length) + + +def _tuple(data: bytes) -> tuple: + src = dst = sport = dport = None + proto = 0 + for kind, payload, _ in attrs(data): + if kind == CTA_TUPLE_IP: + for ip_kind, value, _ in attrs(payload): + if ip_kind in (CTA_IP_V4_SRC, CTA_IP_V6_SRC): + src = ipaddress.ip_address(value) + elif ip_kind in (CTA_IP_V4_DST, CTA_IP_V6_DST): + dst = ipaddress.ip_address(value) + elif kind == CTA_TUPLE_PROTO: + for proto_kind, value, _ in attrs(payload): + if proto_kind == CTA_PROTO_NUM: + proto = value[0] + elif proto_kind == CTA_PROTO_SRC_PORT: + sport = struct.unpack("!H", value[:2])[0] + elif proto_kind == CTA_PROTO_DST_PORT: + dport = struct.unpack("!H", value[:2])[0] + return src, dst, sport, dport, proto + + +def parse_flow(payload: bytes) -> Flow | None: + """A conntrack entry out of one `IPCTNL_MSG_CT_NEW` message's payload.""" + if len(payload) < _NFGEN.size: + return None + family = payload[0] + orig = reply = None + raw_orig = raw_zone = b"" + status = mark = 0 + for kind, value, whole in attrs(payload[_NFGEN.size:]): + if kind == CTA_TUPLE_ORIG: + orig, raw_orig = _tuple(value), whole + elif kind == CTA_TUPLE_REPLY: + reply = _tuple(value) + elif kind == CTA_STATUS: + status = struct.unpack("!I", value[:4])[0] + elif kind == CTA_MARK: + mark = struct.unpack("!I", value[:4])[0] + elif kind == CTA_ZONE: + raw_zone = whole + if orig is None: + return None + reply_src, _, reply_sport, _, _ = reply if reply else (None, None, None, None, 0) + return Flow( + family=family, l4proto=orig[4], + orig_src=orig[0], orig_dst=orig[1], orig_sport=orig[2], orig_dport=orig[3], + reply_src=reply_src, reply_sport=reply_sport, + status=status, mark=mark, raw_orig=raw_orig, raw_zone=raw_zone, + ) + + +def delete_request(flow: Flow, seq: int) -> bytes: + return message(IPCTNL_MSG_CT_DELETE, NLM_F_REQUEST | NLM_F_ACK, seq, flow.family, + flow.raw_orig + flow.raw_zone) + + +# --- talking to the kernel ------------------------------------------------------ + + +def _read_until_done(sock: socket.socket) -> Iterable[bytes]: + """The payload of every entry a dump answers with.""" + while True: + data = sock.recv(1 << 18) + if not data: + return + for kind, payload in messages(data): + if kind == NLMSG_DONE: + return + if kind == NLMSG_ERROR: + error = struct.unpack_from("=i", payload)[0] if len(payload) >= 4 else 0 + if error: + raise OSError(-error, "conntrack dump refused") + return + yield payload + + +def _acknowledged(sock: socket.socket) -> int: + """The error a request was answered with, 0 for none.""" + while True: + for kind, payload in messages(sock.recv(1 << 16)): + if kind == NLMSG_ERROR: + return -struct.unpack_from("=i", payload)[0] if len(payload) >= 4 else 0 + + +def delete_where(chosen: Callable[[Flow], bool]) -> int: + """Delete every UDP entry `chosen` picks. Returns how many went. + + The table is read once and the deletes follow: a flow that ended in between is + answered `ENOENT`, which is the state being asked for. + """ + with socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, NETLINK_NETFILTER) as sock: + sock.settimeout(5) + sock.bind((0, 0)) + sock.send(message(IPCTNL_MSG_CT_GET, NLM_F_REQUEST | NLM_F_DUMP, 1, socket.AF_UNSPEC)) + doomed = [] + for payload in _read_until_done(sock): + flow = parse_flow(payload) + if flow is not None and flow.l4proto == IPPROTO_UDP and chosen(flow): + doomed.append(flow) + gone = 0 + for seq, flow in enumerate(doomed, start=2): + sock.send(delete_request(flow, seq)) + error = _acknowledged(sock) + if error == 0: + gone += 1 + elif error != 2: # ENOENT: it ended by itself meanwhile + raise OSError(error, "conntrack refused a delete") + return gone + + +# --- what to forget ------------------------------------------------------------- + + +Target = tuple[ipaddress.IPv4Network | ipaddress.IPv6Network, int] + + +def targets(pairs: Iterable[tuple[str, int]]) -> list[Target]: + """`(address or network, port)` pairs as something a flow can be compared with.""" + return [(ipaddress.ip_network(str(ip), strict=False), int(port)) for ip, port in pairs] + + +def headed_for(flow: Flow, wanted: list[Target]) -> bool: + """Whether a flow was going to one of `wanted` before anything rewrote it.""" + if flow.orig_dst is None or flow.orig_dport is None: + return False + return any( + flow.orig_dport == port and flow.orig_dst.version == network.version + and flow.orig_dst in network + for network, port in wanted + ) + + +def chooser(*, moved: list[Target], unsteered: list[Target], spare_mark: int, + bound: set[int]) -> Callable[[Flow], bool]: + """Which flows `forget_udp` lets go of. Split out so the decision can be tested + without a kernel.""" + + def chosen(flow: Flow) -> bool: + if flow.mark == spare_mark: + return False + if headed_for(flow, moved): + return True + if headed_for(flow, unsteered): + return not flow.redirected or flow.reply_sport not in bound + return False + + return chosen + + +def forget_udp(*, moved: Iterable[tuple[str, int]] = (), + unsteered: Iterable[tuple[str, int]] = (), spare_mark: int) -> int: + """Forget UDP flows headed for these addresses, so their next datagram is steered by + the rules in force now. + + * `moved`: every flow headed there, including the ones a redirect already took to an + engine — a datagram flow can change engines between two datagrams and lose nothing + but the filter's own state for it, which is what a new configuration costs anyway. + * `unsteered`: only the ones nothing is carrying — never redirected at all, or + redirected to a port nobody listens on any more. What a QUIC edge needs: its + connections are terminated by an engine, and moved to another one mid-connection + they would end, while one still draining is carrying them. + + `spare_mark` is the engine's own: its dials to the service are flows headed for the + same address, and they are its to keep. + """ + moved_to, unsteered_to = targets(moved), targets(unsteered) + if not moved_to and not unsteered_to: + return 0 + return delete_where(chooser( + moved=moved_to, unsteered=unsteered_to, spare_mark=spare_mark, + bound=bound_udp_ports() if unsteered_to else set(), + )) + + +def bound_udp_ports() -> set[int]: + """Every UDP port some socket on this host is bound to, from `/proc/net/udp{,6}`.""" + found = set() + for path in ("/proc/net/udp", "/proc/net/udp6"): + try: + with open(path) as table: + next(table, None) + for line in table: + fields = line.split() + if len(fields) > 1 and ":" in fields[1]: + found.add(int(fields[1].rsplit(":", 1)[1], 16)) + except OSError: + continue + return found + + +def forget_udp_redirected_to(ports: Iterable[int], local: set) -> int: + """Forget the UDP flows a redirect sent to one of these ports on this host. + + What is left of an engine once it has gone: every such flow points at a relay nobody + listens on any more. Recognised by the reply coming from that port on one of this + host's own addresses — which is what a redirect writes into the reply tuple, and what + no flow forwarded to a container or another host can have. + """ + wanted = {int(port) for port in ports} + if not wanted: + return 0 + return delete_where( + lambda flow: flow.redirected and flow.reply_sport in wanted + and flow.reply_src is not None and flow.reply_src in local + ) + + +def local_addresses() -> set: + """Every address this host holds right now.""" + import psutil + + found = set() + for entries in psutil.net_if_addrs().values(): + for entry in entries: + if entry.family in (socket.AF_INET, socket.AF_INET6): + try: + found.add(ipaddress.ip_address(entry.address.split("%")[0])) + except ValueError: + pass + return found diff --git a/backend/modules/services/firewall.py b/backend/modules/services/firewall.py index 57bac58..3346092 100644 --- a/backend/modules/services/firewall.py +++ b/backend/modules/services/firewall.py @@ -20,14 +20,16 @@ import os import time import traceback +from modules.services import conntrack from modules.services import mirror from modules.services import transports from modules.services import stats from modules.services.logs import LEVEL, forget as forget_log, log_for from modules.services.models import (KIND, L4, STATUS, UPSTREAM, Address, Filter, Regex, Service) -from modules.services.nftables import (FiregexTables, NoRelayAddress, +from modules.services.nftables import (PROXY_SELF_MARK, FiregexTables, NoRelayAddress, interface_addresses, udp_relay_host) +from utils import is_ip_parse from utils.sqlite import SQLite nft = FiregexTables() @@ -141,6 +143,8 @@ class ServiceManager: #: token rather than once per block, and forgotten whenever the chain changes. self._named: dict[tuple[str, str], str] = {} self._owners: dict[str, str] = {} + #: Work started after an engine has gone, kept referenced until it is done. + self._background: set[asyncio.Task] = set() # --- reading the configuration ------------------------------------------- @@ -487,6 +491,7 @@ class ServiceManager: self.transport = None raise self._set_status(True) + await self._claim_udp(self.srv.addresses) self.log.add( LEVEL.INFO, f"started on the {self.srv.transport} layer, " @@ -510,20 +515,29 @@ class ServiceManager: async with self.lock: self.flush_blocks() if not self.active: + # One that was meant to run and could not — at boot, or after a failed + # rebuild — is not running, and stopping it is still a decision: kept as + # it was, the next boot would try to start it again. + if persist and self.srv.status == STATUS.ACTIVE: + self._set_status(False) return - guarded = nft.delete(self.srv, keep_guards=drain) + guarded = nft.delete(self.srv, keep_guards=True) self._steer = {} + # A restart claims them again in a moment, from the engine replacing this one. + if not handing_over: + await self._release_udp(self.srv.addresses) if self.transport: if drain: await self.transport.retire( keep_filtering=handing_over, - then=lambda: nft.release_guards(guarded), + then=lambda: self._in_background(self._engine_gone(guarded)), ) else: await self.transport.stop() + await self._engine_gone(guarded) self.transport = None else: - nft.release_guards(guarded) + await self._engine_gone(guarded) self._set_status(False, persist=persist) self.log.add(LEVEL.INFO, "stopped") @@ -544,6 +558,9 @@ class ServiceManager: return nft.delete(self.srv) nft.add(self.srv, **self._steer) + # Whatever arrived while the rules were gone went straight to the service, and + # a datagram flow that did so would go on doing it. + await self._claim_udp(self.srv.addresses) self._rules_lost += 1 now = time.time() if now - self._rules_lost_told < TABLE_TOLD_QUIET: @@ -559,7 +576,29 @@ class ServiceManager: "manages this host's nftables leave the `fgex` tables alone.", ) - async def update_chain(self): + async def _back_as_it_was(self, undo) -> None: + """A rebuild the new configuration could not survive: put the old one back, and + the service with it. + + The rebuild stopped the service before it found out, so undoing the edit in the + database alone left it stopped — the operator was told their change had been + refused, which reads as "nothing happened", while the service they had been + editing was no longer protecting anything. + """ + if undo: + undo() + self.reload_addresses() + if self.active: + return + try: + await self.enable() + except Exception as e: + self.log.add(LEVEL.ERROR, f"could not start it again as it was either: {e}") + return + self.log.add(LEVEL.WARN, "the change was refused, so the service was started again " + "as it was before it") + + async def update_chain(self, undo=None): """Push the current chain to a running datapath. Editing a rule costs nobody their connection: the datapath swaps its @@ -571,6 +610,11 @@ class ServiceManager: own chain priority, so the shape of the chain *is* the arrangement of processes and rules. That has to be rebuilt, which is a visible interruption — better than silently enforcing the old order. + + `undo` puts the database back as it was before the edit. A rebuild that fails has + already stopped the service, so it is not enough for the caller to undo the edit + afterwards: the service is started again on what `undo` restored — see + `_back_as_it_was`. """ restart = False # What was refused before this edit is written under the names it had, and what @@ -587,7 +631,11 @@ class ServiceManager: restart = True if restart: self.log.add(LEVEL.INFO, "the chain changed shape; rebuilding it") - await self.restart() + try: + await self.restart() + except Exception: + await self._back_as_it_was(undo) + raise return self.log.add( LEVEL.INFO, @@ -596,7 +644,7 @@ class ServiceManager: f"(no connection was dropped)", ) - async def address_added(self, address_id: str): + async def address_added(self, address_id: str, undo=None): """Steer one more address at the datapath that is already running. Deliberately not a restart. The chain is unchanged and the datapath is already @@ -628,7 +676,11 @@ class ServiceManager: LEVEL.INFO, "the first IPv6 address needs a listener that can accept one; rebuilding", ) - await self.restart() + try: + await self.restart() + except Exception: + await self._back_as_it_was(undo) + raise return added = [addr for addr in self.srv.addresses if addr.id == address_id] if not added: @@ -684,6 +736,7 @@ class ServiceManager: host, addr.target_port or addr.port, UPSTREAM.env(addr.upstream) ) nft.add(self.srv, added, **self._steer) + await self._claim_udp(added) self.log.add( LEVEL.INFO, f"also protecting {added[0].ip_int}:{added[0].port} (no connection was dropped)", @@ -696,6 +749,7 @@ class ServiceManager: async with self.lock: if self.active: nft.delete(self.srv, gone) + await self._release_udp(gone) # And the engine's own note of where this address fronted, so its # map cannot come to disagree with the rules that feed it. for addr in gone: @@ -719,6 +773,95 @@ class ServiceManager: # address that no longer exists. self.srv.addresses = [addr for addr in self.srv.addresses if addr.id != address_id] + # --- UDP flows conntrack is still carrying ------------------------------------- + + def _udp_edges(self, addresses: list[Address]) -> tuple[list, list]: + """The datagram addresses and the QUIC ones among these, as `(address, port)` + pairs a flow can be matched against — an interface by what it carries.""" + plain, quic = [], [] + for addr in addresses: + if L4.l4_of(addr.proto or self.srv.proto) != L4.UDP: + continue + hosts = ([addr.ip_int] if is_ip_parse(addr.ip_int) + else interface_addresses(addr.ip_int)) + (quic if str(addr.edge) == L4.QUIC else plain).extend( + (host, addr.port) for host in hosts) + return plain, quic + + async def _forget_flows(self, work, said: str) -> None: + """Run a conntrack sweep off the event loop, and say what it did. + + Never fatal: at worst a flow goes on where it was going, which is what happened + before anything here existed. + """ + try: + gone = await asyncio.to_thread(work) + except Exception as e: + self.log.add(LEVEL.WARN, f"could not re-steer the UDP flows already open: {e}") + return + if gone: + self.log.add(LEVEL.INFO, said.format(gone=gone)) + + async def _claim_udp(self, addresses: list[Address]) -> None: + """Make the UDP flows already open to these addresses meet the rules just installed. + + Conntrack translates a flow at its first packet, so a datagram flow older than the + redirect went on straight to the service — unfiltered, for as long as it kept + talking — and one a previous engine carried stayed with it. Forgotten, each one's + next datagram is a new flow and is redirected like any other. A QUIC edge keeps + the connections an engine is still carrying, since moving one would end it. + """ + if self.srv.transport != transports.TRANSPORT.PROXY: + return + plain, quic = self._udp_edges(addresses) + if not plain and not quic: + return + await self._forget_flows( + lambda: conntrack.forget_udp(moved=plain, unsteered=quic, + spare_mark=PROXY_SELF_MARK), + "{gone} UDP flow(s) already open reach the filters from their next datagram", + ) + + async def _release_udp(self, addresses: list[Address]) -> None: + """Send the datagram flows the engine carried straight to the service. + + What stopping means for them. Left alone, each went on to the engine being retired + for as long as it kept talking — kept the engine alive until the drain gave up, and + then went nowhere at all. QUIC connections are the engine's, and are carried until + they close. + """ + if self.srv.transport != transports.TRANSPORT.PROXY: + return + plain, _ = self._udp_edges(addresses) + if not plain: + return + await self._forget_flows( + lambda: conntrack.forget_udp(moved=plain, spare_mark=PROXY_SELF_MARK), + "{gone} UDP flow(s) go straight to the service from their next datagram", + ) + + async def _engine_gone(self, guarded: set[tuple[str, int]]) -> None: + """What is left to do once an engine has exited: its ports stop being guarded, and + the UDP flows still pointed at them are let go. + + Those flows were headed for a relay nobody listens on now, and every datagram + refreshed the entry that sent it there — a client that kept talking was cut off + from the service until it happened to fall silent for two minutes. + """ + nft.release_guards(guarded) + ports = [port for l4, port in guarded if l4 == L4.UDP] + if not ports: + return + await self._forget_flows( + lambda: conntrack.forget_udp_redirected_to(ports, conntrack.local_addresses()), + "{gone} UDP flow(s) left pointing at a stopped engine were let go", + ) + + def _in_background(self, work) -> None: + task = asyncio.get_running_loop().create_task(work) + self._background.add(task) + task.add_done_callback(self._background.discard) + def traffic(self) -> dict: """How much has arrived, and how much of it was refused. diff --git a/backend/proxysrc/src/filter.rs b/backend/proxysrc/src/filter.rs index 1d2290d..21a6cc1 100644 --- a/backend/proxysrc/src/filter.rs +++ b/backend/proxysrc/src/filter.rs @@ -391,7 +391,7 @@ impl FilterChain { slot.rearm_at_ms.store(at, Ordering::Relaxed); let _ = self.next_rearm_ms.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |next| (next == 0 || at < next).then_some(at)); - eprintln!( + crate::diag!( "[warn] [filter] '{}' kept missing its {:?} deadline: traffic keeps \ flowing without it, and it is asked again in {}s", slot.filter.name(), @@ -399,7 +399,7 @@ impl FilterChain { self.rearm_after.as_secs(), ); } - DisableReason::Panicked => eprintln!( + DisableReason::Panicked => crate::diag!( "[warn] [filter] '{}' panicked: traffic keeps flowing without it until the \ filters are next applied — save a change to them, or restart the service", slot.filter.name(), @@ -411,7 +411,7 @@ impl FilterChain { .all(|s| s.disabled.load(Ordering::Relaxed)) { self.degraded.store(true, Ordering::Relaxed); - eprintln!("[warn] [filter] every filter is disabled, chain is now a plain relay"); + crate::diag!("[warn] [filter] every filter is disabled, chain is now a plain relay"); } } @@ -456,7 +456,7 @@ impl FilterChain { slot.consecutive_timeouts.store(0, Ordering::Relaxed); slot.disabled.store(false, Ordering::Relaxed); self.degraded.store(false, Ordering::Relaxed); - eprintln!( + crate::diag!( "[info] [filter] '{}' is back in the chain after sitting out {}s", slot.filter.name(), self.rearm_after.as_secs(), diff --git a/backend/proxysrc/src/h1up.rs b/backend/proxysrc/src/h1up.rs index 744c103..47d63ff 100644 --- a/backend/proxysrc/src/h1up.rs +++ b/backend/proxysrc/src/h1up.rs @@ -108,7 +108,7 @@ impl H1Upstream { Ok(Err(e)) => e.to_string(), _ => "timed out (is the return path diverted?)".to_string(), }; - eprintln!( + crate::diag_throttled!("[warn] [h1] dials that could not keep the client's address", 1, 60, "[warn] [h1] cannot reach {} as {}: {why}. Falling back to our own \ address — the service will not see real client IPs.", self.upstream, @@ -210,7 +210,7 @@ impl H1Upstream { // detached — one connection per exchange is what makes that safe to say. let driving = tokio::spawn(async move { if let Err(e) = connection.await { - eprintln!("[info] [h1] the connection to the service ended: {e}"); + crate::diag_throttled!("[info] [h1] connections to the service that ended with an error", 20, 10, "[info] [h1] the connection to the service ended: {e}"); } }); // Not awaited here: the service is allowed to answer before it has read the whole diff --git a/backend/proxysrc/src/h2.rs b/backend/proxysrc/src/h2.rs index b63c974..f0edc60 100644 --- a/backend/proxysrc/src/h2.rs +++ b/backend/proxysrc/src/h2.rs @@ -408,12 +408,12 @@ where // refused request and a completed one both end it. ctx.chain.current().connection_closed(connection); if let Err(e) = outcome { - eprintln!("[info] [h2] stream ended: {e}"); + crate::diag_throttled!("[info] [h2] streams that ended with an error", 20, 10, "[info] [h2] stream ended: {e}"); } }); } Some(Err(e)) => { - eprintln!("[info] [h2] the client's connection ended: {e}"); + crate::diag_throttled!("[info] [h2] client connections that ended with an error", 20, 10, "[info] [h2] the client's connection ended: {e}"); break; } // The client is done with the connection, which is the ordinary way out. @@ -425,7 +425,7 @@ where // shared driver to outlive and nothing to watch here. reason = async { driving.as_mut().unwrap().await }, if driving.is_some() => { if let Ok(Err(e)) = reason { - eprintln!("[info] [h2] the service ended the connection: {e}"); + crate::diag_throttled!("[info] [h2] connections the service ended", 20, 10, "[info] [h2] the service ended the connection: {e}"); } break; } @@ -526,7 +526,7 @@ async fn exchange( // A tunnel is opaque bytes with no HTTP/1.1 message to render, so it is refused rather // than shown to the chain as something it is not. Naming it beats a stream that hangs. if head.method() == http::Method::CONNECT { - eprintln!( + crate::diag_throttled!("[warn] [h2] CONNECT streams refused", 5, 30, "[warn] [h2] {}: a CONNECT stream arrived. firegex terminates HTTP/2 to show \ the filters each exchange as HTTP/1.1, and a tunnel has no such exchange in \ it, so the stream is refused rather than carried unread.", @@ -541,7 +541,7 @@ async fn exchange( // request as it came, so two that disagree would be a filter reading one host while // the service answers for another. Malformed by RFC 9113 §8.3.1, and treated so. if let Some(why) = conflicting_authority(&head) { - eprintln!( + crate::diag_throttled!("[warn] [h2] requests with two authorities", 5, 30, "[warn] [h2] {}: {why}, so the filters would be shown one host while the \ service routes on the other. The stream is reset as malformed.", ctx.client diff --git a/backend/proxysrc/src/h3.rs b/backend/proxysrc/src/h3.rs index 51fdfc7..eddabc4 100644 --- a/backend/proxysrc/src/h3.rs +++ b/backend/proxysrc/src/h3.rs @@ -164,7 +164,7 @@ pub(crate) async fn carry(carrier: &Carrier) { { Ok(connection) => connection, Err(e) => { - eprintln!("[info] [h3] the client's HTTP/3 connection did not start: {e}"); + crate::diag_throttled!("[info] [h3] HTTP/3 connections that did not start", 20, 10, "[info] [h3] the client's HTTP/3 connection did not start: {e}"); return; } }; @@ -178,7 +178,7 @@ pub(crate) async fn carry(carrier: &Carrier) { { Ok(pair) => pair, Err(e) => { - eprintln!("[warn] [h3] {} does not speak HTTP/3: {e}", carrier.upstream); + crate::diag_throttled!("[warn] [h3] connections to a service that did not speak HTTP/3", 1, 30, "[warn] [h3] {} does not speak HTTP/3: {e}", carrier.upstream); carrier.peer.close(0u32.into(), b"the service did not speak HTTP/3"); return; } @@ -226,7 +226,7 @@ async fn accept_loop( } } => { if let Some(reason) = reason { - eprintln!("[info] [h3] the service ended the connection: {reason}"); + crate::diag_throttled!("[info] [h3] connections the service ended", 20, 10, "[info] [h3] the service ended the connection: {reason}"); } carrier.peer.close(0u32.into(), b"the service closed the connection"); break; @@ -243,14 +243,14 @@ async fn accept_loop( // refused request and a completed one both end it. carrier.chain.current().connection_closed(connection); if let Err(e) = outcome { - eprintln!("[info] [h3] request ended: {e}"); + crate::diag_throttled!("[info] [h3] requests that ended with an error", 20, 10, "[info] [h3] request ended: {e}"); } }); } // The client is done with the connection, which is the ordinary way out. Ok(None) => break, Err(e) => { - eprintln!("[info] [h3] the client's connection ended: {e}"); + crate::diag_throttled!("[info] [h3] client connections that ended with an error", 20, 10, "[info] [h3] the client's connection ended: {e}"); break; } } @@ -286,7 +286,7 @@ async fn exchange( // request as it came, so two that disagree would be a filter reading one host while // the service answers for another. Malformed, and the stream is stopped as such. if let Some(why) = conflicting_authority(&request) { - eprintln!( + crate::diag_throttled!("[warn] [h3] requests with two authorities", 5, 30, "[warn] [h3] {}: {why}, so the filters would be shown one host while the \ service routes on the other. The stream is stopped as malformed.", carrier.client @@ -297,7 +297,7 @@ async fn exchange( // Malformed as well (RFC 9114 §4.2), and not refused by the h3 crate: see // `connection_specific`. HTTP/2 needs no such line, because h2 refuses them itself. if let Some(why) = connection_specific(request.headers()) { - eprintln!( + crate::diag_throttled!("[warn] [h3] requests carrying connection-specific headers", 5, 30, "[warn] [h3] {}: {why}. The stream is stopped as malformed.", carrier.client ); diff --git a/backend/proxysrc/src/http1.rs b/backend/proxysrc/src/http1.rs index 22c4695..ef98891 100644 --- a/backend/proxysrc/src/http1.rs +++ b/backend/proxysrc/src/http1.rs @@ -274,7 +274,7 @@ pub(crate) fn carried( trailers: Option, ) -> Option { if framing == Framing::Sized && trailers.is_some() { - eprintln!( + crate::diag_throttled!("[warn] [http] trailer sections dropped from a message of declared length", 5, 30, "[warn] [{}] {}: a trailer section arrived on a message that declared its \ length. The HTTP/1.1 the filters are shown has nowhere to put one, so it is \ dropped rather than forwarded unread.", diff --git a/backend/proxysrc/src/proxy.rs b/backend/proxysrc/src/proxy.rs index 95fce34..bb46934 100644 --- a/backend/proxysrc/src/proxy.rs +++ b/backend/proxysrc/src/proxy.rs @@ -365,7 +365,7 @@ impl Proxy { // Per-connection accept errors (fd limits, a client gone between // SYN and accept) must not be fatal. Err(e) => { - eprintln!("[warn] [proxy] accept failed: {e}"); + crate::diag_throttled!("[warn] [proxy] failed accepts", 5, 10, "[warn] [proxy] accept failed: {e}"); // Out of descriptors, the condition does not clear by trying again // immediately: the loop spins on `accept` → `EMFILE` → print → // `accept`, which was measured burning about 40% of a core with four @@ -390,7 +390,7 @@ impl Proxy { // is that connections are arriving faster than they leave, and a line // each would be the flood arriving twice. if !self.stats.warned_limit.swap(true, Ordering::Relaxed) { - eprintln!( + crate::diag!( "[warn] [proxy] {limit} concurrent connections reached; further \ connections are being {} until it clears", if self.cfg.over_limit_forwards { "forwarded unfiltered" } else { "refused" }, @@ -424,7 +424,7 @@ impl Proxy { handle_connection(client, peer, cfg, chain, stats).await }; if let Err(e) = result { - eprintln!("[info] [proxy] connection from {peer} ended: {e}"); + crate::diag_throttled!("[info] [proxy] connections that ended with an error", 20, 10, "[info] [proxy] connection from {peer} ended: {e}"); } }); } @@ -465,7 +465,7 @@ async fn dial( Ok(Err(e)) => e.to_string(), _ => "timed out (is the return path diverted?)".to_string(), }; - eprintln!( + crate::diag!( "[warn] [proxy] cannot reach {upstream} as {}: {why}. \ Falling back to our own address — the service will not see real client IPs.", peer.ip() @@ -583,7 +583,7 @@ async fn handle_connection( // which of the two ends is wrong. It is nearly always the address: a port carrying // the cleartext site declared as the encrypted one. Naming the address and what // was expected turns an afternoon into a sentence. - eprintln!( + crate::diag_throttled!("[warn] [proxy] clients that spoke something other than TLS at an encrypted address", 1, 30, "[warn] [proxy] {peer} spoke something that is not TLS to {original}, which \ this service declares as its encrypted address. The connection is refused \ rather than carried to a service expecting HTTPS. If that address is meant \ diff --git a/backend/proxysrc/src/pyworker.rs b/backend/proxysrc/src/pyworker.rs index 8cfc1d1..054906f 100644 --- a/backend/proxysrc/src/pyworker.rs +++ b/backend/proxysrc/src/pyworker.rs @@ -261,7 +261,7 @@ impl PyWorkerRule { let _ = child.kill(); let _ = child.wait(); self.stats.restarts.fetch_add(1, Ordering::Relaxed); - eprintln!("[warn] [pyworker] '{}' killed: {why}", self.id); + crate::diag!("[warn] [pyworker] '{}' killed: {why}", self.id); } } diff --git a/backend/proxysrc/src/quic.rs b/backend/proxysrc/src/quic.rs index 556365a..4998c3e 100644 --- a/backend/proxysrc/src/quic.rs +++ b/backend/proxysrc/src/quic.rs @@ -265,7 +265,7 @@ impl QuicRelay { let relay = Arc::clone(&self); tokio::spawn(async move { relay.admit(incoming).await }); } - eprintln!("[warn] [quic] the endpoint for {} closed", self.upstream); + crate::diag!("[warn] [quic] the endpoint for {} closed", self.upstream); } /// Decide whether this connection is carried at all, and on what terms. @@ -296,7 +296,7 @@ impl QuicRelay { if over { self.stats.over_limit.fetch_add(1, Ordering::Relaxed); if !self.stats.warned_limit.swap(true, Ordering::Relaxed) { - eprintln!( + crate::diag!( "[warn] [quic] {limit} concurrent connections reached; further \ connections are being {} until it clears", if self.cfg.over_limit_forwards { @@ -331,7 +331,7 @@ impl QuicRelay { let _slot = slot; if let Err(e) = self.serve_connection(incoming, client, chain).await { - eprintln!("[info] [quic] connection from {client} ended: {e}"); + crate::diag_throttled!("[info] [quic] connections that ended with an error", 20, 10, "[info] [quic] connection from {client} ended: {e}"); } } @@ -394,7 +394,7 @@ impl QuicRelay { // and with HTTP/1.1 behind it is `h3`, which is what firegex itself is // offering to speak on the service's behalf. if let Some(protocol) = &agreed { - eprintln!( + crate::diag_throttled!("[warn] [quic] clients that refused the protocol the service chose", 5, 30, "[warn] [quic] {client} would not take `{}`, which is what {} chose: {e}", String::from_utf8_lossy(protocol), match behind { @@ -442,7 +442,7 @@ impl QuicRelay { Some(offered) if !offered.is_empty() => offered, _ => { if !self.warned_hello.swap(true, Ordering::Relaxed) { - eprintln!( + crate::diag!( "[warn] [quic] could not read which protocols {client} offered; \ offering {} to {} instead. A client speaking something else is \ refused by the handshake.", @@ -471,7 +471,7 @@ impl QuicRelay { .source_spoof_failures .fetch_add(1, Ordering::Relaxed); if !self.stats.warned_spoof.swap(true, Ordering::Relaxed) { - eprintln!( + crate::diag!( "[warn] [quic] cannot reach {} as {}: {e}. Falling back to our own \ address — the service will not see real client IPs.", self.upstream, @@ -594,7 +594,7 @@ impl QuicRelay { // have no HTTP/1.1 form to be sent in. Refused here, once, with a reason the // client can read — rather than carried into a translation nobody could // describe, or dropped silently. - eprintln!( + crate::diag_throttled!("[warn] [quic] connections that agreed on something other than HTTP/3", 5, 30, "[warn] [quic] {client}: this service is reached over HTTP/1.1, so only \ HTTP/3 can be carried to it — the connection agreed on something else.", ); @@ -721,7 +721,7 @@ fn log_end(client: SocketAddr, side: &str, e: quinn::ConnectionError) { quinn::ConnectionError::ApplicationClosed(_) | quinn::ConnectionError::LocallyClosed | quinn::ConnectionError::ConnectionClosed(_) => {} - other => eprintln!("[info] [quic] {client}: the {side} ended the connection: {other}"), + other => crate::diag_throttled!("[info] [quic] connections ended by one side", 20, 10, "[info] [quic] {client}: the {side} ended the connection: {other}"), } } @@ -867,7 +867,7 @@ impl Carrier { // reason as the first and the thousandth would too. if !complained { complained = true; - eprintln!("[info] [quic] {}: cannot carry a datagram: {e}", self.client); + crate::diag_throttled!("[info] [quic] connections whose datagrams could not be carried", 20, 10, "[info] [quic] {}: cannot carry a datagram: {e}", self.client); } } } @@ -906,7 +906,7 @@ impl Carrier { let (far_send, far_recv) = match far { Ok(pair) => pair, Err(e) => { - eprintln!("[info] [quic] cannot carry a stream through: {e}"); + crate::diag_throttled!("[info] [quic] streams that could not be carried", 20, 10, "[info] [quic] cannot carry a stream through: {e}"); chain.current().connection_closed(connection); return; } @@ -963,7 +963,7 @@ impl Carrier { let far_send = match far { Ok(send) => send, Err(e) => { - eprintln!("[info] [quic] cannot carry a stream through: {e}"); + crate::diag_throttled!("[info] [quic] streams that could not be carried", 20, 10, "[info] [quic] cannot carry a stream through: {e}"); chain.current().connection_closed(connection); return; } diff --git a/backend/proxysrc/src/quic_hello.rs b/backend/proxysrc/src/quic_hello.rs index 8d5ce0f..daaab46 100644 --- a/backend/proxysrc/src/quic_hello.rs +++ b/backend/proxysrc/src/quic_hello.rs @@ -20,7 +20,7 @@ //! ordinary) is reassembled from its CRYPTO frames across datagrams, in whatever order //! they arrive. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::fmt; use std::io::{self, IoSliceMut}; use std::net::SocketAddr; @@ -34,21 +34,34 @@ use quinn::{AsyncUdpSocket, UdpPoller}; use rustls::quic::{Suite, Version}; use rustls::Side; -/// How many clients' hellos may be in progress at once. Past it new ones are not read — -/// their connections fall back rather than this growing without end under a flood of -/// forged Initials, which cost the sender nothing: the keys are public, so a valid one can -/// be made from any address. With `MAX_HELLO` this bounds what a flood can make this -/// process hold to 16 MiB. +/// How many clients' hellos may be in progress at once. Past it the one that has been +/// waiting longest makes room — a hello is claimed within milliseconds of arriving, so the +/// oldest is the one least likely to still be wanted. Forged Initials cost the sender +/// nothing (the keys are public, so a valid one can be made from any address), and with +/// `MAX_HELLO` this bounds what a flood can make this process hold to 16 MiB. +/// +/// It used to refuse newcomers instead, so a thousand forged Initials every `KEPT_FOR` +/// were enough to stop every real client's hello from being read. const MAX_PENDING: usize = 1024; /// How long an unfinished or unclaimed hello is kept. const KEPT_FOR: Duration = Duration::from_secs(10); +/// How often the table is swept for hellos older than `KEPT_FOR`. Not on every packet: a +/// sweep walks the whole table, and every packet of a flood would pay for it. +const SWEEP_EVERY: Duration = Duration::from_secs(1); + /// Most CRYPTO bytes kept for one hello. A ClientHello is one or two kilobytes, a few with /// post-quantum key shares and a resumption ticket; this is a cap on what a peer can make /// this process hold, not a size anybody meets. const MAX_HELLO: usize = 16 * 1024; +/// Most CRYPTO fragments kept for one hello, whatever their size. Real clients split a +/// hello into a handful; the cap is what stops a peer sending pieces nobody counted — +/// empty ones, which `MAX_HELLO` did not see at all, grew one hello without end and had +/// every packet sort all of them again, inside the endpoint's receive path. +const MAX_FRAGMENTS: usize = 64; + /// One client's hello as it arrives. struct Pending { /// The connection ID its Initials were protected with. A new one means a new attempt — @@ -60,9 +73,56 @@ struct Pending { since: Instant, } +/// The hellos in progress, and the order they began in. +struct Table { + by_client: HashMap, + /// When each entry began, oldest first. An entry replaced or claimed leaves its line + /// here behind it; eviction skips a line whose time no longer matches the entry's. + order: VecDeque<(SocketAddr, Instant)>, + swept: Instant, +} + +impl Table { + fn new() -> Self { + Self { by_client: HashMap::new(), order: VecDeque::new(), swept: Instant::now() } + } + + fn sweep(&mut self, now: Instant) { + if now.duration_since(self.swept) < SWEEP_EVERY && self.by_client.len() < MAX_PENDING { + return; + } + self.swept = now; + self.by_client.retain(|_, p| now.duration_since(p.since) < KEPT_FOR); + let by_client = &self.by_client; + self.order.retain(|(from, since)| by_client.get(from).is_some_and(|p| p.since == *since)); + } + + /// Make room for one more by letting go of the entry that began longest ago. + fn evict_oldest(&mut self) { + while let Some((from, since)) = self.order.pop_front() { + if self.by_client.get(&from).is_some_and(|p| p.since == since) { + self.by_client.remove(&from); + return; + } + } + } + + fn start(&mut self, from: SocketAddr, dcid: Vec, now: Instant) -> &mut Pending { + self.order.push_back((from, now)); + self.by_client.insert(from, Pending { + dcid, + fragments: Vec::new(), + held: 0, + alpn: None, + since: now, + }); + self.by_client.get_mut(&from).expect("just inserted") + } +} + /// The hellos being read, by the address they came from. pub struct Hellos { - pending: Mutex>, + pending: Mutex, arrived: tokio::sync::Notify, suite: Suite, } @@ -88,7 +148,7 @@ impl Hellos { .and_then(|s| s.quic_suite()) .expect("TLS13_AES_128_GCM_SHA256 carries QUIC"); Self { - pending: Mutex::new(HashMap::new()), + pending: Mutex::new(Table::new()), arrived: tokio::sync::Notify::new(), suite, } @@ -106,39 +166,48 @@ impl Hellos { } fn absorb(&self, from: SocketAddr, packet: Initial) { - let Ok(mut pending) = self.pending.lock() else { return }; + let Ok(mut table) = self.pending.lock() else { return }; let now = Instant::now(); - if pending.len() >= MAX_PENDING { - pending.retain(|_, p| now.duration_since(p.since) < KEPT_FOR); - if pending.len() >= MAX_PENDING && !pending.contains_key(&from) { - return; + table.sweep(now); + // A new attempt — a Retry answered, another connection from the same port — or one + // kept past its time starts again; an address seen for the first time needs room. + let fresh = match table.by_client.get(&from) { + Some(entry) => entry.dcid != packet.dcid || now.duration_since(entry.since) >= KEPT_FOR, + None => { + if table.by_client.len() >= MAX_PENDING { + table.evict_oldest(); + } + true } - } - let entry = pending.entry(from).or_insert_with(|| Pending { - dcid: packet.dcid.clone(), - fragments: Vec::new(), - held: 0, - alpn: None, - since: now, - }); - if entry.dcid != packet.dcid { - *entry = Pending { - dcid: packet.dcid.clone(), - fragments: Vec::new(), - held: 0, - alpn: None, - since: now, - }; - } + }; + let entry = if fresh { + table.start(from, packet.dcid.clone(), now) + } else { + table.by_client.get_mut(&from).expect("looked up above") + }; if entry.alpn.is_some() { return; } + let mut grew = false; for (offset, data) in packet.crypto { - if entry.held + data.len() > MAX_HELLO { - return; + // An empty piece says nothing, and one past where any hello could reach is not + // part of one; neither is worth keeping. + if data.is_empty() { + continue; + } + let reaches = offset.saturating_add(data.len() as u64); + if reaches > MAX_HELLO as u64 + || entry.fragments.len() >= MAX_FRAGMENTS + || entry.held + data.len() > MAX_HELLO + { + break; } entry.held += data.len(); entry.fragments.push((offset, data)); + grew = true; + } + if !grew { + return; } if let Some(alpn) = client_hello_alpn(&entry.fragments) { entry.alpn = Some(alpn); @@ -148,9 +217,9 @@ impl Hellos { } fn take(&self, from: SocketAddr) -> Option>> { - let mut pending = self.pending.lock().ok()?; - if pending.get(&from)?.alpn.is_some() { - return pending.remove(&from).and_then(|p| p.alpn); + let mut table = self.pending.lock().ok()?; + if table.by_client.get(&from)?.alpn.is_some() { + return table.by_client.remove(&from).and_then(|p| p.alpn); } None } @@ -536,6 +605,63 @@ mod tests { assert_eq!(hellos.take(from()), None); } + fn kept_for(hellos: &Hellos, client: SocketAddr) -> Option { + hellos.pending.lock().unwrap().by_client.get(&client).map(|p| p.fragments.len()) + } + + /// Empty CRYPTO frames cost nothing to send and were counted by nothing: an Initial + /// full of them grew one client's hello by hundreds of pieces per packet, without end, + /// and every packet sorted all of them again inside the endpoint's receive path. + #[test] + fn empty_pieces_are_not_kept() { + let hellos = Hellos::new(); + let empties: Vec<(u64, &[u8])> = (0..300u64).map(|i| (i, &b""[..])).collect(); + let dcid = b"\x31\x32\x33\x34\x35\x36\x37\x38"; + for number in 0..50 { + hellos.observe(from(), &client_initial(dcid, number, &empties)); + } + assert_eq!(kept_for(&hellos, from()), Some(0)); + } + + /// Pieces that never join up from the start are held, but only so many of them. + #[test] + fn a_hello_is_kept_in_a_bounded_number_of_pieces() { + let hellos = Hellos::new(); + let dcid = b"\x41\x42\x43\x44\x45\x46\x47\x48"; + for number in 0..200u64 { + hellos.observe(from(), &client_initial(dcid, number, &[(100 + number * 2, b"x")])); + } + let kept = kept_for(&hellos, from()).unwrap(); + assert!(kept <= MAX_FRAGMENTS, "{kept} pieces kept for one hello"); + } + + #[test] + fn a_piece_beyond_any_hello_is_not_kept() { + let hellos = Hellos::new(); + let dcid = b"\x51\x52\x53\x54\x55\x56\x57\x58"; + hellos.observe(from(), &client_initial(dcid, 0, &[(MAX_HELLO as u64 + 1, b"far")])); + assert_eq!(kept_for(&hellos, from()), Some(0)); + } + + /// A table full of hellos nobody finished makes room rather than turning newcomers + /// away: refusing them meant a thousand forged Initials every few seconds stopped + /// every real client's hello from being read. + #[test] + fn a_full_table_makes_room_for_the_next_client() { + let hellos = Hellos::new(); + let unfinished = hello_with(&[b"h3"]); + for n in 0..MAX_PENDING as u32 { + let forged = SocketAddr::from(([10, (n >> 16) as u8, (n >> 8) as u8, n as u8], 443)); + let dcid = n.to_be_bytes().repeat(2); + hellos.observe(forged, &client_initial(&dcid, 0, &[(0, &unfinished[..10])])); + } + assert_eq!(hellos.pending.lock().unwrap().by_client.len(), MAX_PENDING); + let hello = hello_with(&[b"fgex-late"]); + hellos.observe(from(), &client_initial(b"\x61\x62\x63\x64\x65\x66\x67\x68", 0, &[(0, &hello)])); + assert_eq!(hellos.take(from()), Some(vec![b"fgex-late".to_vec()])); + assert!(hellos.pending.lock().unwrap().by_client.len() <= MAX_PENDING); + } + #[tokio::test] async fn a_hello_still_arriving_is_waited_for() { let hellos = Arc::new(Hellos::new()); diff --git a/backend/proxysrc/src/report.rs b/backend/proxysrc/src/report.rs index ac83e07..e8559f3 100644 --- a/backend/proxysrc/src/report.rs +++ b/backend/proxysrc/src/report.rs @@ -82,3 +82,175 @@ pub fn event(line: String) { pub fn blocked(id: &str) { event(format!("BLOCKED {id}")); } + +// --- what the datapath says about itself ------------------------------------------------ +// +// stderr is a pipe the backend reads too, and it fills the same way stdout did. Worse, it +// is written straight from the threads carrying traffic: a line per connection that ended +// with an error, or per client that spoke the wrong thing at a TLS address, was a write +// every runtime thread queued up on — `eprintln!` takes one lock for the whole process — +// and once the backend fell behind, one that blocked them all. Several of those lines are +// the client's to trigger at will. So the datapath's diagnostics go through a queue and a +// thread of their own, exactly as its reports do, and the ones a client can cause as often +// as it likes are counted rather than written past a few. + +static DIAG: OnceLock> = OnceLock::new(); +static DIAG_DROPPED: AtomicU64 = AtomicU64::new(0); + +fn diag_sender() -> &'static SyncSender { + DIAG.get_or_init(|| { + let (tx, rx) = sync_channel::(QUEUE); + std::thread::Builder::new() + .name("fgex-diag".to_string()) + .spawn(move || { + let stderr = std::io::stderr(); + while let Ok(first) = rx.recv() { + let mut out = stderr.lock(); + let _ = writeln!(out, "{first}"); + while let Ok(line) = rx.try_recv() { + let _ = writeln!(out, "{line}"); + } + let dropped = DIAG_DROPPED.swap(0, Ordering::Relaxed); + if dropped > 0 { + let _ = writeln!( + out, + "[warn] [report] {dropped} diagnostic line(s) were dropped: the \ + backend was not reading them fast enough" + ); + } + let _ = out.flush(); + } + }) + .expect("cannot start the thread that writes the engine's diagnostics"); + tx + }) +} + +/// A line about the datapath's own health. Never waits: dropped and counted when the +/// backend is too far behind, so nothing carrying traffic is held up by it. +pub fn diag(line: String) { + if let Err(TrySendError::Full(_)) = diag_sender().try_send(line) { + DIAG_DROPPED.fetch_add(1, Ordering::Relaxed); + } +} + +/// Milliseconds on a monotonic clock, never zero. +fn now_ms() -> u64 { + static START: OnceLock = OnceLock::new(); + START.get_or_init(Instant::now).elapsed().as_millis() as u64 + 1 +} + +/// At most `per_window` lines per window from one place in the code; the rest counted and +/// said as one line when the next window opens. +/// +/// For the lines a client can make the engine write as often as it likes — a connection +/// ending badly, a cleartext client at a TLS address, a malformed request. The first few +/// of a window are what an operator needs to see what is happening; the count says how +/// much of it there is; the thousands in between would only push everything else out of +/// a log that holds a few hundred lines. +pub struct Throttle { + /// Says what was held back, tag included: `[warn] [proxy] cleartext clients at a TLS + /// address`, so the summary is classified like the lines it stands for. + what: &'static str, + per_window: u64, + window_ms: u64, + opened_ms: AtomicU64, + said: AtomicU64, + held: AtomicU64, +} + +impl Throttle { + pub const fn new(what: &'static str, per_window: u64, window_secs: u64) -> Self { + Self { + what, + per_window, + window_ms: window_secs * 1000, + opened_ms: AtomicU64::new(0), + said: AtomicU64::new(0), + held: AtomicU64::new(0), + } + } + + pub fn say(&self, line: impl FnOnce() -> String) { + self.say_at(now_ms(), line, &mut diag); + } + + fn say_at(&self, now: u64, line: impl FnOnce() -> String, out: &mut dyn FnMut(String)) { + let opened = self.opened_ms.load(Ordering::Relaxed); + if now.saturating_sub(opened) >= self.window_ms + && self + .opened_ms + .compare_exchange(opened, now, Ordering::Relaxed, Ordering::Relaxed) + .is_ok() + { + self.said.store(0, Ordering::Relaxed); + let held = self.held.swap(0, Ordering::Relaxed); + if held > 0 { + out(format!( + "{}: {held} more like this in the last {}s, not written", + self.what, + now.saturating_sub(opened) / 1000, + )); + } + } + if self.said.fetch_add(1, Ordering::Relaxed) < self.per_window { + out(line()); + } else { + self.held.fetch_add(1, Ordering::Relaxed); + } + } +} + +/// `eprintln!` for the datapath: through the queue, never blocking. See [`diag`]. +#[macro_export] +macro_rules! diag { + ($($arg:tt)*) => { + $crate::report::diag(format!($($arg)*)) + }; +} + +/// [`diag!`], at most `per` times every `secs` seconds from this call site. See [`Throttle`]. +#[macro_export] +macro_rules! diag_throttled { + ($what:expr, $per:expr, $secs:expr, $($arg:tt)*) => {{ + static THROTTLE: $crate::report::Throttle = + $crate::report::Throttle::new($what, $per, $secs); + THROTTLE.say(|| format!($($arg)*)); + }}; +} + +#[cfg(test)] +mod tests { + use super::*; + + fn run(throttle: &Throttle, at: u64, n: usize) -> Vec { + let mut said = Vec::new(); + for i in 0..n { + throttle.say_at(at, || format!("line {i}"), &mut |line| said.push(line)); + } + said + } + + #[test] + fn a_window_says_its_first_lines_and_holds_the_rest() { + let throttle = Throttle::new("[info] [test] things", 3, 10); + assert_eq!(run(&throttle, 50_000, 100), vec!["line 0", "line 1", "line 2"]); + } + + #[test] + fn what_was_held_is_counted_when_the_next_window_opens() { + let throttle = Throttle::new("[warn] [test] things", 2, 10); + run(&throttle, 50_000, 7); + let said = run(&throttle, 61_000, 1); + assert_eq!(said.len(), 2, "{said:?}"); + assert!(said[0].starts_with("[warn] [test] things: 5 more"), "{said:?}"); + assert_eq!(said[1], "line 0"); + } + + #[test] + fn a_quiet_window_says_nothing_about_holding_anything_back() { + let throttle = Throttle::new("[info] [test] things", 2, 10); + run(&throttle, 50_000, 1); + assert_eq!(run(&throttle, 70_000, 1), vec!["line 0"]); + } +} diff --git a/backend/proxysrc/src/rules.rs b/backend/proxysrc/src/rules.rs index 0d64cb1..dc1d72b 100644 --- a/backend/proxysrc/src/rules.rs +++ b/backend/proxysrc/src/rules.rs @@ -152,7 +152,7 @@ impl Filter for HyperscanRule { // engine could not allocate scratch space, which is the traffic paying // for our problem. Err(e) => { - eprintln!("[warn] [filter] '{}' could not open a scanner ({e}): this connection goes unfiltered", self.label); + crate::diag_throttled!("[warn] [filter] connections a scanner could not be opened for", 5, 10, "[warn] [filter] '{}' could not open a scanner ({e}): this connection goes unfiltered", self.label); None } } @@ -187,7 +187,7 @@ impl FilterSession for HyperscanSession { // goes through. Reset the scanner so future chunks/datagrams can still be scanned. Err(e) => { let _ = self.scanner.reset(); - eprintln!("[warn] [filter] scan failed ({e}): scanner reset, chunk forwarded"); + crate::diag_throttled!("[warn] [filter] scans that failed", 5, 10, "[warn] [filter] scan failed ({e}): scanner reset, chunk forwarded"); Verdict::Accept } } diff --git a/backend/proxysrc/src/udp.rs b/backend/proxysrc/src/udp.rs index bff5a80..6c74664 100644 --- a/backend/proxysrc/src/udp.rs +++ b/backend/proxysrc/src/udp.rs @@ -46,23 +46,27 @@ const IDLE: Duration = Duration::from_secs(60); /// How often idle flows are swept. const SWEEP: Duration = Duration::from_secs(10); +/// How many of one flow's datagrams may wait for its verdicts. Past it they are dropped, +/// which is what the socket buffer would have done with them: a flow whose filter is slow +/// falls behind on its own, and the budget bounds what it can make this process hold. +const FLOW_QUEUE: usize = 32; + struct Flow { - /// The socket this relay dials the service from, bound to the client's own address - /// where the kernel allows it. - upstream: Arc, + /// This flow's datagrams, on their way to be judged and sent on — by a task of the + /// flow's own, so a slow verdict holds up this client and nobody else. + /// + /// The receive loop used to judge every datagram itself, one after the other, for + /// every client of the address: a filter taking a second over one client's datagram + /// was a second in which no other client's was even read, and the socket buffer + /// overflowing behind it dropped theirs. + to_service: tokio::sync::mpsc::Sender>, + forward_task: tokio::task::JoinHandle<()>, connection: ConnectionId, - /// Filter state for the client→service direction. Owned by the receive loop, which - /// is single-threaded, so it needs no lock. - sessions: ChainSessions, /// What this flow is judged by: the service's chain, or no chain at all for a flow /// admitted past the limit because the operator chose to forward what does not fit. /// Decided once, when the flow opens — the same trade a TCP connection admitted past /// the limit makes for its whole life. chain: ChainHandle, - /// Whether anything of this flow has been refused yet, in either direction. A refusal - /// is counted once per flow, because the number it goes into is compared with flows - /// seen, and one flow refusing a thousand datagrams is still one flow refused. - refused: Arc, reply_task: tokio::task::JoinHandle<()>, last_seen: Instant, /// This flow's place in the service's count of what it is carrying, released when the @@ -140,7 +144,7 @@ impl UdpRelay { // A per-datagram error (an ICMP port-unreachable landing on the // socket, most often) must not end the relay. Err(e) => { - eprintln!("[warn] [udp] receive failed: {e}"); + crate::diag_throttled!("[warn] [udp] failed receives", 5, 10, "[warn] [udp] receive failed: {e}"); continue; } }, @@ -173,7 +177,7 @@ impl UdpRelay { let chain = if over { self.stats.over_limit.fetch_add(1, Ordering::Relaxed); if !self.stats.warned_limit.swap(true, Ordering::Relaxed) { - eprintln!( + crate::diag!( "[warn] [udp] {limit} concurrent connections and flows \ reached; new flows are being {} until it clears", if self.over_limit_forwards { "carried unfiltered" } else { "dropped" }, @@ -202,38 +206,32 @@ impl UdpRelay { flows.entry(client).or_insert(flow) } Err(e) => { - eprintln!("[warn] [udp] cannot reach {} for {client}: {e}", self.upstream); + crate::diag_throttled!("[warn] [udp] flows that could not reach the service", 5, 10, "[warn] [udp] cannot reach {} for {client}: {e}", self.upstream); continue; } } } }; flow.last_seen = Instant::now(); - - // Re-read the handle every datagram, so a chain swapped in mid-flow takes - // effect without anyone losing their session. - let verdict = flow - .chain - .current() - .run(Direction::ClientToServer, &buf[..len], &mut flow.sessions) - .await; - let payload: &[u8] = match &verdict { - Verdict::Accept => &buf[..len], - Verdict::Reject(_) => { - if !flow.refused.swap(true, Ordering::Relaxed) { - self.stats.closed_by_filter.fetch_add(1, Ordering::Relaxed); - } - continue; + match flow.to_service.try_send(buf[..len].to_vec()) { + Ok(()) => {} + Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => { + crate::diag_throttled!("[warn] [udp] datagrams dropped behind a slow verdict", 5, 10, "[warn] [udp] {client} is sending faster than its datagrams are judged; one was dropped"); + } + // Its task is gone — it only ends when the flow does — so the flow goes + // too, and the client's next datagram opens a new one. + Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => { + if let Some(flow) = flows.remove(&client) { + flow.close(); + } } - }; - if let Err(e) = flow.upstream.send(payload).await { - eprintln!("[warn] [udp] cannot forward to {}: {e}", self.upstream); } } } - /// Start relaying one client's flow: a socket towards the service, and a task - /// carrying the answers back, both judged by `chain`. + /// Start relaying one client's flow: a socket towards the service, a task judging and + /// sending on what the client sends, and one carrying the answers back, all by + /// `chain`. async fn open(&self, client: SocketAddr, chain: ChainHandle, slot: Slot) -> io::Result { let upstream = if self.spoof_source { match crate::transparent::connect_as_udp(client.ip(), self.upstream, self.self_mark).await { @@ -241,7 +239,7 @@ impl UdpRelay { Err(e) => { self.stats.source_spoof_failures.fetch_add(1, Ordering::Relaxed); if !self.stats.warned_spoof.swap(true, Ordering::Relaxed) { - eprintln!( + crate::diag!( "[warn] [udp] cannot reach {} as {}: {e}. \ Falling back to our own address — the service will not see real client IPs.", self.upstream, @@ -270,6 +268,16 @@ impl UdpRelay { ); let refused = Arc::new(AtomicBool::new(false)); + let (to_service, datagrams) = tokio::sync::mpsc::channel(FLOW_QUEUE); + let forward_task = tokio::spawn(forward( + datagrams, + Arc::clone(&upstream), + self.upstream, + chain.clone(), + connection, + Arc::clone(&refused), + Arc::clone(&self.stats), + )); let reply_task = tokio::spawn(replies( Arc::clone(&upstream), Arc::clone(&self.listener), @@ -281,11 +289,10 @@ impl UdpRelay { )); Ok(Flow { - upstream, + to_service, + forward_task, connection, - sessions: ChainSessions::new(connection), chain, - refused, reply_task, last_seen: Instant::now(), _slot: slot, @@ -306,13 +313,54 @@ impl UdpRelay { .collect(); for addr in done { if let Some(flow) = flows.remove(&addr) { - flow.reply_task.abort(); - flow.chain.current().connection_closed(flow.connection); + flow.close(); } } } } +impl Flow { + /// Let go of everything this flow holds, and tell the filters it is over. + fn close(self) { + self.forward_task.abort(); + self.reply_task.abort(); + self.chain.current().connection_closed(self.connection); + } +} + +/// Judge one flow's datagrams in the order they arrived, and send on what is accepted. +async fn forward( + mut datagrams: tokio::sync::mpsc::Receiver>, + upstream: Arc, + service: SocketAddr, + chain: ChainHandle, + connection: ConnectionId, + refused: Arc, + stats: Arc, +) { + // This direction's own filter state, exactly as the TCP pumps keep theirs. + let mut sessions = ChainSessions::new(connection); + while let Some(datagram) = datagrams.recv().await { + // Re-read the handle every datagram, so a chain swapped in mid-flow takes effect + // without anyone losing their session. + let verdict = chain + .current() + .run(Direction::ClientToServer, &datagram, &mut sessions) + .await; + if let Verdict::Reject(_) = verdict { + // There is no connection to close, so refusing the datagram is the whole of + // what refusing can mean here. + if !refused.swap(true, Ordering::Relaxed) { + stats.closed_by_filter.fetch_add(1, Ordering::Relaxed); + } + continue; + } + if let Err(e) = upstream.send(&datagram).await { + crate::diag_throttled!("[warn] [udp] datagrams that could not be forwarded", 5, 10, "[warn] [udp] cannot forward to {service}: {e}"); + } + } +} + /// Carry one flow's answers back, inspecting them on the way. /// /// Replies leave through the *listener* socket, so conntrack rewrites them to appear @@ -341,7 +389,7 @@ async fn replies( // back again while its datagrams kept the flow from ever going idle. Err(e) if e.kind() == io::ErrorKind::ConnectionRefused => continue, Err(e) => { - eprintln!("[info] [udp] flow for {client} ended: {e}"); + crate::diag_throttled!("[info] [udp] flows that ended with an error", 20, 10, "[info] [udp] flow for {client} ended: {e}"); return; } }; @@ -361,7 +409,7 @@ async fn replies( } }; if let Err(e) = listener.send_to(payload, client).await { - eprintln!("[warn] [udp] cannot answer {client}: {e}"); + crate::diag_throttled!("[warn] [udp] answers that could not be delivered", 5, 10, "[warn] [udp] cannot answer {client}: {e}"); } } } @@ -424,7 +472,7 @@ impl UdpManager { map.insert(upstream, port); tokio::spawn(async move { if let Err(e) = relay.serve().await { - eprintln!("[fatal] [udp] relay for {upstream} died: {e}"); + crate::diag!("[fatal] [udp] relay for {upstream} died: {e}"); } }); Ok(port) diff --git a/backend/proxysrc/tests/udp.rs b/backend/proxysrc/tests/udp.rs index be17643..3a7aa97 100644 --- a/backend/proxysrc/tests/udp.rs +++ b/backend/proxysrc/tests/udp.rs @@ -9,7 +9,7 @@ use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; -use fgex_proxy::filter::{ChainHandle, FilterChain}; +use fgex_proxy::filter::{ChainHandle, Direction, Filter, FilterChain, FilterCtx, Verdict}; use fgex_proxy::proxy::ProxyStats; use fgex_proxy::spec::parse_filters; use fgex_proxy::udp::UdpRelay; @@ -216,3 +216,104 @@ async fn flows_and_refusals_are_counted_once_each() { "three refused datagrams of one flow are one refused flow" ); } + + +/// Takes its time over a client's datagram carrying `slow`, and none over anything else — +/// the echo coming back included, or each exchange would pay for it twice. +struct SlowOn; + +impl Filter for SlowOn { + fn name(&self) -> &str { + "slow-on" + } + fn inspect(&self, ctx: &FilterCtx<'_>) -> Verdict { + if ctx.direction == Direction::ClientToServer && ctx.chunk.windows(4).any(|w| w == b"slow") { + std::thread::sleep(Duration::from_millis(1500)); + } + Verdict::Accept + } +} + +/// One client's slow verdict holds up that client, and nobody else. +/// +/// The receive loop used to judge every datagram of every client itself, one after the +/// other: a filter taking a second and a half over one client's datagram was a second and +/// a half in which no other client of the address was even read. +#[tokio::test] +async fn a_slow_verdict_for_one_client_does_not_hold_up_another() { + let upstream = spawn_echo().await; + let chain = ChainHandle::new(FilterChain::new( + vec![Arc::new(SlowOn) as Arc], + Duration::from_secs(5), + )); + let relay = UdpRelay::bind( + "127.0.0.1:0".parse().unwrap(), + upstream, + chain, + None, + false, + 0, + false, + Arc::new(ProxyStats::default()), + ) + .await + .unwrap(); + let relay_addr = relay.local_addr().unwrap(); + tokio::spawn(relay.serve()); + + let slow = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + slow.send_to(b"slow please", relay_addr).await.unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + + let started = std::time::Instant::now(); + assert_eq!(ask(relay_addr, b"quick").await.as_deref(), Some(&b"quick"[..]), + "the other client was not answered while the first waited for its verdict"); + assert!(started.elapsed() < Duration::from_millis(600), + "the other client waited {:?} behind the slow verdict", started.elapsed()); + + let mut buf = vec![0u8; 64]; + let n = tokio::time::timeout(Duration::from_secs(3), slow.recv(&mut buf)) + .await + .expect("the slow client was never answered") + .unwrap(); + assert_eq!(&buf[..n], b"slow please"); +} + +/// Within one flow the order is kept: a datagram is not overtaken by the one after it +/// while its verdict is still being decided. +#[tokio::test] +async fn one_flows_datagrams_keep_their_order() { + let upstream = spawn_echo().await; + let chain = ChainHandle::new(FilterChain::new( + vec![Arc::new(SlowOn) as Arc], + Duration::from_secs(5), + )); + let relay = UdpRelay::bind( + "127.0.0.1:0".parse().unwrap(), + upstream, + chain, + None, + false, + 0, + false, + Arc::new(ProxyStats::default()), + ) + .await + .unwrap(); + let relay_addr = relay.local_addr().unwrap(); + tokio::spawn(relay.serve()); + + let client = UdpSocket::bind("127.0.0.1:0").await.unwrap(); + client.send_to(b"slow first", relay_addr).await.unwrap(); + client.send_to(b"then second", relay_addr).await.unwrap(); + let mut buf = vec![0u8; 64]; + let mut got = Vec::new(); + for _ in 0..2 { + let n = tokio::time::timeout(Duration::from_secs(3), client.recv(&mut buf)) + .await + .expect("an answer never came") + .unwrap(); + got.push(buf[..n].to_vec()); + } + assert_eq!(got, vec![b"slow first".to_vec(), b"then second".to_vec()]); +} diff --git a/backend/routers/firewall.py b/backend/routers/firewall.py index 591076d..fcfe534 100644 --- a/backend/routers/firewall.py +++ b/backend/routers/firewall.py @@ -1,4 +1,6 @@ +import re import sqlite3 +import traceback from fastapi import APIRouter, HTTPException from utils.sqlite import SQLite from utils import ip_parse, ip_family, socketio_emit @@ -58,8 +60,29 @@ async def shutdown(): async def refresh_frontend(additional:list[str]=[]): await socketio_emit(["firewall"]+additional) -async def apply_changes(): - await firewall.reload() +async def apply_changes(undo=None): + """Put what the database now says into the kernel, or put the database back. + + `undo` restores what was stored before the change. nft refusing the new firewall + leaves the old one in force (`FiregexTables.set` sends it as one batch), so the + database has to follow: kept, the refused configuration would be what the watcher + tries to put back every few seconds and what the next boot comes up with — failing + each time, with no firewall behind it. + """ + try: + await firewall.reload() + except Exception as e: + if undo is None: + raise + undo() + try: + await firewall.reload() + except Exception: + traceback.print_exc() + raise HTTPException( + status_code=400, + detail=f"nftables refused the new firewall, so nothing was changed: {e}", + ) await refresh_frontend() return {'status': 'ok'} @@ -72,8 +95,13 @@ async def get_settings(): @app.put("/settings", response_model=StatusMessageModel) async def set_settings(form: FirewallSettings): """Set the firewall settings""" + before = firewall.settings firewall.settings = form - return await apply_changes() + + def undo(): + firewall.settings = before + + return await apply_changes(undo) @app.get('/rules', response_model=RuleInfo) async def get_rule_list(): @@ -87,8 +115,13 @@ async def get_rule_list(): @app.post('/enable', response_model=StatusMessageModel) async def enable_firewall(): """Request enabling the firewall""" + was = firewall.enabled firewall.enabled = True - return await apply_changes() + + def undo(): + firewall.enabled = was + + return await apply_changes(undo) @app.post('/disable', response_model=StatusMessageModel) async def disable_firewall(): @@ -96,27 +129,40 @@ async def disable_firewall(): firewall.enabled = False return await apply_changes() +#: What an address field may hold when it is not an address: an interface name, as the +#: kernel takes one, or one ending in `*` to match every interface starting that way +#: (`br-*`). Fifteen characters at most, which is what `IFNAMSIZ` leaves for the name. +#: +#: Anything else used to be stored on the strength of containing no `/`, and nft then +#: refused the whole firewall over it — a name one character too long, or with a quote in +#: it, was enough to take every rule down, the policy with them. +INTERFACE = re.compile(r"[A-Za-z0-9_.:-]{1,15}|[A-Za-z0-9_.:-]{0,14}\*") + +def _address_or_interface(value: str, what: str) -> tuple[str, bool]: + """`value` normalised, and whether it is an IP address rather than an interface.""" + value = (value or "").strip() + if value == "": + return value, False + try: + return ip_parse(value), True + except ValueError: + pass + if not INTERFACE.fullmatch(value): + raise HTTPException( + status_code=400, + detail=f"Invalid {what} address {value!r}: it is neither an IP address nor an " + f"interface name (at most 15 letters, digits and '_.:-', optionally " + f"ending in '*')", + ) + return value, False + def parse_and_check_rule(rule:RuleModel): if rule.table == Table.MANGLE and rule.mode == Mode.FORWARD: raise HTTPException(status_code=400, detail="Mangle table does not support forward mode") - is_src_ip = is_dst_ip = True - - try: - rule.src = ip_parse(rule.src) - except ValueError: - is_src_ip = False - - try: - rule.dst = ip_parse(rule.dst) - except ValueError: - is_dst_ip = False - - if not is_src_ip and "/" in rule.src: # Slash is not allowed in ip interfaces names - raise HTTPException(status_code=400, detail="Invalid source address") - if not is_dst_ip and "/" in rule.dst: - raise HTTPException(status_code=400, detail="Invalid destination address") + rule.src, is_src_ip = _address_or_interface(rule.src, "source") + rule.dst, is_dst_ip = _address_or_interface(rule.dst, "destination") if is_src_ip and is_dst_ip and ip_family(rule.dst) != ip_family(rule.src): raise HTTPException(status_code=400, detail="Destination and source addresses must be of the same family") @@ -126,30 +172,43 @@ def parse_and_check_rule(rule:RuleModel): return rule +RULE_COLUMNS = ("active", "name", "proto", "src", "dst", "port_src_from", "port_dst_from", + "port_src_to", "port_dst_to", "action", "mode", "`table`") + +def _write_rules(rows: list[dict]) -> None: + """Replace every stored rule with `rows`, in order, in one transaction.""" + db.queries(["DELETE FROM rules"] + [ + ( + f"INSERT INTO rules (rule_id, {', '.join(RULE_COLUMNS)}) " + f"VALUES (?, {', '.join('?' for _ in RULE_COLUMNS)})", + rid, *(row[column.strip('`')] for column in RULE_COLUMNS), + ) + for rid, row in enumerate(rows) + ]) + @app.post('/rules', response_model=StatusMessageModel) async def add_new_service(form: RuleFormAdd): """Edit rule table""" rules = [parse_and_check_rule(ele) for ele in form.rules] + before = db.query(f"SELECT {', '.join(RULE_COLUMNS)} FROM rules ORDER BY rule_id;") + policy_before = firewall.policy try: - db.queries(["DELETE FROM rules"]+ - [(""" - INSERT INTO rules ( - rule_id, active, name, - proto, - src, dst, - port_src_from, port_dst_from, - port_src_to, port_dst_to, - action, mode, `table` - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ? ,?, ?, ?)""", - rid, ele.active, ele.name, - ele.proto, - ele.src, ele.dst, - ele.port_src_from, ele.port_dst_from, - ele.port_src_to, ele.port_dst_to, - ele.action, ele.mode, ele.table - ) for rid, ele in enumerate(rules)] - ) + _write_rules([ + { + "active": ele.active, "name": ele.name, "proto": ele.proto, + "src": ele.src, "dst": ele.dst, + "port_src_from": ele.port_src_from, "port_dst_from": ele.port_dst_from, + "port_src_to": ele.port_src_to, "port_dst_to": ele.port_dst_to, + "action": ele.action, "mode": ele.mode, "table": ele.table, + } + for ele in rules + ]) firewall.policy = form.policy.value except sqlite3.IntegrityError: raise HTTPException(status_code=400, detail="Error saving the rules: maybe there are duplicated rules") - return await apply_changes() + + def undo(): + _write_rules(before) + firewall.policy = policy_before + + return await apply_changes(undo) diff --git a/backend/routers/services.py b/backend/routers/services.py index 9829753..b752c46 100644 --- a/backend/routers/services.py +++ b/backend/routers/services.py @@ -36,11 +36,13 @@ from modules.services import stats from modules.services.logs import log_for from modules.services.models import (KIND, L4, MODE, PROTO, STATUS, Service, TRANSPORT, UPSTREAM) -from modules.services.nftables import FiregexTables, hijack_endpoint +from modules.services.nftables import FiregexTables, hijack_endpoint, one_address from modules.services import transports from modules.services.transports import PROXY_ENGINE, PYWORKER, UnsupportedChain from utils import ( PortType, + ip_family, + ip_parse, is_ip_parse, parse_ip_or_int, refactor_name, @@ -654,6 +656,11 @@ def _note_filtering(service_id: str) -> None: has a real gap in the middle of its history, and moving the mark forward would hide it; clearing it would make the chart claim the earlier hours never happened. """ + manager = firewall.services.get(service_id) + if manager is not None and not manager.active: + # Meant to run and not running — one that could not start at boot, say — has not + # begun filtering, whatever the stored status says. + return row = db.query( "SELECT s.filtering_since since, s.status status, " "(SELECT COUNT(*) FROM filters f WHERE f.service_id = s.service_id AND f.active = 1) n " @@ -688,19 +695,35 @@ async def _apply_chain(service_id: str, undo=None): the list, and the next start would fail for a reason they thought they had avoided. A refusal has to mean nothing changed. """ + undo = _once(undo) try: - await firewall.get(service_id).update_chain() + await firewall.get(service_id).update_chain(undo) except UnsupportedChain as e: - if undo: - undo() + undo() raise HTTPException(status_code=400, detail=str(e)) except Exception as e: - if undo: - undo() + undo() raise HTTPException(status_code=500, detail=str(e)) _note_filtering(service_id) +def _once(undo): + """`undo`, or nothing, made safe to call from both the manager and the caller. + + A rebuild that fails has the manager put the database back before it starts the + service again, and the caller then reaches its own `except` — where the undo it has + always run is now the second one. + """ + done = [] + + def run(): + if undo is not None and not done: + done.append(True) + undo() + + return run + + # --- services ---------------------------------------------------------------- SERVICE_QUERY = """ @@ -731,6 +754,7 @@ def _address_row(row: dict) -> dict: def _with_addresses(row: dict) -> dict: row = dict(row) + row["status"] = _running_status(row) row["addresses"] = [ _address_row(a) for a in _addresses(row["service_id"]) ] @@ -738,6 +762,20 @@ def _with_addresses(row: dict) -> dict: return row +def _running_status(row: dict) -> str: + """Whether the service is running, not whether it is meant to. + + The stored status is what the operator asked for, and what the next boot starts: a + service that could not come back up at boot kept `active` there, and the interface + showed it as active while nothing was filtering it — the one failure this whole module + is arranged to avoid. The reason is in its log, and on the list as its problem. + """ + manager = firewall.services.get(row["service_id"]) + if manager is None: + return row["status"] + return STATUS.ACTIVE if manager.active else STATUS.STOP + + @app.get("", response_model=list[ServiceModel]) async def get_services(): """Every service, with how many filters it runs and how much they have blocked.""" @@ -851,6 +889,24 @@ def _hijack_ip(transport: str, service_ip: str, form: AddressForm) -> str | None """ if str(transport) != TRANSPORT.EXTERNAL: return None + if form.proxy_ip and str(form.proxy_ip).strip(): + # Checked here, where the row is written: it used to be stored as typed and only + # parsed when the rules were built, so a mistyped endpoint was a service that + # would not start, with the reason in its log rather than in this answer. + proxy = str(form.proxy_ip).strip() + if not is_ip_parse(proxy) or _shown(ip_parse(proxy)) != one_address(ip_parse(proxy)): + raise HTTPException( + status_code=400, + detail=f"Your proxy's address has to be one IP address, not {proxy!r}", + ) + if is_ip_parse(service_ip) and ip_family(proxy) != ip_family(service_ip): + raise HTTPException( + status_code=400, + detail="Your proxy's address has to be in the same family as the address " + "it stands in for: the rewrite changes the destination and nothing " + "else, so an IPv4 packet cannot be sent to an IPv6 proxy.", + ) + form.proxy_ip = proxy if not form.proxy_port: return form.proxy_ip or None return hijack_endpoint(form.proxy_ip, service_ip) @@ -1107,6 +1163,14 @@ async def add_service(form: ServiceAddForm): detail="The external transport hands traffic to your own proxy and rewrites the " "source address on return, which requires a concrete IP address rather than an interface.", ) + # What the rows will hold, worked out before the first one is written: the addresses + # go in after the service, so a refusal half way through left the service and the + # addresses before it behind — holding the name, unknown to the running firewall. + for address in form.addresses: + try: + _hijack_ip(form.transport, parse_ip_or_int(address.ip_int), address) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) service_id = gen_id() try: db.query( @@ -1137,12 +1201,16 @@ async def add_service(form: ServiceAddForm): try: for address in form.addresses: _insert_address(service_id, form.proto, form.transport, address) - except sqlite3.IntegrityError as e: + except BaseException as e: # All of it or none: a service that came up on half the addresses the operator - # listed is one they would believe is protecting the other half. + # listed is one they would believe is protecting the other half. Whatever the + # reason — a refusal from a check on the way in used to leave the service and its + # first addresses behind, holding the name and unknown to the running firewall. db.query("DELETE FROM service_addresses WHERE service_id = ?;", service_id) db.query("DELETE FROM services WHERE service_id = ?;", service_id) - return {"status": _address_taken(e)} + if isinstance(e, sqlite3.IntegrityError): + return {"status": _address_taken(e)} + raise await firewall.reload() await refresh_frontend() return {"status": "ok", "service_id": service_id} @@ -1285,6 +1353,12 @@ async def edit_service(service_id: str, form: ServiceSettingsForm): transport = str(fields.get("transport", row["transport"])) old_addresses = _addresses(service_id) new_addresses = [_address_under(address, proto) for address in old_addresses] + if transport != TRANSPORT.EXTERNAL: + # The endpoint of the operator's own proxy is the hand-off layer's alone. Left + # on the rows it is read by nothing and still sits in the unique index, refusing + # another service's hand-off the endpoint this one no longer uses. + new_addresses = [{**address, "proxy_ip": None, "proxy_port": None} + for address in new_addresses] # What the addresses say has to mean something on the layer they land on. A port the # service is published on takes a layer that dials; on the others it would be read by @@ -1329,9 +1403,10 @@ async def edit_service(service_id: str, form: ServiceSettingsForm): )] for address in addresses: queries.append(( - "UPDATE service_addresses SET proto = ?, edge = ?, upstream = ? " - "WHERE address_id = ?;", - address["proto"], address["edge"], address["upstream"], address["address_id"], + "UPDATE service_addresses SET proto = ?, edge = ?, upstream = ?, " + "proxy_ip = ?, proxy_port = ? WHERE address_id = ?;", + address["proto"], address["edge"], address["upstream"], + address["proxy_ip"], address["proxy_port"], address["address_id"], )) return queries @@ -1414,6 +1489,7 @@ async def add_address(service_id: str, form: AddressForm): address_id = _insert_address(service_id, row["proto"], row["transport"], form) except sqlite3.IntegrityError as e: raise HTTPException(status_code=400, detail=_address_taken(e)) + @_once def undo(): # The row *and* the manager's copy of it: the manager re-read the list on its # way in, so leaving it there would mean the next start protecting an address @@ -1422,7 +1498,7 @@ async def add_address(service_id: str, form: AddressForm): firewall.get(service_id).reload_addresses() try: - await firewall.get(service_id).address_added(address_id) + await firewall.get(service_id).address_added(address_id, undo) except UnsupportedChain as e: undo() raise HTTPException(status_code=400, detail=str(e)) @@ -1524,7 +1600,11 @@ async def edit_address(service_id: str, address_id: str, form: AddressForm): await restore() raise HTTPException(status_code=400, detail=_address_taken(e)) try: - await manager.address_added(address_id) + # A rebuild failing on the new address stops the service before it finds out; + # the manager puts the old row back and starts it again on that. + await manager.address_added(address_id, lambda: write( + was["ip_int"], was["port"], was["proto"], was["edge"], was["target_port"], + was["upstream"], was["proxy_ip"], was["proxy_port"])) except UnsupportedChain as e: await restore() raise HTTPException(status_code=400, detail=str(e)) diff --git a/docs/firewall.md b/docs/firewall.md index 3d47811..e159230 100644 --- a/docs/firewall.md +++ b/docs/firewall.md @@ -7,13 +7,13 @@ Firewall Rules is a plain [nftables](https://netfilter.org/) allow/drop/reject r Each rule matches on: - **Protocol**: `tcp`, `udp`, `both` (adds a matching TCP and UDP rule) or `any` (matches all protocols, ports are ignored). -- **Source / Destination**: an IP/CIDR, or an interface name (e.g. `eth0`) instead of an address. +- **Source / Destination**: an IP/CIDR, or an interface name (e.g. `eth0`) instead of an address — at most 15 characters of letters, digits and `_.:-`, as the kernel takes them, optionally ending in `*` to match every interface starting that way (`br-*`). - **Source / Destination port range**: `from`/`to` for each side; leave as the full `1-65535` range to match any port. - **Direction** (`mode`): `in` (incoming traffic — the nftables `INPUT`/`PREROUTING` hook), `out` (outgoing traffic — `OUTPUT`/`POSTROUTING`), or `forward` (traffic routed through this host, `FORWARD` — only meaningful with the `filter` table, see below). - **Table**: `filter` — standard firewall rules evaluated at the normal input/output/forward hooks; `mangle` — rules evaluated earlier in the pipeline (`prerouting`/`postrouting`, before routing decisions), useful when a rule needs to run before other processing (e.g. before a an external-proxy [service](services.md) or TLS rule takes effect on the same traffic). - **Action**: `accept`, `drop`, or `reject` (closes the connection with an ICMP/RST reply instead of silently dropping it). `reject` on outgoing (`out`) traffic isn't meaningful — Firegex silently treats it as `drop` in that direction. -Rules are evaluated in order; the first match wins. Traffic that matches no rule falls through to the global **policy** (`accept`/`drop`/`reject`), which applies to incoming and forwarded traffic — outgoing traffic is always allowed by default regardless of the policy, so Firegex itself is never at risk of losing its own outbound connectivity by misconfiguring rules. +Rules are evaluated in order; the first match wins. A change nftables refuses is refused as a whole: the firewall already in force stays exactly as it was, and so does the saved configuration. Traffic that matches no rule falls through to the global **policy** (`accept`/`drop`/`reject`), which applies to incoming and forwarded traffic — outgoing traffic is always allowed by default regardless of the policy, so Firegex itself is never at risk of losing its own outbound connectivity by misconfiguring rules. ### Global settings diff --git a/docs/services.md b/docs/services.md index 96aa51f..053ed01 100644 --- a/docs/services.md +++ b/docs/services.md @@ -285,6 +285,17 @@ What UDP on the proxy layer provides: to observe, so a timeout is the only thing that can end one. - **Both directions inspected**, and replies leave through the listener socket so conntrack rewrites them to appear from the address the client dialled. +- **One slow client holds up only itself.** Each flow's datagrams are judged in order by + a task of their own, so a filter taking its time over one client's datagram does not + keep the relay from reading everyone else's. +- **Flows already open follow the rules in force.** The kernel remembers where a UDP flow + was sent for as long as datagrams keep arriving, so firegex tells it to forget the + flows of an address whenever what steers them changes: one open before the service + started is filtered from its next datagram instead of going round the protection, one + carried across a restart moves to the new engine and its filters, one whose engine died + is answered again as soon as the engine is back, and once the service is stopped they go + straight to it. A QUIC connection an engine is still carrying is left with it until it + closes, because moving it to another engine would end it. `REJECT` means something narrower here: there is no connection to close, so the datagram is simply not forwarded, and the next one from that client is judged afresh. diff --git a/tests/helpers/udpserver.py b/tests/helpers/udpserver.py index f0d3bed..eb5cfbf 100644 --- a/tests/helpers/udpserver.py +++ b/tests/helpers/udpserver.py @@ -86,6 +86,16 @@ class UdpEcho: finally: sock.close() + def flow(self, timeout: float = 1.5) -> "HeldFlow": + """One client socket kept across several exchanges — one flow, to conntrack. + + `exchange` opens a socket per datagram, so every one of them is a flow of its own + and meets whatever rules are in force when it is sent. What a long-lived client + does is keep its socket, and that is the case where what conntrack remembers + decides where the next datagram goes. + """ + return HeldFlow(self, timeout) + def answered_from(self, payload: bytes, timeout: float = 1.5) -> tuple | None: """Which address the answer came back from. @@ -103,3 +113,39 @@ class UdpEcho: return None finally: sock.close() + + +class HeldFlow: + """A UDP client that keeps its source port for as long as it is open. See `flow`.""" + + def __init__(self, echo: UdpEcho, timeout: float): + self.echo = echo + self.sock = socket.socket(socket.AF_INET6 if echo.ipv6 else socket.AF_INET, + socket.SOCK_DGRAM) + self.sock.settimeout(timeout) + + def exchange(self, payload: bytes) -> bytes | None: + """Send one datagram on this flow and return its echo, or `None` if none came. + + Answers to earlier datagrams arriving late are passed over rather than taken for + this one's, which on one socket is a real possibility. + """ + try: + self.sock.sendto(payload, (self.echo.host, self.echo.port)) + deadline = time.monotonic() + (self.sock.gettimeout() or 1.5) + while time.monotonic() < deadline: + data = self.sock.recvfrom(65535)[0] + if data == payload: + return data + except (TimeoutError, OSError): + pass + return None + + def close(self): + self.sock.close() + + def __enter__(self) -> "HeldFlow": + return self + + def __exit__(self, *exc): + self.close() diff --git a/tests/integration/test_service_edit.py b/tests/integration/test_service_edit.py index 89aabb0..b0df088 100644 --- a/tests/integration/test_service_edit.py +++ b/tests/integration/test_service_edit.py @@ -99,3 +99,21 @@ def test_a_running_service_whose_edit_will_not_start_is_put_back(api, protected, now = api.services_get(service_id) assert (now["status"], now["proto"]) == ("active", "tcp"), str(now) assert channel.gets_through(b"after"), "the service was not started again as it was" + + +def test_leaving_the_hand_off_layer_gives_the_proxy_endpoint_back(api, service): + """The endpoint of the operator's own proxy is the hand-off layer's alone. It stayed + on the rows when a service moved off that layer — read by nothing, and still in the + unique index, refusing another service's hand-off the endpoint.""" + from helpers.net import free_port + + endpoint = free_port() + first = service(f"handoff-a-{endpoint}", "127.0.0.1", free_port(), "external", + proxy_ip="127.0.0.1", proxy_port=endpoint) + assert api.services_edit(first, transport="proxy") + address = api.services_addresses(first)[0] + assert address["proxy_port"] is None and address["proxy_ip"] is None, address + + second = service(f"handoff-b-{endpoint}", "127.0.0.1", free_port(), "external", + proxy_ip="127.0.0.1", proxy_port=endpoint) + assert second, "the endpoint was still held by a service no longer handing off" diff --git a/tests/integration/test_service_lifecycle.py b/tests/integration/test_service_lifecycle.py index 93857cd..73bc517 100644 --- a/tests/integration/test_service_lifecycle.py +++ b/tests/integration/test_service_lifecycle.py @@ -145,3 +145,21 @@ def test_a_restart_hands_over_without_cutting_connections(api, protected, proxy_ channel = Channel(server, port, proxy_layer.ipv6) assert channel.gets_through(b"harmless"), "the new engine did not take new connections" assert channel.is_blocked(b"carrying BLOCKME"), "the new engine came up without the filters" + + +def test_a_refused_address_leaves_no_half_made_service_behind(api): + """The addresses are written after the service, and a refusal half way through left + the service and the addresses before it — holding the name, unknown to the running + firewall, and failing every start with a bare 500.""" + name = "half-made" + why = api.services_add_error( + name=name, transport="proxy", proto="tcp", + addresses=[{"ip_int": "127.0.0.1", "port": 40123}, + {"ip_int": "not an address!", "port": 40124}], + ) + assert why, "an address that is neither an IP nor an interface was accepted" + assert name not in [s["name"] for s in api.services_list()], \ + "the refused service was left behind" + service_id = api.services_add(name, "127.0.0.1", 40123, "proxy") + assert service_id, "the name of a refused service stayed taken" + api.services_delete(service_id) diff --git a/tests/integration/test_udp.py b/tests/integration/test_udp.py index b4867b0..23e6e6c 100644 --- a/tests/integration/test_udp.py +++ b/tests/integration/test_udp.py @@ -250,3 +250,94 @@ def test_a_queued_udp_service_keeps_flows_up_to_its_own_limit(api, service, udp_ assert answered is None, "with no limit, a flow lost its state to two others" else: assert answered == b"once more", "past the limit, the oldest flow kept its state" + + +# --- flows conntrack is still carrying ------------------------------------------------- +# +# Conntrack translates a flow at its first packet, and a UDP flow lasts for as long as +# datagrams keep arriving. So on the proxy layer, which is a translation, what a flow +# already had decided where each of its datagrams went — until firegex started forgetting +# the flows whose answer the rules had just changed. + + +def _proxy_only(layer: Layer): + if layer.transport != "proxy": + pytest.skip("NFQUEUE queues every packet as it goes by; conntrack decides nothing") + + +def test_a_flow_open_before_the_service_started_meets_its_filters(api, service, + udp_stand_in, udp_layer): + """Nothing redirected it, and nothing would: it went on straight to the service, + unfiltered, for as long as it kept talking — a way past the protection for anybody + who opened one before it went up.""" + echo = udp_stand_in(udp_layer.ipv6) + service_id = service(f"udp-early-{echo.port}", udp_layer.ip, echo.port, + udp_layer.transport, proto="udp") + add_regex_filter(api, service_id, "DENYME") + with echo.flow() as flow: + assert flow.exchange(b"before the service") == b"before the service" + start_and_settle(api, service_id, wait=1.2) + assert flow.exchange(b"carrying DENYME") is None, \ + "a flow older than the service went on unfiltered" + assert flow.exchange(b"still talking") == b"still talking", \ + "the flow stopped being answered once the service started" + + +def test_a_flow_carried_across_a_restart_meets_the_new_filters(api, udp_service, udp_layer): + """A restart hands the connections over to the engine replacing the old one; a flow + stayed with the old one instead, under the filters it started with, for as long as + it kept talking.""" + _proxy_only(udp_layer) + service_id, echo = udp_service + with echo.flow() as flow: + assert flow.exchange(b"before the restart") == b"before the restart" + # Anything but its name rebuilds the datapath. + assert api.services_edit(service_id, max_connections=500) + time.sleep(1.2) + add_regex_filter(api, service_id, "LATECOMER", name="later") + time.sleep(0.5) + assert flow.exchange(b"carrying LATECOMER") is None, \ + "the flow was still carried by the engine from before the restart" + assert flow.exchange(b"after the restart") == b"after the restart" + + +def _engines_relaying_to(port: int) -> list[int]: + """The proxy engines holding a UDP relay towards `port`, by their environment.""" + import os + found = [] + for pid in filter(str.isdigit, os.listdir("/proc")): + try: + with open(f"/proc/{pid}/environ", "rb") as f: + environ = f.read().split(b"\0") + except OSError: + continue + for entry in environ: + if entry.startswith(b"FGEX_PROXY_UDP=") and f":{port}|".encode() in entry: + found.append(int(pid)) + return found + + +@pytest.mark.root +def test_a_flow_is_answered_again_after_its_engine_is_killed(api, udp_service, udp_layer): + """Conntrack went on sending the flow to the relay port of an engine that no longer + existed, every datagram refreshing the entry that did so: a client that kept talking + was cut off from the service until it fell silent for two minutes.""" + import os + import signal + + _proxy_only(udp_layer) + _, echo = udp_service + with echo.flow() as flow: + assert flow.exchange(b"before the crash") == b"before the crash" + engines = _engines_relaying_to(echo.port) + if not engines: + pytest.skip("cannot find the engine to kill; is the instance elsewhere?") + for pid in engines: + os.kill(pid, signal.SIGKILL) + answered, deadline, n = None, time.monotonic() + 10, 0 + while answered is None and time.monotonic() < deadline: + n += 1 + answered = flow.exchange(f"after the crash {n}".encode()) + assert answered is not None, "the flow was never answered again after the engine died" + assert flow.exchange(b"carrying DENYME") is None, \ + "the engine that came back is not filtering the flow" diff --git a/tests/unit/test_conntrack.py b/tests/unit/test_conntrack.py new file mode 100644 index 0000000..39ffffa --- /dev/null +++ b/tests/unit/test_conntrack.py @@ -0,0 +1,287 @@ +"""Which UDP flows are forgotten, and what the kernel is asked to forget them by. + +Conntrack translates a flow at its first packet, and a UDP flow lasts for as long as +datagrams keep arriving: on the proxy layer, which is a translation, a flow older than a +service went on unfiltered for as long as it talked, one carried across a restart stayed +with the engine being retired, and one whose engine died went nowhere until the client +fell silent. Forgetting the flow is the fix, and forgetting the wrong one is a new bug — +so the decision is pinned here without a kernel, and so is the wire format it is sent in. +""" + +import asyncio +import ipaddress +import socket +import struct + +import pytest + +from modules.services import conntrack as ct +from modules.services import firewall +from modules.services.models import TRANSPORT, Address, Service + +NESTED = 0x8000 +SELF_MARK = 0x133A + + +def _tuple(kind: int, src: str, dst: str, sport: int, dport: int, proto: int = 17) -> bytes: + v6 = ipaddress.ip_address(src).version == 6 + ip = (ct.attr(ct.CTA_IP_V6_SRC if v6 else ct.CTA_IP_V4_SRC, ipaddress.ip_address(src).packed) + + ct.attr(ct.CTA_IP_V6_DST if v6 else ct.CTA_IP_V4_DST, ipaddress.ip_address(dst).packed)) + ports = (ct.attr(ct.CTA_PROTO_NUM, bytes([proto])) + + ct.attr(ct.CTA_PROTO_SRC_PORT, struct.pack("!H", sport)) + + ct.attr(ct.CTA_PROTO_DST_PORT, struct.pack("!H", dport))) + return ct.attr(kind | NESTED, ct.attr(ct.CTA_TUPLE_IP | NESTED, ip) + + ct.attr(ct.CTA_TUPLE_PROTO | NESTED, ports)) + + +def entry(orig: tuple, reply: tuple, status: int = 0, mark: int = 0, zone: int | None = None, + proto: int = 17) -> bytes: + """One conntrack entry, as a dump carries it.""" + family = socket.AF_INET6 if ipaddress.ip_address(orig[0]).version == 6 else socket.AF_INET + payload = (_tuple(ct.CTA_TUPLE_ORIG, *orig, proto=proto) + + _tuple(ct.CTA_TUPLE_REPLY, *reply, proto=proto) + + ct.attr(ct.CTA_STATUS, struct.pack("!I", status)) + + ct.attr(ct.CTA_MARK, struct.pack("!I", mark)) + + (ct.attr(ct.CTA_ZONE, struct.pack("!H", zone)) if zone is not None else b"")) + return ct.message(0, 0x2, 7, family, payload) + + +def flow_of(raw: bytes) -> ct.Flow: + [(_, payload)] = list(ct.messages(raw)) + parsed = ct.parse_flow(payload) + assert parsed is not None + return parsed + + +# A client that dialled the protected address; the redirect sent it to a relay on 40000. +CLIENT = ("192.0.2.7", "10.0.0.1", 5555, 53) +REDIRECTED = flow_of(entry(CLIENT, ("10.0.0.1", "192.0.2.7", 40000, 5555), status=ct.IPS_DST_NAT)) +# The same, from before the service started: nothing rewrote it. +UNSTEERED = flow_of(entry(CLIENT, ("10.0.0.1", "192.0.2.7", 53, 5555))) +# The engine's own dial to the service, from the client's address. +ENGINES_OWN = flow_of(entry(("192.0.2.7", "10.0.0.1", 33333, 53), + ("10.0.0.1", "192.0.2.7", 53, 33333), mark=SELF_MARK)) +ELSEWHERE = flow_of(entry(("192.0.2.7", "10.0.0.1", 5555, 54), ("10.0.0.1", "192.0.2.7", 54, 5555))) + + +# --- the wire ----------------------------------------------------------------------- + + +def test_an_entry_is_read_back_as_the_kernel_sent_it(): + flow = REDIRECTED + assert flow.family == socket.AF_INET and flow.l4proto == 17 + assert (str(flow.orig_src), str(flow.orig_dst)) == ("192.0.2.7", "10.0.0.1") + assert (flow.orig_sport, flow.orig_dport) == (5555, 53) + assert (str(flow.reply_src), flow.reply_sport) == ("10.0.0.1", 40000) + assert flow.redirected and not UNSTEERED.redirected + + +def test_an_ipv6_entry_and_its_zone_are_read(): + flow = flow_of(entry(("2001:db8::7", "2001:db8::1", 5555, 443), + ("2001:db8::1", "2001:db8::7", 443, 5555), zone=3)) + assert flow.family == socket.AF_INET6 + assert str(flow.orig_dst) == "2001:db8::1" and flow.orig_dport == 443 + assert flow.raw_zone, "the zone was not kept, and a delete without it names another entry" + + +def test_a_delete_names_the_entry_by_its_original_tuple_and_zone(): + flow = flow_of(entry(CLIENT, ("10.0.0.1", "192.0.2.7", 40000, 5555), zone=3)) + [(kind, payload)] = list(ct.messages(ct.delete_request(flow, 9))) + assert kind == (ct.NFNL_SUBSYS_CTNETLINK << 8) | ct.IPCTNL_MSG_CT_DELETE + assert payload[0] == socket.AF_INET + sent = [k for k, _, _ in ct.attrs(payload[4:])] + assert sent == [ct.CTA_TUPLE_ORIG, ct.CTA_ZONE], sent + assert payload[4:] == flow.raw_orig + flow.raw_zone + + +def test_several_messages_in_one_read_are_all_read(): + raw = entry(CLIENT, ("10.0.0.1", "192.0.2.7", 1, 5555)) * 3 + assert len(list(ct.messages(raw))) == 3 + + +# --- which flows are forgotten --------------------------------------------------------- + + +def chosen(flow, moved=(), unsteered=(), bound=frozenset({40000})): + return ct.chooser(moved=ct.targets(moved), unsteered=ct.targets(unsteered), + spare_mark=SELF_MARK, bound=set(bound))(flow) + + +def test_a_datagram_address_forgets_every_flow_headed_for_it(): + """A datagram flow changes engines between two datagrams and loses nothing but the + filter's own state for it — so the ones an old engine carries move too.""" + target = [("10.0.0.1/32", 53)] + assert chosen(UNSTEERED, moved=target) + assert chosen(REDIRECTED, moved=target) + + +def test_the_engines_own_dials_are_never_forgotten(): + assert not chosen(ENGINES_OWN, moved=[("10.0.0.1/32", 53)]) + assert not chosen(ENGINES_OWN, unsteered=[("10.0.0.1/32", 53)], bound=set()) + + +def test_a_flow_headed_elsewhere_is_left_alone(): + assert not chosen(ELSEWHERE, moved=[("10.0.0.1/32", 53)]) + assert not chosen(REDIRECTED, moved=[("10.0.0.2/32", 53)]) + assert not chosen(REDIRECTED, moved=[("2001:db8::/32", 53)]) + + +def test_a_range_is_matched_as_a_range(): + assert chosen(REDIRECTED, moved=[("10.0.0.0/24", 53)]) + + +def test_a_quic_address_keeps_what_a_live_engine_carries(): + """Moving a QUIC connection to another engine ends it; one still draining is carrying + it. Only what nothing redirected, or what points at a port nobody listens on, goes.""" + target = [("10.0.0.1/32", 53)] + assert chosen(UNSTEERED, unsteered=target) + assert not chosen(REDIRECTED, unsteered=target, bound={40000}) + assert chosen(REDIRECTED, unsteered=target, bound={40001}) + + +def test_what_an_engine_leaves_behind_is_recognised_by_its_ports(monkeypatch): + seen = {} + monkeypatch.setattr(ct, "delete_where", lambda pick: seen.setdefault("pick", pick) and 0) + local = {ipaddress.ip_address("10.0.0.1")} + ct.forget_udp_redirected_to([40000], local) + pick = seen["pick"] + assert pick(REDIRECTED) + assert not pick(UNSTEERED), "a flow nothing redirected is not the engine's" + assert not pick(flow_of(entry(CLIENT, ("10.0.0.1", "192.0.2.7", 40001, 5555), + status=ct.IPS_DST_NAT))) + # Forwarded to a container that happens to answer from the same port number. + assert not pick(flow_of(entry(CLIENT, ("172.17.0.2", "192.0.2.7", 40000, 5555), + status=ct.IPS_DST_NAT))) + + +def test_nothing_to_forget_asks_the_kernel_nothing(monkeypatch): + monkeypatch.setattr(ct, "delete_where", lambda pick: pytest.fail("the kernel was asked")) + assert ct.forget_udp(spare_mark=SELF_MARK) == 0 + assert ct.forget_udp_redirected_to([], set()) == 0 + + +# --- when the service manager asks ----------------------------------------------------- + + +class _Rules: + def __init__(self, guarded): + self.guarded = guarded + self.released = [] + + def add(self, *args, **kwargs): + pass + + def delete(self, *args, keep_guards=False, **kwargs): + return set(self.guarded) if keep_guards else set() + + def release_guards(self, ports): + self.released.append(set(ports)) + + +class _Log: + def add(self, *args, **kwargs): + pass + + def query(self, *args, **kwargs): + return [] + + +class _Engine: + async def start(self, chain): + return {} + + async def stop(self): + pass + + async def retire(self, keep_filtering, then=None): + if then: + then() + + +@pytest.fixture +def asked(monkeypatch): + calls = [] + monkeypatch.setattr(firewall.conntrack, "forget_udp", + lambda **kw: calls.append(("forget", kw)) or 0) + monkeypatch.setattr(firewall.conntrack, "forget_udp_redirected_to", + lambda ports, local: calls.append(("gone", sorted(ports))) or 0) + monkeypatch.setattr(firewall.conntrack, "local_addresses", lambda: set()) + monkeypatch.setattr(firewall, "log_for", lambda service_id: _Log()) + monkeypatch.setattr(firewall.transports, "build", lambda srv, **kw: _Engine()) + rules = _Rules({("udp", 40000), ("udp", 40001), ("tcp", 40002)}) + monkeypatch.setattr(firewall, "nft", rules) + return calls, rules + + +def manager(transport=TRANSPORT.PROXY) -> firewall.ServiceManager: + srv = Service(service_id="s", name="n", status="stop", proto="http", transport=transport) + srv.addresses = [ + Address("plain", "s", "10.0.0.1/32", 53, proto="udp", edge="udp"), + Address("quic", "s", "10.0.0.1/32", 443, proto="udp", edge="quic"), + Address("tcp", "s", "10.0.0.1/32", 80, proto="tcp", edge="tcp"), + ] + return firewall.ServiceManager(srv, _Log()) + + +async def settled(service: firewall.ServiceManager): + while service._background: + await asyncio.gather(*list(service._background)) + + +def test_starting_claims_every_udp_flow_and_spares_the_quic_ones_an_engine_carries(asked): + calls, _ = asked + asyncio.run(manager().enable()) + assert calls == [("forget", {"moved": [("10.0.0.1/32", 53)], + "unsteered": [("10.0.0.1/32", 443)], + "spare_mark": firewall.PROXY_SELF_MARK})], calls + + +def test_stopping_lets_the_datagram_flows_go_and_the_engines_ports_after_it(asked): + calls, rules = asked + service = manager() + + async def run(): + await service.enable() + calls.clear() + await service.disable() + await settled(service) + + asyncio.run(run()) + assert calls[0] == ("forget", {"moved": [("10.0.0.1/32", 53)], + "spare_mark": firewall.PROXY_SELF_MARK}), calls + assert calls[1] == ("gone", [40000, 40001]), "the engine's UDP ports were not let go" + assert rules.released, "the engine's ports stayed guarded after it went" + + +def test_a_restart_leaves_the_flows_for_the_engine_replacing_it(asked): + """Released at the stop half of a restart, a flow would reach the service unfiltered + until the start half claimed it again.""" + calls, _ = asked + service = manager() + + async def run(): + await service.enable() + calls.clear() + await service.restart() + await settled(service) + + asyncio.run(run()) + forgets = [kw for kind, kw in calls if kind == "forget"] + assert forgets == [{"moved": [("10.0.0.1/32", 53)], "unsteered": [("10.0.0.1/32", 443)], + "spare_mark": firewall.PROXY_SELF_MARK}], \ + f"only the start half should have touched the flows: {calls}" + assert ("gone", [40000, 40001]) in calls, "the retired engine's ports were not let go" + + +def test_the_other_layers_leave_conntrack_alone(asked): + calls, _ = asked + service = manager(TRANSPORT.NFQUEUE) + service.srv.proto = "udp" + + async def run(): + await service.enable() + await service.disable() + await settled(service) + + asyncio.run(run()) + assert [c for c in calls if c[0] == "forget"] == [], calls diff --git a/tests/unit/test_firewall_rules.py b/tests/unit/test_firewall_rules.py index 703ad30..b7ce6fd 100644 --- a/tests/unit/test_firewall_rules.py +++ b/tests/unit/test_firewall_rules.py @@ -64,3 +64,107 @@ def test_container_traffic_is_left_to_the_container_runtime(): assert all(c["add"]["rule"]["chain"] == "fgex_forward" for c in commands) assert FiregexTables().dnat_rules(settings(allow_dnat=False)) == [] + + +# --- replacing the firewall whole, or not at all ------------------------------ + + +def test_the_firewall_is_replaced_in_one_batch_with_its_teardown(monkeypatch): + """The teardown used to be a batch of its own, sent first. When nft then refused a rule + the old tables were already gone and the new ones never arrived: a default-deny + firewall became no firewall. One batch is applied whole or not at all.""" + tables = FiregexTables() + sent = [] + + def cmd(*commands): + sent.append(commands) + return {"nftables": []} + + def raw_cmd(*commands): + sent.append(commands) + return 0, {"nftables": []}, "" + + monkeypatch.setattr(tables, "cmd", cmd) + monkeypatch.setattr(tables, "raw_cmd", raw_cmd) + tables.set([rule()], policy=Action.DROP, opt=settings()) + + assert len(sent) == 1, f"the firewall went out in {len(sent)} batches" + batch = sent[0] + deleted = [i for i, c in enumerate(batch) if "delete" in c] + chains = [i for i, c in enumerate(batch) if "chain" in c.get("add", {})] + assert deleted and chains, "the batch neither tears down nor builds" + assert max(deleted) < min(chains), "the teardown has to come before what replaces it" + + +def test_a_refused_firewall_leaves_nothing_half_applied(monkeypatch): + """nft refusing the batch must not leave a separate teardown behind it.""" + tables = FiregexTables() + sent = [] + + def cmd(*commands): + sent.append(commands) + raise Exception("Error: syntax error") + + monkeypatch.setattr(tables, "cmd", cmd) + monkeypatch.setattr(tables, "raw_cmd", lambda *c: sent.append(c) or (1, {}, "no")) + try: + tables.set([rule()], policy=Action.DROP, opt=settings()) + except Exception: + pass + assert len(sent) == 1, "something was sent to the kernel besides the refused batch" + + +# --- what the router lets through, and what it does when nft says no ----------- + + +def test_an_address_field_is_an_address_or_an_interface_the_kernel_takes(): + import pytest + from fastapi import HTTPException + + from modules.firewall.models import RuleModel + from routers.firewall import parse_and_check_rule + + def model(**kw) -> RuleModel: + fields = dict(active=True, name="r", proto="tcp", table="filter", src="", dst="", + port_src_from=1, port_dst_from=80, port_src_to=65535, + port_dst_to=80, action="accept", mode="in") + fields.update(kw) + return RuleModel(**fields) + + assert parse_and_check_rule(model(src="eth0")).src == "eth0" + assert parse_and_check_rule(model(src=" br-* ")).src == "br-*" + assert parse_and_check_rule(model(src="10.0.0.1")).src == "10.0.0.1/32" + for bad in ("a-name-far-too-long-for-a-kernel", 'eth0"', "eth 0", "10.0.0.0/33x"): + with pytest.raises(HTTPException) as refused: + parse_and_check_rule(model(src=bad)) + assert refused.value.status_code == 400, bad + + +def test_a_change_nft_refuses_puts_the_database_back(monkeypatch): + """Kept, the refused configuration is what the watcher retries every few seconds and + what the next boot comes up with — failing each time, with no firewall behind it.""" + import asyncio + + import pytest + from fastapi import HTTPException + + import routers.firewall as router + + attempts = [] + + async def reload(): + attempts.append(len(attempts)) + if len(attempts) == 1: + raise Exception("Error: Could not process rule") + + async def refresh(): + pass + + undone = [] + monkeypatch.setattr(router.firewall, "reload", reload) + monkeypatch.setattr(router, "refresh_frontend", refresh) + with pytest.raises(HTTPException) as refused: + asyncio.run(router.apply_changes(lambda: undone.append(True))) + assert refused.value.status_code == 400 + assert undone == [True], "the database was left holding what nft refused" + assert len(attempts) == 2, "the previous firewall was not put back after the undo" diff --git a/tests/unit/test_refused_rebuild.py b/tests/unit/test_refused_rebuild.py new file mode 100644 index 0000000..0e2fefc --- /dev/null +++ b/tests/unit/test_refused_rebuild.py @@ -0,0 +1,142 @@ +"""An edit refused by a rebuild, and a service that could not start at all. + +Both used to leave the same thing behind: a service that was not running, while what the +operator was shown said otherwise. An edit whose rebuild failed was answered "refused" — +which reads as "nothing happened" — after the rebuild had already stopped the service. And +a service that could not come back at boot kept `active` in the database, which is what +the list showed, with nothing filtering it. +""" + +import asyncio +from types import SimpleNamespace + +import pytest + +from modules.services import firewall +from modules.services.models import STATUS, TRANSPORT, Service + + +def _link(name: str): + return SimpleNamespace(name=name, filter=SimpleNamespace(name=name, active=True)) + + +class _Queued: + """Stands in for the NFQUEUE layer: a new shape cannot be pushed, only rebuilt, and a + chain holding the filter called `broken` does not start.""" + + def __init__(self, started: list): + self.started = started + + async def start(self, chain): + names = [link.name for link in chain] + self.started.append(names) + if "broken" in names: + raise Exception("the nfqueue binary rejected the filters") + return {} + + async def reload(self, chain): + raise firewall.transports.ChainShapeChanged() + + async def stop(self): + pass + + async def retire(self, keep_filtering, then=None): + if then: + then() + + +class _Quiet: + """The rules and the log: none of them is the question.""" + + def add(self, *args, **kwargs): + pass + + def delete(self, *args, **kwargs): + return set() + + def release_guards(self, *args, **kwargs): + pass + + +class _Db: + def __init__(self): + self.written = [] + + def query(self, sql, *args): + if sql.startswith("UPDATE"): + self.written.append((sql, args)) + return [] + + +@pytest.fixture +def service(monkeypatch): + started = [] + state = {"chain": ["ok"]} + monkeypatch.setattr(firewall, "nft", _Quiet()) + monkeypatch.setattr(firewall, "log_for", lambda service_id: _Quiet()) + monkeypatch.setattr(firewall.transports, "build", lambda srv, **kw: _Queued(started)) + srv = Service(service_id="s", name="n", status=STATUS.STOP, proto="tcp", + transport=TRANSPORT.NFQUEUE) + db = _Db() + manager = firewall.ServiceManager(srv, db) + manager.chain = lambda: [_link(name) for name in state["chain"]] + return manager, state, started, db + + +def test_a_rebuild_the_edit_cannot_survive_starts_the_service_again_as_it_was(service): + manager, state, started, _ = service + + async def run(): + await manager.enable() + state["chain"] = ["ok", "broken"] + with pytest.raises(Exception): + await manager.update_chain(lambda: state.update(chain=["ok"])) + + asyncio.run(run()) + assert manager.active, "an edit that was refused left the service stopped" + assert started == [["ok"], ["ok", "broken"], ["ok"]], started + + +def test_the_undo_is_run_once_whoever_asks_for_it(): + """The manager runs it before starting the service again; the router's own `except`, + which has always run it, now arrives second.""" + from routers.services import _once + + ran = [] + undo = _once(lambda: ran.append(1)) + undo() + undo() + assert ran == [1] + _once(None)() # nothing to undo is not an error + + +def test_stopping_a_service_that_could_not_start_is_remembered(service): + """It is not running, so stopping it changed nothing in the kernel — but it is still + a decision, and kept as `active` the next boot would try to start it again.""" + manager, state, _, db = service + manager.srv.status = STATUS.ACTIVE # what the database says: meant to run + state["chain"] = ["broken"] + + async def run(): + with pytest.raises(Exception): + await manager.enable() + assert not manager.active + await manager.disable() + + asyncio.run(run()) + assert manager.srv.status == STATUS.STOP + assert any(args[0] == STATUS.STOP for _, args in db.written), db.written + + +def test_the_list_says_whether_a_service_is_running_not_whether_it_is_meant_to(monkeypatch): + import routers.services as router + + class Manager: + active = False + + monkeypatch.setitem(router.firewall.services, "x", Manager()) + assert router._running_status({"service_id": "x", "status": STATUS.ACTIVE}) == STATUS.STOP + Manager.active = True + assert router._running_status({"service_id": "x", "status": STATUS.STOP}) == STATUS.ACTIVE + assert router._running_status({"service_id": "nobody", "status": STATUS.ACTIVE}) \ + == STATUS.ACTIVE, "a service the firewall does not know keeps what it stored" diff --git a/tests/unit/test_service_validation.py b/tests/unit/test_service_validation.py index 8ee7908..690691c 100644 --- a/tests/unit/test_service_validation.py +++ b/tests/unit/test_service_validation.py @@ -309,3 +309,16 @@ def test_the_index_name_is_still_understood_if_it_is_ever_reported(): assert "proxy endpoint" in _address_taken(sqlite3.IntegrityError( "UNIQUE constraint failed: index 'unique_hijack_target'")) + + +def test_an_endpoint_is_one_address_in_the_family_of_what_it_stands_in_for(): + """Stored as typed, a mistyped endpoint was a service that would not start, with the + reason in its log instead of in the answer to the request that caused it.""" + assert "one IP address" in refusal(_hijack_ip, TRANSPORT.EXTERNAL, "10.0.0.1/32", + form(proxy_ip="not-an-ip", proxy_port=8080)) + assert "one IP address" in refusal(_hijack_ip, TRANSPORT.EXTERNAL, "10.0.0.1/32", + form(proxy_ip="10.0.0.0/24", proxy_port=8080)) + assert "same family" in refusal(_hijack_ip, TRANSPORT.EXTERNAL, "10.0.0.1/32", + form(proxy_ip="::1", proxy_port=8080)) + assert _hijack_ip(TRANSPORT.EXTERNAL, "10.0.0.1/32", + form(proxy_ip=" 127.0.0.2 ", proxy_port=8080)) == "127.0.0.2"