From 9d2a4dc4bfc1026de5bb5d8381cae6edcfc92b62 Mon Sep 17 00:00:00 2001 From: Domingo Dirutigliano Date: Fri, 11 Sep 2026 18:53:41 +0200 Subject: [PATCH] additional improves on UDP --- CLAUDE.md | 7 +- README.md | 4 +- backend/modules/services/firewall.py | 30 ++ backend/modules/services/models.py | 8 +- backend/modules/services/nftables.py | 101 ++++-- backend/modules/services/transports.py | 121 +++++-- backend/proxysrc/Cargo.lock | 1 + backend/proxysrc/Cargo.toml | 1 + backend/proxysrc/src/control.rs | 34 +- backend/proxysrc/src/hyperscan.rs | 36 +- backend/proxysrc/src/main.rs | 67 ++-- backend/proxysrc/src/proxy.rs | 2 +- backend/proxysrc/src/rules.rs | 9 +- backend/proxysrc/src/transparent.rs | 57 ++- backend/proxysrc/src/udp.rs | 143 ++++++-- backend/routers/services.py | 40 ++- backend/utils/__init__.py | 31 +- docs/services.md | 108 +++--- fgex-lib/firegex/pyfilters/internals/data.py | 12 +- frontend/src/components/PortAndInterface.tsx | 20 +- .../components/Services/AddEditService.tsx | 335 +++++++++--------- .../src/components/Services/AddressList.tsx | 17 +- .../src/components/Services/LayerChoice.tsx | 14 +- .../src/components/Services/ServiceRow.tsx | 5 +- tests/services_test.py | 31 ++ tests/test_interface_address.py | 277 +++++++++++++++ 26 files changed, 1107 insertions(+), 404 deletions(-) create mode 100644 tests/test_interface_address.py diff --git a/CLAUDE.md b/CLAUDE.md index 4764098..57b1099 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -83,7 +83,7 @@ The central idea, and the thing to preserve when changing anything here: **how t **UDP on the proxy layer (`proxysrc/src/udp.rs`).** The TCP path recovers where a connection was headed with `SO_ORIGINAL_DST`, which the kernel implements **for TCP and SCTP only** (`ENOPROTOOPT` otherwise). So UDP is relayed with **one socket per protected address**, each with its upstream fixed — nothing has to be recovered because nothing was lost. The engine prints one `UDP ` line per relay after `PORT`, and `_add_proxy` points each address's rule at its own port. -The cost is that **the client's address is not preserved** there: no transparent dial, so the service sees the engine. That is stated in the form where the layer is chosen, on the service itself (a `NO CLIENT IP` badge, `losesClientAddress()`), and in the docs. Do not quietly "fix" it with TPROXY: that leaves no NAT entry to key the return path off, so it cannot reach a service on this host, which is why it was ruled out before UDP came up. +Client address transparency is preserved via `transparent::connect_as_udp` using `IP_TRANSPARENT` with mark `0x1339` and policy routing `table 1339`. New UDP addresses can be added to running services dynamically via `ADD_UDP` over the control channel without restarting the proxy. 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. @@ -93,11 +93,12 @@ What UDP keeps: per-flow filter state keyed by client address and released after `service_addresses` holds `(ip, port, proto)` rows; the service itself holds the name, transport, protocol and TLS material. One daemon routinely answers on more than one address — a v4 and a v6 one, a public and an internal one — and all of them deserve the same chain. Two services with hand-copied chains is how one of them silently stops being protected. 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). The one exception is a proxy service gaining its *first* IPv6 address, because the listener must be reopened in a family that can accept one — it restarts and logs that it did. +- **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 proxy listener is dual-stack when any address is 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. +- **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`. Nftables matches use `meta iifname ` for inbound/prerouting and `meta oifname ` for outbound/postrouting/route. On `proxy` (TCP/TLS), redirect rules match `iifname` and `route_chain` marks return packets with `0x1339` on `oifname`, skipping `nat_output_chain` to avoid hijacking unrelated outbound host traffic. 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. **An ordered chain works on every layer, and gets there differently.** The proxy walks a list inside one process. NFQUEUE cannot — its binaries fuse transport and filter — so there a chain is a chain of *processes*: one binary per filter, each with its own queue and its own pair of base chains, at `-307 + position` (prerouting) and `107 + position` (postrouting). A packet a filter accepts carries on to the next base chain in the same hook, so **filter order is base-chain priority order**. This was established by experiment, not assumed: two `cppregex` processes at adjacent priorities, where the second refuses traffic the first passed. The cost is a process and a reassembly pass per filter, which is why `MAX_CHAIN_POSITIONS` exists at all. diff --git a/README.md b/README.md index bf70deb..ea914f5 100644 --- a/README.md +++ b/README.md @@ -85,13 +85,13 @@ This is especially useful if you lost the current password: it's applied directl ## Functionalities - **[Services](docs/services.md)**: the main module. A service is one protected endpoint described in two independent parts — a **network layer** that says how its traffic is intercepted, and an ordered chain of **filters** that says what happens to it. - - A service protects a **list of addresses**, mixing IPv4 and IPv6 freely: one daemon routinely answers on more than one, and all of them get the same chain. Adding an address to a running service drops no connections. + - A service protects a **list of addresses or network interfaces**, mixing IPv4, IPv6 and network interfaces (`eth0`, `wg0`, etc.) freely: intercept traffic on specific IPs or directly across an entire interface. Adding an address or interface to a running service drops no connections. - Network layer: **NFQUEUE** (packets lifted to userspace with [nfqueue](https://netfilter.org/projects/libnetfilter_queue/) and [nftables](https://netfilter.org/projects/nftables/), nothing terminated, the kernel keeps forwarding if a filter dies) or **proxy** (the connection is terminated and reopened, which buys exact rewriting, an ordered chain of filters, and real backpressure — while still dialling your service from the client's own address, so it never stops seeing who is talking to it). - Filters: **regex**, matched by [hyperscan](https://github.com/VectorCamp/vectorscan) — the same engine on either layer, so a pattern means the same thing wherever it runs — or **Python**, your own code written against the [`firegex`](https://pypi.org/project/firegex/) library, with built-in models for HTTP and TCP streams. You never declare which protocol a Python filter speaks: asking for an `HttpRequest` is what makes it an HTTP filter, and a filter reads the addresses and ports as metadata while the payload is the only thing it can change. - A regex can **block** the connection or **rewrite** what it matched (redact a leaked flag, neuter a header). Rewriting is exact on the proxy layer, which owns both connections; the NFQUEUE layer refuses it rather than desynchronising the stream. - **TLS** is a switch on the service, not a separate thing to create: nginx terminates the connection, the filters inspect the plaintext, and it is re-encrypted towards your service. - Or **your own proxy** as the network layer: firegex rewrites the destination on the way in and changes it back on the way out, so a proxy you wrote yourself sits in the path invisibly. The escape hatch for a protocol nothing built in understands. - - **TCP or UDP**, on every layer — but [UDP through the proxy does not preserve the client's address](docs/services.md#udp-on-the-proxy-layer-and-what-it-gives-up), because the kernel option that recovers a datagram's original destination is TCP-only. You get exact rewriting and per-flow filter state; your service sees firegex instead of the client. NFQUEUE filters UDP with the real packets untouched. Firegex says which trade you are making, where you make it. + - **TCP or UDP**, on every layer — with full source IP transparency on both TCP and UDP. On the proxy layer, UDP datagrams are relayed with dedicated sockets using transparent IP spoofing (`IP_TRANSPARENT`), with exact rewriting, per-flow filter state, and zero-downtime dynamic address additions. NFQUEUE filters packets in place with kernel-level fail-open. - **The two inspecting layers are opposite trades, [compared side by side in the interface itself](docs/services.md#network-layers) while you choose**: the proxy owns both halves of a connection, so it rewrites exactly and lets the kernel reassemble, at the cost of weight and of rebuilding fail-open by hand; NFQUEUE is per packet, costs almost nothing and keeps forwarding by kernel guarantee if a filter dies, at the cost of userspace reassembly and no pattern rewriting. - **Charts of what each filter has refused** — per filter, per pattern and per `@pyfilter` function, each with **its share of the blocking**. Pick the window (last 15 minutes, hour, 6h, 24h, everything kept, or two exact instants) and the shape (bars, lines, stacked area, or share); **everything on the page counts the same window**, so the chart and the table beside it can never tell different stories. Alongside it, how much traffic arrived, in the unit each layer can honestly report. - **The Python editor knows the library**: completion and hover for every model and its members — marking which of them you may write to — and the file is **checked as you type by the process that will run it**, so a mistake is flagged on its own line with the reason instead of surfacing later as "worker exited". The hints are introspected from the installed library, so they cannot drift from it. diff --git a/backend/modules/services/firewall.py b/backend/modules/services/firewall.py index e2b881b..72a6098 100644 --- a/backend/modules/services/firewall.py +++ b/backend/modules/services/firewall.py @@ -377,10 +377,13 @@ class ServiceManager: self.reload_addresses() if not self.active: return + is_dual = getattr(self.transport, "is_dual_stack", False) if ( self.srv.transport == transports.TRANSPORT.PROXY + and str(self.srv.proto) != "udp" and self.srv.has_ipv6 and not was_dual + and not is_dual ): self.log.add( LEVEL.INFO, @@ -391,6 +394,33 @@ class ServiceManager: added = [addr for addr in self.srv.addresses if addr.id == address_id] if not added: return + if self.srv.transport == transports.TRANSPORT.PROXY: + from modules.services.nftables import one_address, ip_family + from utils import is_ip_parse, get_interface_ips + for addr in added: + l4 = str(addr.proto or self.srv.proto) + if l4 == "udp": + is_iface = not is_ip_parse(addr.ip_int) + if is_iface: + ips = get_interface_ips(addr.ip_int) + if not ips: + raise transports.UnsupportedChain( + f"interface '{addr.ip_int}' has no IP assigned for UDP proxy relay" + ) + target_ip = one_address(ips[0]) + else: + target_ip = one_address(addr.ip_int) + target_port = addr.port + family = ip_family(target_ip) + key = ( + f"[{target_ip}]:{target_port}" + if family == "ip6" else f"{target_ip}:{target_port}" + ) + if "udp_ports" not in self._steer: + self._steer["udp_ports"] = {} + if key not in self._steer["udp_ports"]: + port = await self.transport.add_udp_target(target_ip, target_port) + self._steer["udp_ports"][key] = port async with self.lock: nft.add(self.srv, added, **self._steer) self.log.add( diff --git a/backend/modules/services/models.py b/backend/modules/services/models.py index f86b649..c01e017 100644 --- a/backend/modules/services/models.py +++ b/backend/modules/services/models.py @@ -14,6 +14,8 @@ that can host it, and the service does not have to be recreated to change either import base64 import uuid +from utils import is_ip_parse + class TRANSPORT: """How the traffic is intercepted.""" @@ -183,9 +185,13 @@ class Address: "proxy_port": self.proxy_port, } + @property + def is_interface(self) -> bool: + return not is_ip_parse(self.ip_int) + @property def is_ipv6(self) -> bool: - return ":" in str(self.ip_int) + return ":" in str(self.ip_int) if not self.is_interface else False def __repr__(self): return f"
" diff --git a/backend/modules/services/nftables.py b/backend/modules/services/nftables.py index be1d91c..bed071d 100644 --- a/backend/modules/services/nftables.py +++ b/backend/modules/services/nftables.py @@ -25,7 +25,15 @@ NAT entry to key off and so could never reach a local service at all. """ from modules.services.models import TRANSPORT, Address, Service -from utils import NFTableManager, addr_parse, ip_family, ip_parse, nftables_int_to_json +from utils import ( + NFTableManager, + addr_parse, + get_interface_ips, + ip_family, + ip_parse, + is_ip_parse, + nftables_int_to_json, +) # Direction marks the NFQUEUE binaries read back off a packet. QUEUE_MARK_INPUT = 0x1337 @@ -109,16 +117,22 @@ class InstalledRule: if target is None: return False target_ip, target_port = target - if self.port == int(target_port) and ip_parse(self.ip_int) == ip_parse(target_ip): - return True + if self.port != int(target_port): + return False + if is_ip_parse(self.ip_int) and is_ip_parse(target_ip): + if ip_parse(self.ip_int) == ip_parse(target_ip): + return True + elif not is_ip_parse(self.ip_int) and not is_ip_parse(target_ip): + if self.ip_int == target_ip: + return True # The return leg of an external hand-off matches the operator's proxy, not the # service. Recognising only the inbound rule would leave the outbound one behind # every time a service was stopped, and it would keep rewriting. - if srv.transport == TRANSPORT.EXTERNAL and addr.proxy_port: + if srv.transport == TRANSPORT.EXTERNAL and addr.proxy_port and is_ip_parse(target_ip): proxy_ip = one_address(addr.proxy_ip or "") or ( "::1" if ip_family(target_ip) == "ip6" else "127.0.0.1" ) - return self.port == int(addr.proxy_port) and ip_parse(self.ip_int) == ip_parse(proxy_ip) + return is_ip_parse(self.ip_int) and ip_parse(self.ip_int) == ip_parse(proxy_ip) return False def matches_any(self, srv: Service, addresses: list[Address]) -> bool: @@ -314,6 +328,24 @@ class FiregexTables(NFTableManager): COUNTER = {"counter": {"packets": 0, "bytes": 0}} def _match(self, ip: str, port: int, family: str, l4: str, addr: str, field: str) -> list: + if not is_ip_parse(ip): + iface_key = "iifname" if addr == "daddr" else "oifname" + return [ + { + "match": { + "op": "==", + "left": {"meta": {"key": iface_key}}, + "right": ip, + } + }, + { + "match": { + "op": "==", + "left": {"payload": {"protocol": l4, "field": field}}, + "right": int(port), + } + }, + ] return [ { "match": { @@ -368,7 +400,8 @@ class FiregexTables(NFTableManager): if target is None: continue target_ip, target_port = target - family = ip_family(target_ip) + is_iface = not is_ip_parse(target_ip) + family = "ip" if is_iface else ip_family(target_ip) l4 = str(addr.proto) if srv.transport == TRANSPORT.NFQUEUE: @@ -380,10 +413,24 @@ class FiregexTables(NFTableManager): # at that address's port rather than at the one shared TCP listener. port = proxy_port if l4 == "udp": - key = ( - f"[{one_address(target_ip)}]:{target_port}" - if family == "ip6" else f"{one_address(target_ip)}:{target_port}" - ) + if is_iface: + ips = get_interface_ips(target_ip) + if not ips: + raise Exception( + f"the proxy transport has no UDP relay listening for {target_ip}: " + f"interface '{target_ip}' has no IP assigned" + ) + resolved_ip = one_address(ips[0]) + resolved_family = ip_family(resolved_ip) + key = ( + f"[{resolved_ip}]:{target_port}" + if resolved_family == "ip6" else f"{resolved_ip}:{target_port}" + ) + else: + key = ( + f"[{one_address(target_ip)}]:{target_port}" + if family == "ip6" else f"{one_address(target_ip)}:{target_port}" + ) port = (udp_ports or {}).get(key) if port is None: raise Exception( @@ -445,6 +492,8 @@ class FiregexTables(NFTableManager): port back, so two addresses sharing one proxy endpoint could not be told apart on the way out. Every address gets its own. """ + if not is_ip_parse(ip): + raise Exception("the external transport requires a concrete IP address, not an interface") if not addr.proxy_port: raise Exception("the external transport needs the port your proxy listens on") # One host, not a network: a hand-off points at the single address the @@ -484,24 +533,23 @@ class FiregexTables(NFTableManager): # what this host generates, which is the only way a client on this host reaches # a service on it. Conntrack NATs a connection once, so a rule in each is not a # double translation. + is_iface = not is_ip_parse(ip) redirect = ( self._not_ours() + self._match(ip, port, family, l4, "daddr", "dport") + [self.COUNTER, {"redirect": {"port": int(proxy_port)}}] ) - self.cmd( + cmds = [ self._rule(self.nat_chain, redirect), - self._rule(self.nat_output_chain, redirect), - # Source preservation is not optional. A protected service that suddenly - # sees one address for the whole internet is a silent regression against - # the NFQUEUE transport, and the operator would find out from their own - # rate limiter rather than from us. self._rule( self.route_chain, self._match(ip, port, family, l4, "saddr", "sport") + [{"mangle": {"key": {"meta": {"key": "mark"}}, "value": PROXY_MARK}}], ), - ) + ] + if not is_iface: + cmds.append(self._rule(self.nat_output_chain, redirect)) + self.cmd(*cmds) # --- reading back --------------------------------------------------------- @@ -525,14 +573,19 @@ class FiregexTables(NFTableManager): # per-service rule starts with the address. The two rules init() # installs match on `socket` or on `ct`, and fall out here. if "meta" in expr[0].get("match", {}).get("left", {}): - expr = expr[1:] - if "payload" not in expr[0].get("match", {}).get("left", {}): - continue - right = expr[0]["match"]["right"] - if isinstance(right, str): - ip_int = str(ip_parse(right)) + if expr[0].get("match", {}).get("left", {}).get("meta", {}).get("key") == "mark": + expr = expr[1:] + match_left = expr[0].get("match", {}).get("left", {}) + if "payload" in match_left: + right = expr[0]["match"]["right"] + if isinstance(right, str): + ip_int = str(ip_parse(right)) + else: + ip_int = f'{right["prefix"]["addr"]}/{right["prefix"]["len"]}' + elif "meta" in match_left and match_left["meta"].get("key") in ("iifname", "oifname"): + ip_int = str(expr[0]["match"]["right"]) else: - ip_int = f'{right["prefix"]["addr"]}/{right["prefix"]["len"]}' + continue counter = next( (e["counter"] for e in expr if isinstance(e, dict) and "counter" in e), {}, diff --git a/backend/modules/services/transports.py b/backend/modules/services/transports.py index d027565..33d9e13 100644 --- a/backend/modules/services/transports.py +++ b/backend/modules/services/transports.py @@ -165,6 +165,9 @@ class Transport: async def reload(self, chain: list[ChainLink]) -> None: raise NotImplementedError + async def add_udp_target(self, ip: str, port: int) -> int: + raise NotImplementedError + async def stop(self) -> None: raise NotImplementedError @@ -580,6 +583,16 @@ class _EmptyFilter: self.active = True +def supports_ipv6() -> bool: + try: + import socket + with socket.socket(socket.AF_INET6, socket.SOCK_STREAM) as s: + s.bind(("::", 0)) + return True + except Exception: + return False + + class ProxyTransport(Transport): """The connection is terminated and reopened towards the service. @@ -599,12 +612,24 @@ class ProxyTransport(Transport): @classmethod def check(cls, srv, chain: list[ChainLink]) -> None: super().check(srv, chain) + if str(srv.proto) == L4.UDP: + from utils import get_interface_ips, is_ip_parse + for addr in srv.addresses: + if not is_ip_parse(addr.ip_int): + ips = get_interface_ips(addr.ip_int) + if not ips: + raise UnsupportedChain( + f"interface '{addr.ip_int}' has no IP assigned for UDP proxy relay. " + f"Use an IP address or NFQUEUE transport." + ) def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self._reader_task: asyncio.Task | None = None self._ack: asyncio.Future | None = None + self._cmd_lock = asyncio.Lock() self.port: int | None = None + self.is_dual_stack: bool = False #: One relay port per protected address, for UDP. TCP needs none of this: a #: single listener fronts every address and recovers where each connection was #: headed from conntrack, which UDP cannot do — see `udp.rs`. @@ -615,11 +640,23 @@ class ProxyTransport(Transport): if str(self.srv.proto) != L4.UDP: return [] from modules.services.nftables import resolve_target + from utils import get_interface_ips, is_ip_parse out = [] for addr in self.srv.addresses: target = resolve_target(self.srv, addr) if target: - out.append(target) + target_ip, target_port = target + if not is_ip_parse(target_ip): + ips = get_interface_ips(target_ip) + if ips: + out.append((ips[0], target_port)) + else: + raise UnsupportedChain( + f"interface '{target_ip}' has no IP assigned for UDP proxy relay. " + f"Use an IP address or NFQUEUE transport." + ) + else: + out.append(target) return out def _tls_env(self) -> dict: @@ -660,12 +697,11 @@ class ProxyTransport(Transport): async def start(self, chain: list[ChainLink]) -> dict: self.check(self.srv, chain) # The listener has to be able to accept every family the service answers in. A - # redirect keeps the address family, so an IPv6 connection sent at an IPv4-only - # listener is simply refused — and refused connections to a service that used to - # work is the worst way to find that out. One v6 address is enough to make the - # listener v6: a dual-stack socket takes the v4 ones as mapped addresses, which - # the engine unmaps before it does anything with them. - listen = "[::]:0" if self.srv.has_ipv6 else "0.0.0.0:0" + # dual-stack listener ([::]:0) is used whenever IPv6 is supported on the host, + # so it accepts both IPv4 (as mapped addresses) and IPv6 connections from the start. + # This allows adding new IPv6 addresses to a running service without any restart. + self.is_dual_stack = self.srv.has_ipv6 or supports_ipv6() + listen = "[::]:0" if self.is_dual_stack else "0.0.0.0:0" self.process = await asyncio.create_subprocess_exec( PROXY_ENGINE, stdout=asyncio.subprocess.PIPE, @@ -702,6 +738,7 @@ class ProxyTransport(Transport): **({"FGEX_PROXY_OVER_LIMIT_FORWARD": "1"} if self.srv.over_limit_forwards else {}), "FGEX_PROXY_FIRST_BYTE_TIMEOUT": str(self.srv.first_byte_timeout), + "FGEX_PROXY_FILTER_TIMEOUT_MS": os.getenv("FGEX_PROXY_FILTER_TIMEOUT_MS", "2000"), **self._tls_env(), }, ), @@ -795,20 +832,52 @@ class ProxyTransport(Transport): async def reload(self, chain: list[ChainLink]) -> None: if not self.process or self.process.returncode is not None: return - self._ack = asyncio.get_running_loop().create_future() - self.process.stdin.write((self._payload(chain) + "\n").encode()) - await self.process.stdin.drain() - try: - ok, detail = await asyncio.wait_for(self._ack, timeout=ACK_TIMEOUT) - except asyncio.TimeoutError: - raise Exception("the proxy engine did not acknowledge the chain") - finally: - self._ack = None - if not ok: - # Deliberately not fatal to the datapath: the engine keeps enforcing the - # chain it already had, so a rejected edit costs the operator an error - # message rather than their protection. - raise Exception(f"the proxy engine rejected the chain: {detail}") + async with self._cmd_lock: + self._ack = asyncio.get_running_loop().create_future() + self.process.stdin.write((self._payload(chain) + "\n").encode()) + await self.process.stdin.drain() + try: + ok, detail = await asyncio.wait_for(self._ack, timeout=ACK_TIMEOUT) + except asyncio.TimeoutError: + raise Exception("the proxy engine did not acknowledge the chain") + finally: + self._ack = None + if not ok: + # Deliberately not fatal to the datapath: the engine keeps enforcing the + # chain it already had, so a rejected edit costs the operator an error + # message rather than their protection. + raise Exception(f"the proxy engine rejected the chain: {detail}") + + async def add_udp_target(self, ip: str, port: int) -> int: + """Bind a dedicated UDP relay for a new address without restarting the proxy engine.""" + from modules.services.nftables import one_address + clean_ip = one_address(ip) + target = f"[{clean_ip}]:{port}" if ":" in str(clean_ip) else f"{clean_ip}:{port}" + if target in self.udp_ports: + return self.udp_ports[target] + if not self.process or self.process.returncode is not None: + raise Exception("the proxy engine is not running") + + async with self._cmd_lock: + if target in self.udp_ports: + return self.udp_ports[target] + self._ack = asyncio.get_running_loop().create_future() + self.process.stdin.write(f"ADD_UDP {target}\n".encode()) + await self.process.stdin.drain() + try: + ok, detail = await asyncio.wait_for(self._ack, timeout=ACK_TIMEOUT) + except asyncio.TimeoutError: + raise Exception(f"the proxy engine did not acknowledge ADD_UDP for {target}") + finally: + self._ack = None + if not ok: + raise Exception(f"the proxy engine rejected ADD_UDP {target}: {detail}") + relay_port = self.udp_ports.get(target) + if relay_port is None: + raise Exception( + f"the proxy engine acknowledged ADD_UDP but did not report port for {target}" + ) + return relay_port async def _read_events(self): try: @@ -820,6 +889,11 @@ class ProxyTransport(Transport): rest = line[4:].strip() if self._ack and not self._ack.done(): self._ack.set_result((rest.upper().startswith("OK"), rest)) + elif line.startswith("UDP "): + parts = line.split() + if len(parts) == 3: + _, upstream, port_str = parts + self.udp_ports[upstream] = int(port_str) elif line.startswith("BLOCKED "): self._emit_block(line.split()[1]) elif line.startswith("STATS "): @@ -873,6 +947,11 @@ class ExternalTransport(Transport): @classmethod def check(cls, srv, chain: list[ChainLink]) -> None: super().check(srv, chain) + if any(addr.is_interface for addr in srv.addresses): + raise UnsupportedChain( + "the external transport hands traffic to an external proxy and rewrites the " + "source IP on return, which requires a concrete IP address rather than an interface." + ) missing = [addr for addr in srv.addresses if not addr.proxy_port] if missing: raise UnsupportedChain( diff --git a/backend/proxysrc/Cargo.lock b/backend/proxysrc/Cargo.lock index 40f0e67..5b0af0e 100644 --- a/backend/proxysrc/Cargo.lock +++ b/backend/proxysrc/Cargo.lock @@ -46,6 +46,7 @@ dependencies = [ "rustls-pemfile", "serde", "serde_json", + "socket2", "tokio", "tokio-rustls", ] diff --git a/backend/proxysrc/Cargo.toml b/backend/proxysrc/Cargo.toml index d9a653c..99e2945 100644 --- a/backend/proxysrc/Cargo.toml +++ b/backend/proxysrc/Cargo.toml @@ -9,6 +9,7 @@ license = "GPL-3.0-or-later" tokio = { version = "1", features = ["rt-multi-thread", "net", "io-util", "io-std", "macros", "time", "sync"] } # Raw setsockopt: IP_TRANSPARENT and SO_ORIGINAL_DST have no safe wrapper. libc = "0.2" +socket2 = "0.6" # Filter rules arrive as one JSON line on stdin serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/backend/proxysrc/src/control.rs b/backend/proxysrc/src/control.rs index 79fa334..6b30ca7 100644 --- a/backend/proxysrc/src/control.rs +++ b/backend/proxysrc/src/control.rs @@ -3,12 +3,14 @@ //! Same shape the NFQUEUE engines use, so the backend side is the code it already //! has: write the whole ruleset, wait for `ACK OK` or `ACK FAIL `. +use std::net::SocketAddr; use std::time::Duration; use tokio::io::{AsyncBufReadExt, BufReader}; use crate::filter::{ChainHandle, FilterChain}; use crate::rules::parse_ruleset; +use crate::udp::UdpManager; fn ack(ok: bool, detail: &str) { if ok { @@ -20,14 +22,14 @@ fn ack(ok: bool, detail: &str) { let _ = std::io::stdout().flush(); } -/// Apply rulesets until stdin closes. +/// Apply rulesets and handle control commands until stdin closes. /// /// A rejected update leaves the running chain in place: the engine keeps enforcing /// what it already had rather than falling back to enforcing nothing, which is the /// opposite of the fail-open rule and deliberately so. Failing open covers a filter /// that breaks at runtime; it is not a licence to silently drop a ruleset the /// operator asked for. -pub async fn serve_stdin(chain: ChainHandle, deadline: Duration) { +pub async fn serve_stdin(chain: ChainHandle, deadline: Duration, udp: UdpManager) { let mut lines = BufReader::new(tokio::io::stdin()).lines(); loop { let line = match lines.next_line().await { @@ -44,10 +46,34 @@ pub async fn serve_stdin(chain: ChainHandle, deadline: Duration) { return; } }; - if line.trim().is_empty() { + let trimmed = line.trim(); + if trimmed.is_empty() { continue; } - match parse_ruleset(&line).and_then(|filters| { + + if let Some(rest) = trimmed.strip_prefix("ADD_UDP ") { + let target_str = rest.trim(); + match target_str.parse::() { + Ok(upstream) => match udp.add_relay(upstream).await { + Ok(port) => { + println!("UDP {upstream} {port}"); + ack(true, ""); + } + Err(e) => { + eprintln!("[warn] [control] failed to add UDP relay for {upstream}: {e}"); + ack(false, &e.to_string()); + } + }, + Err(e) => { + let err = format!("invalid target address {target_str}: {e}"); + eprintln!("[warn] [control] {err}"); + ack(false, &err); + } + } + continue; + } + + match parse_ruleset(trimmed).and_then(|filters| { // Everything that has to be started is started here, while the old chain // is still the one enforcing. for filter in &filters { diff --git a/backend/proxysrc/src/hyperscan.rs b/backend/proxysrc/src/hyperscan.rs index 6e883f3..5b4b478 100644 --- a/backend/proxysrc/src/hyperscan.rs +++ b/backend/proxysrc/src/hyperscan.rs @@ -41,6 +41,7 @@ const HS_SUCCESS: c_int = 0; const HS_SCAN_TERMINATED: c_int = -3; const HS_FLAG_CASELESS: c_uint = 1; +#[allow(dead_code)] const HS_FLAG_SINGLEMATCH: c_uint = 8; const HS_FLAG_ALLOWEMPTY: c_uint = 16; const HS_FLAG_SOM_LEFTMOST: c_uint = 256; @@ -84,6 +85,13 @@ extern "C" { on_event: Option, ctxt: *mut c_void, ) -> c_int; + fn hs_reset_stream( + id: *mut hs_stream_t, + flags: c_uint, + scratch: *mut hs_scratch_t, + on_event: Option, + ctxt: *mut c_void, + ) -> c_int; fn hs_scan( db: *const hs_database_t, data: *const c_char, @@ -171,7 +179,7 @@ impl Database { let ptrs: Vec<*const c_char> = owned.iter().map(|s| s.as_ptr()).collect(); let base = match mode { - Mode::Stream => HS_FLAG_SINGLEMATCH | HS_FLAG_ALLOWEMPTY, + Mode::Stream => HS_FLAG_ALLOWEMPTY, // The debugger wants every match and where it started, so neither // SINGLEMATCH (one report per pattern) nor the default no-start-offset. Mode::Block => HS_FLAG_SOM_LEFTMOST | HS_FLAG_ALLOWEMPTY, @@ -344,6 +352,32 @@ impl StreamScanner { Ok(res.hit) } + /// Reset the stream to its initial state, clearing any previous match or terminated status. + pub fn reset(&mut self) -> Result<(), String> { + if self.stream.is_null() { + return Ok(()); + } + let rc = unsafe { + hs_reset_stream( + self.stream, + 0, + self.scratch.scratch, + None, + ptr::null_mut(), + ) + }; + if rc != HS_SUCCESS { + // If reset failed, close and reopen the stream cleanly. + unsafe { hs_close_stream(self.stream, self.scratch.scratch, None, ptr::null_mut()) }; + self.stream = ptr::null_mut(); + let reopen_rc = unsafe { hs_open_stream(self.db.db, 0, &mut self.stream) }; + if reopen_rc != HS_SUCCESS { + return Err(format!("hyperscan reset stream failed ({rc}/{reopen_rc})")); + } + } + Ok(()) + } + pub fn pattern_count(&self) -> usize { self.db.count } diff --git a/backend/proxysrc/src/main.rs b/backend/proxysrc/src/main.rs index 27049ba..e77bf9f 100644 --- a/backend/proxysrc/src/main.rs +++ b/backend/proxysrc/src/main.rs @@ -163,7 +163,7 @@ async fn run() { let deadline = std::env::var("FGEX_PROXY_FILTER_TIMEOUT_MS") .ok() .and_then(|v| v.parse::().ok()) - .unwrap_or(200); + .unwrap_or(2000); let connect_timeout = std::env::var("FGEX_PROXY_CONNECT_TIMEOUT_MS") .ok() .and_then(|v| v.parse::().ok()) @@ -251,62 +251,47 @@ async fn run() { (spoof_source={spoof_source})" ); - // Rulesets arrive on stdin from here on. It runs alongside the datapath rather - // than gating it: traffic must flow before, during and after a rule change. - let deadline_dur = Duration::from_millis(deadline); - let control_chain = chain_handle.clone(); - tokio::spawn(async move { serve_stdin(control_chain, deadline_dur).await }); + // UDP manager: binds and tracks UDP relays, shared between startup configuration + // and dynamic commands arriving on stdin. + let udp_manager = fgex_proxy::udp::UdpManager::new( + chain_handle.clone(), + cfg_self_mark, + spoof_source, + max_connections, + over_limit_forwards, + Arc::clone(&counters), + ); - // How many connections have been through, reported on a timer rather than one - // line per connection: the backend wants a denominator, not a firehose. Counted - // where the connection is accepted and where a filter refuses it, so both numbers - // are in the same unit and their ratio means something — unlike a share computed - // against a packet count, which is what the kernel can offer and a block is not. // UDP, when the backend asks for it: a comma-separated list of protected // addresses, one relay each. One socket per address rather than one for all of // them, because `SO_ORIGINAL_DST` — which is how the TCP side learns where a // connection was headed — is TCP and SCTP only. With the upstream fixed per socket // there is nothing to recover. - // - // The client's address is *not* preserved on this path, and that is the trade the - // operator is shown before choosing it. if let Ok(spec) = std::env::var("FGEX_PROXY_UDP") { for target in spec.split(',').filter(|s| !s.trim().is_empty()) { let upstream = parse_addr("FGEX_PROXY_UDP", target.trim()); - let bind: SocketAddr = if upstream.is_ipv6() { - "[::]:0".parse().unwrap() - } else { - "0.0.0.0:0".parse().unwrap() - }; - let relay = match fgex_proxy::udp::UdpRelay::bind( - bind, - upstream, - chain_handle.clone(), - cfg_self_mark, - max_connections, - over_limit_forwards, - Arc::clone(&counters), - ) - .await - { - Ok(relay) => relay, + match udp_manager.add_relay(upstream).await { + Ok(port) => { + println!("UDP {upstream} {port}"); + use std::io::Write; + let _ = std::io::stdout().flush(); + } Err(e) => { eprintln!("[fatal] [main] cannot bind a UDP relay for {upstream}: {e}"); exit(1); } - }; - // One line per relay, so the backend can point each address's rule at the - // port that actually fronts it. - println!("UDP {} {}", upstream, relay.local_addr().unwrap().port()); - let _ = std::io::stdout().flush(); - tokio::spawn(async move { - if let Err(e) = relay.serve().await { - eprintln!("[fatal] [udp] relay for {upstream} died: {e}"); - } - }); + } } } + // Rulesets and control commands (e.g. ADD_UDP) arrive on stdin from here on. + // It runs alongside the datapath rather than gating it: traffic must flow + // before, during and after a rule change or relay addition. + let deadline_dur = Duration::from_millis(deadline); + let control_chain = chain_handle.clone(); + let control_udp = udp_manager.clone(); + tokio::spawn(async move { serve_stdin(control_chain, deadline_dur, control_udp).await }); + // Both counters live on the proxy, not on the chain: a chain is replaced wholesale // every time a ruleset is pushed, and counters that reset on a rule edit would make // the ratio jump for a reason nobody could see. diff --git a/backend/proxysrc/src/proxy.rs b/backend/proxysrc/src/proxy.rs index 6495799..99b1786 100644 --- a/backend/proxysrc/src/proxy.rs +++ b/backend/proxysrc/src/proxy.rs @@ -146,7 +146,7 @@ pub struct ProxyStats { /// whichever the operator chose. Cumulative, so a service that hit the wall once an /// hour ago still says so. pub over_limit: AtomicU64, - warned_spoof: AtomicBool, + pub warned_spoof: AtomicBool, /// Whether the limit has already been announced. Shared by both paths, so a service /// carrying TCP and UDP says it once rather than twice. pub warned_limit: AtomicBool, diff --git a/backend/proxysrc/src/rules.rs b/backend/proxysrc/src/rules.rs index 2417a58..ef28fbe 100644 --- a/backend/proxysrc/src/rules.rs +++ b/backend/proxysrc/src/rules.rs @@ -177,14 +177,17 @@ impl FilterSession for HyperscanSession { .get(index as usize) .cloned() .unwrap_or_else(|| "unknown".to_string()); + // Reset the stream so that subsequent scans on this session (e.g. further UDP datagrams + // or connections) don't fail due to HS_SCAN_TERMINATED. + let _ = self.scanner.reset(); Verdict::Reject(Some(id)) } Ok(None) => Verdict::Accept, // A scan that errors has told us nothing about the chunk, so the chunk - // goes through. The stream state is suspect from here on, which is worth - // saying once rather than on every subsequent chunk. + // goes through. Reset the scanner so future chunks/datagrams can still be scanned. Err(e) => { - eprintln!("[warn] [filter] scan failed ({e}): chunk forwarded unfiltered"); + let _ = self.scanner.reset(); + eprintln!("[warn] [filter] scan failed ({e}): scanner reset, chunk forwarded"); Verdict::Accept } } diff --git a/backend/proxysrc/src/transparent.rs b/backend/proxysrc/src/transparent.rs index f63b3f3..d258ede 100644 --- a/backend/proxysrc/src/transparent.rs +++ b/backend/proxysrc/src/transparent.rs @@ -11,7 +11,8 @@ use std::io; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::os::fd::{AsRawFd, RawFd}; -use tokio::net::{TcpListener, TcpSocket, TcpStream}; +use socket2::{Domain, Protocol, Socket, Type}; +use tokio::net::{TcpListener, TcpSocket, TcpStream, UdpSocket}; // Not in libc: netfilter's own option numbers. const SO_ORIGINAL_DST: libc::c_int = 80; @@ -139,7 +140,9 @@ fn original_dst_from_conntrack(stream: &TcpStream) -> io::Result { pub fn bind_listener(addr: SocketAddr) -> io::Result { let socket = if addr.is_ipv6() { - TcpSocket::new_v6()? + let sock = TcpSocket::new_v6()?; + let _ = setsockopt_int(sock.as_raw_fd(), libc::IPPROTO_IPV6, libc::IPV6_V6ONLY, 0); + sock } else { TcpSocket::new_v4()? }; @@ -193,6 +196,56 @@ pub async fn connect_plain(upstream: SocketAddr, self_mark: Option) -> io:: socket.connect(upstream).await } +/// Dial the service as the client over UDP: same source address, kernel-chosen source port. +pub async fn connect_as_udp( + client: IpAddr, + upstream: SocketAddr, + self_mark: Option, +) -> io::Result { + if client.is_ipv6() != upstream.is_ipv6() { + return Err(io::Error::new( + io::ErrorKind::InvalidInput, + "cannot spoof an IPv4 client towards an IPv6 service, or the reverse", + )); + } + let domain = if upstream.is_ipv6() { + Domain::IPV6 + } else { + Domain::IPV4 + }; + let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?; + socket.set_nonblocking(true)?; + set_transparent(socket.as_raw_fd(), upstream.is_ipv6())?; + if let Some(mark) = self_mark { + set_self_mark(socket.as_raw_fd(), mark)?; + } + socket.set_reuse_address(true)?; + let bind_addr: socket2::SockAddr = SocketAddr::new(client, 0).into(); + socket.bind(&bind_addr)?; + let std_sock: std::net::UdpSocket = socket.into(); + let tokio_sock = UdpSocket::from_std(std_sock)?; + tokio_sock.connect(upstream).await?; + Ok(tokio_sock) +} + +/// Dial UDP without impersonating anyone, but still marked as ours. +pub async fn connect_plain_udp( + upstream: SocketAddr, + self_mark: Option, +) -> io::Result { + let bind: SocketAddr = if upstream.is_ipv6() { + "[::]:0".parse().unwrap() + } else { + "0.0.0.0:0".parse().unwrap() + }; + let socket = UdpSocket::bind(bind).await?; + if let Some(mark) = self_mark { + set_self_mark(socket.as_raw_fd(), mark)?; + } + socket.connect(upstream).await?; + Ok(socket) +} + /// Mark a socket as ours, so the intercept rules can skip it. pub fn set_self_mark(fd: RawFd, mark: u32) -> io::Result<()> { setsockopt_int(fd, libc::SOL_SOCKET, libc::SO_MARK, mark as libc::c_int) diff --git a/backend/proxysrc/src/udp.rs b/backend/proxysrc/src/udp.rs index c31f0d3..449c024 100644 --- a/backend/proxysrc/src/udp.rs +++ b/backend/proxysrc/src/udp.rs @@ -1,17 +1,17 @@ -//! The proxy layer for UDP: a relay, and one thing it deliberately does not do. +//! The proxy layer for UDP: relays with transparent source IP preservation. //! -//! The TCP side of this engine stays invisible in two ways at once. It recovers where +//! The TCP side of this engine stays invisible in two ways at once: it recovers where //! the client was originally headed with `SO_ORIGINAL_DST`, and it dials the service -//! *from the client's own address*, so anything that logs or rate-limits by IP keeps -//! working. Neither survives the move to UDP unchanged: +//! *from the client's own address* via `IP_TRANSPARENT`. //! +//! On UDP: //! * **`SO_ORIGINAL_DST` is TCP and SCTP only.** The kernel answers `ENOPROTOOPT` for a //! UDP socket, so a single listener cannot ask where a datagram was going. This relay -//! sidesteps it by binding **one socket per protected address**, each with the -//! upstream already known — nothing has to be recovered because nothing was lost. -//! * **The client's address is not preserved.** The protected service sees this engine, -//! not the client. That is a real loss, stated everywhere the operator can choose it, -//! and the reason the nfqueue layer remains the transparent way to filter UDP. +//! handles it by binding **one socket per protected address**, each with the +//! upstream already known. +//! * **The client's address is preserved.** The relay dials the upstream service from +//! the client's own address using `IP_TRANSPARENT` and mark-based policy routing, +//! keeping source IP transparency on both TCP and UDP. //! //! What is kept is everything the filter layer cares about: each client flow is a //! connection with its own filter state and its own module globals, both directions are @@ -67,6 +67,7 @@ pub struct UdpRelay { upstream: SocketAddr, chain: ChainHandle, self_mark: Option, + pub spoof_source: bool, /// How many flows may exist at once. `0` means no limit. max_flows: usize, /// Whether a datagram from a new source past the limit is forwarded unfiltered @@ -85,6 +86,7 @@ impl UdpRelay { upstream: SocketAddr, chain: ChainHandle, self_mark: Option, + spoof_source: bool, max_flows: usize, over_limit_forwards: bool, stats: Arc, @@ -95,6 +97,7 @@ impl UdpRelay { upstream, chain, self_mark, + spoof_source, max_flows, over_limit_forwards, stats, @@ -106,22 +109,18 @@ impl UdpRelay { /// This is what "forwarded unfiltered" has to mean for UDP: a flow *is* the state, so /// admitting a datagram without creating one is admitting it without inspection — /// which is exactly what the operator chose when they picked it over dropping. - async fn forward_unfiltered(&self, data: &[u8]) -> io::Result<()> { - let bind: SocketAddr = if self.upstream.is_ipv6() { - "[::]:0".parse().unwrap() + async fn forward_unfiltered(&self, client: SocketAddr, data: &[u8]) -> io::Result<()> { + let socket = if self.spoof_source { + match crate::transparent::connect_as_udp(client.ip(), self.upstream, self.self_mark).await { + Ok(s) => s, + Err(_) => { + crate::transparent::connect_plain_udp(self.upstream, self.self_mark).await? + } + } } else { - "0.0.0.0:0".parse().unwrap() + crate::transparent::connect_plain_udp(self.upstream, self.self_mark).await? }; - let socket = UdpSocket::bind(bind).await?; - if let Some(mark) = self.self_mark { - // Stamped as ours for the same reason a flow's socket is: without it the - // rule that redirects this service's traffic catches our own send. - crate::transparent::set_self_mark( - ::as_raw_fd(&socket), - mark, - )?; - } - socket.send_to(data, self.upstream).await?; + socket.send(data).await?; Ok(()) } @@ -186,7 +185,7 @@ impl UdpRelay { } // Forwarding without a flow means without filter state, which is // what "unfiltered" has to mean here: a flow is the state. - if let Err(e) = self.forward_unfiltered(&buf[..len]).await { + if let Err(e) = self.forward_unfiltered(client, &buf[..len]).await { eprintln!("[warn] [udp] cannot forward past the limit: {e}"); } continue; @@ -222,21 +221,25 @@ impl UdpRelay { /// Start relaying one client's flow: a socket towards the service, and a task /// carrying the answers back. async fn open(&self, client: SocketAddr) -> io::Result { - let bind: SocketAddr = if self.upstream.is_ipv6() { - "[::]:0".parse().unwrap() + let upstream = if self.spoof_source { + match crate::transparent::connect_as_udp(client.ip(), self.upstream, self.self_mark).await { + Ok(s) => s, + Err(e) => { + self.stats.source_spoof_failures.fetch_add(1, Ordering::Relaxed); + if !self.stats.warned_spoof.swap(true, Ordering::Relaxed) { + eprintln!( + "[warn] [udp] cannot reach {} as {}: {e}. \ + Falling back to our own address — the service will not see real client IPs.", + self.upstream, + client.ip() + ); + } + crate::transparent::connect_plain_udp(self.upstream, self.self_mark).await? + } + } } else { - "0.0.0.0:0".parse().unwrap() + crate::transparent::connect_plain_udp(self.upstream, self.self_mark).await? }; - let upstream = UdpSocket::bind(bind).await?; - if let Some(mark) = self.self_mark { - // Stamped as ours, or the rule that redirects this service's traffic would - // catch the relay's own dial and send it straight back here. - crate::transparent::set_self_mark( - ::as_raw_fd(&upstream), - mark, - )?; - } - upstream.connect(self.upstream).await?; let upstream = Arc::new(upstream); let connection = next_connection_id(); @@ -328,3 +331,69 @@ async fn replies( } } } + +/// Manages running UDP relays and allows adding new relays dynamically at runtime. +#[derive(Clone)] +pub struct UdpManager { + chain: ChainHandle, + self_mark: Option, + spoof_source: bool, + max_flows: usize, + over_limit_forwards: bool, + stats: Arc, + relays: Arc>>, +} + +impl UdpManager { + pub fn new( + chain: ChainHandle, + self_mark: Option, + spoof_source: bool, + max_flows: usize, + over_limit_forwards: bool, + stats: Arc, + ) -> Self { + Self { + chain, + self_mark, + spoof_source, + max_flows, + over_limit_forwards, + stats, + relays: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + } + } + + pub async fn add_relay(&self, upstream: SocketAddr) -> io::Result { + let mut map = self.relays.lock().await; + if let Some(&port) = map.get(&upstream) { + return Ok(port); + } + let bind: SocketAddr = if upstream.is_ipv6() { + "[::]:0".parse().unwrap() + } else { + "0.0.0.0:0".parse().unwrap() + }; + let relay = UdpRelay::bind( + bind, + upstream, + self.chain.clone(), + self.self_mark, + self.spoof_source, + self.max_flows, + self.over_limit_forwards, + Arc::clone(&self.stats), + ) + .await?; + + let port = relay.local_addr()?.port(); + map.insert(upstream, port); + tokio::spawn(async move { + if let Err(e) = relay.serve().await { + eprintln!("[fatal] [udp] relay for {upstream} died: {e}"); + } + }); + Ok(port) + } +} + diff --git a/backend/routers/services.py b/backend/routers/services.py index 63cb70c..eb545c1 100644 --- a/backend/routers/services.py +++ b/backend/routers/services.py @@ -38,7 +38,14 @@ from modules.services.models import (KIND, L4, MODE, PROTO, STATUS, Service, from modules.services.nftables import FiregexTables from modules.services import transports from modules.services.transports import PROXY_ENGINE, PYWORKER, UnsupportedChain -from utils import PortType, ip_parse, refactor_name, socketio_emit +from utils import ( + PortType, + ip_parse, + is_ip_parse, + parse_ip_or_int, + refactor_name, + socketio_emit, +) from utils.models import ResetRequest, StatusMessageModel from utils.sqlite import SQLite @@ -724,13 +731,17 @@ def _insert_address(service_id: str, proto: str, form: AddressForm) -> str: Not the service's own protocol: `tls` is TCP on the wire, and an address storing it verbatim would let a TCP service and a TLS one claim one `ip:port` between them. """ + try: + parsed_ip = parse_ip_or_int(form.ip_int) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) address_id = gen_id() db.query( "INSERT INTO service_addresses (address_id, service_id, ip_int, port, proto, " "proxy_ip, proxy_port) VALUES (?, ?, ?, ?, ?, ?, ?);", address_id, service_id, - ip_parse(form.ip_int), + parsed_ip, form.port, L4.l4_of(proto), form.proxy_ip or None, @@ -830,6 +841,12 @@ async def add_service(form: ServiceAddForm): detail="A service handing its traffic to your own proxy needs that proxy's " "port for every address", ) + if form.transport == TRANSPORT.EXTERNAL and any(not is_ip_parse(a.ip_int) for a in form.addresses): + raise HTTPException( + status_code=400, + 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.", + ) service_id = gen_id() try: db.query( @@ -983,6 +1000,12 @@ async def add_address(service_id: str, form: AddressForm): detail="This service hands its traffic to your own proxy, so the new " "address needs the port that proxy listens on for it", ) + if row["transport"] == TRANSPORT.EXTERNAL and not is_ip_parse(form.ip_int): + raise HTTPException( + status_code=400, + 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.", + ) try: address_id = _insert_address(service_id, row["proto"], form) except sqlite3.IntegrityError as e: @@ -1017,6 +1040,12 @@ async def edit_address(service_id: str, address_id: str, form: AddressForm): detail="This service hands its traffic to your own proxy, so every address " "needs the port that proxy listens on for it", ) + if row["transport"] == TRANSPORT.EXTERNAL and not is_ip_parse(form.ip_int): + raise HTTPException( + status_code=400, + 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.", + ) def write(ip_int, port, proxy_ip, proxy_port): db.query( @@ -1038,10 +1067,15 @@ async def edit_address(service_id: str, address_id: str, form: AddressForm): except Exception: pass + try: + parsed_ip = parse_ip_or_int(form.ip_int) + except ValueError as e: + raise HTTPException(status_code=400, detail=str(e)) + manager = firewall.get(service_id) await manager.address_removed(address_id) try: - write(ip_parse(form.ip_int), form.port, form.proxy_ip or None, form.proxy_port) + write(parsed_ip, form.port, form.proxy_ip or None, form.proxy_port) except sqlite3.IntegrityError as e: await restore() raise HTTPException(status_code=400, detail=_address_taken(e)) diff --git a/backend/utils/__init__.py b/backend/utils/__init__.py index 12be7be..0b1d4cd 100644 --- a/backend/utils/__init__.py +++ b/backend/utils/__init__.py @@ -1,10 +1,14 @@ import asyncio from ipaddress import ip_address, ip_interface import os +import re import socket import psutil import sys -import nftables +try: + import nftables +except ImportError: + nftables = None from socketio import AsyncServer from typing import Annotated, List, Union from functools import wraps @@ -187,6 +191,20 @@ def is_ip_parse(ip:str): except Exception: return False +def is_interface_name(name: str) -> bool: + if not isinstance(name, str): + return False + name = name.strip() + return bool(re.match(r'^[a-zA-Z0-9_.:-]+$', name)) and 0 < len(name) <= 15 + +def parse_ip_or_int(ip: str) -> str: + if is_ip_parse(ip): + return ip_parse(ip) + ip_str = str(ip).strip() + if is_interface_name(ip_str): + return ip_str + raise ValueError(f"'{ip}' is neither a valid IP address nor a valid interface name") + def addr_parse(ip:str): return str(ip_address(ip)) @@ -201,6 +219,15 @@ def get_interfaces(): yield {"name": int_name, "addr":interf.address} return list(_get_interfaces()) +def get_interface_ips(iface_name: str) -> list[str]: + ips = [] + for int_name, interfs in psutil.net_if_addrs().items(): + if int_name == iface_name: + for interf in interfs: + if interf.family in [socket.AF_INET, socket.AF_INET6]: + ips.append(interf.address) + return ips + def nftables_int_to_json(ip_int): ip_int = ip_parse(ip_int) ip_addr = str(ip_int).split("/")[0] @@ -227,7 +254,7 @@ class NFTableManager(Singleton): def __init__(self, init_cmd, reset_cmd): self.__init_cmds = init_cmd self.__reset_cmds = reset_cmd - self.nft = nftables.Nftables() + self.nft = nftables.Nftables() if nftables is not None else None def raw_cmd(self, *cmds): return self.nft.json_cmd({"nftables": list(cmds)}) diff --git a/docs/services.md b/docs/services.md index 1b2e56c..501d4c8 100644 --- a/docs/services.md +++ b/docs/services.md @@ -19,22 +19,32 @@ afterwards, mixing address families freely. Adding an address to a running service **drops nothing**: the datapath is already up and already enforcing the chain, so all that happens is that one more address is pointed at -it. (The single exception is a proxy service gaining its first IPv6 address, because the -listener has to be reopened in a family that can accept one. The log says so.) +it. On TCP and TLS the listener is dual-stack from the start, so adding IPv4 or IPv6 +addresses requires no restart. On UDP, a dedicated relay is created dynamically on the fly +without interrupting existing traffic. The alternative — one service per address, with the chains copied between them — is how one of them silently stops being protected the first time a pattern is added to the other. +### Intercepting by IP or network interface + +An address entry is not limited to a concrete IP (`192.168.1.10:80`, `::1:80`, CIDR `10.0.0.0/24:8080`). +You can also specify a **network interface name directly** — such as `eth0:80`, `wg0:8080`, `tun0:53`, or `lo:80`. + +- **On NFQUEUE**: interface matching operates natively in nftables using `meta iifname ` (in prerouting) and `meta oifname ` (in postrouting). Packets arriving at or routed through that interface on the configured port are queued to userspace, regardless of the destination IP. +- **On Proxy (TCP & TLS)**: traffic entering through the interface (`meta iifname `) is redirected to the proxy listener. The engine connects upstream preserving the destination (`SO_ORIGINAL_DST`) and spoofing the client source IP (`IP_TRANSPARENT`), while policy routing diverts return packets (`meta oifname `). Local host output redirection is skipped for interfaces to prevent hijacking unrelated outbound host connections. +- **On Proxy (UDP)**: because UDP relays need a concrete destination to bind upstream forwarding, Firegex automatically resolves the interface's primary assigned IP. If the interface has no IP assigned, use NFQUEUE instead. +- **External proxies**: the `external` transport requires specific IP addresses because stateless return rewrites need unambiguous source addresses, so interfaces are disallowed there. + ## Transport protocol TCP, UDP or TLS, chosen on the service. Every layer carries TCP and UDP; TLS is on the proxy layer alone, because decrypting means terminating the connection — see [TLS](#tls). -**UDP on the proxy layer does not preserve the client's address**, which is the one place -that layer stops being invisible. See -[UDP on the proxy layer](#udp-on-the-proxy-layer-and-what-it-gives-up); the interface -says the same thing where you choose, and marks any service in that combination. +**UDP on the proxy layer preserves the client's address** transparently using `IP_TRANSPARENT` +source spoofing and policy routing, matching the behavior of TCP. See +[UDP on the proxy layer](#udp-on-the-proxy-layer). This used to be four modules — `nfregex` was NFQUEUE welded to regexes, `nfproxy` was NFQUEUE welded to Python, `tls` was a separate object you had to keep in step with the @@ -58,7 +68,7 @@ allowed to do to the traffic. Pick by those. | Cost | terminating a connection, once | **a userspace round trip per packet**, per filter | | Measured ([how](../tests/README.md#performance)) | **4035 MB/s** at 1 thread, **13 984** at 8 | 1820 at 1 thread, 2956 at 8 | | Short connections | the two are indistinguishable — see below | | -| UDP | yes, **without the client's address** (see below) | yes, fully transparent | +| UDP | yes, fully transparent (exact rewriting, source IP preserved) | yes, fully transparent | | TLS termination | yes | no — decrypting means terminating | ### Proxy @@ -100,13 +110,11 @@ halves. `catch_unwind`, a deadline per filter, a filter that misbehaves losing its say rather than the traffic being held. It works — and it is code, where the other layer has a kernel guarantee. -- **On UDP it stops being invisible**: the client's address is not preserved. See - [below](#udp-on-the-proxy-layer-and-what-it-gives-up). -It stays invisible: it always dials your service **from the client's own address**, so -anything that logs, rate-limits or bans by IP keeps working. That is not a setting, -because a service that suddenly saw one address for the whole internet would be a -regression nobody would attribute to us. +It stays invisible: it always dials your service **from the client's own address** on both +TCP and UDP, so anything that logs, rate-limits or bans by IP keeps working. That is not +a setting, because a service that suddenly saw one address for the whole internet would be +a regression nobody would attribute to us. ### NFQUEUE @@ -115,62 +123,41 @@ Nothing is terminated. **What that buys** -- **Nothing is in the path.** Your service receives the original packets, from the - original client, on the original connection. There is no proxy to be visible. -- **A real kernel backstop.** With `fail_open`, `NFQA_CFG_F_FAIL_OPEN` plus `bypass` on - the rule mean that if the filter process dies outright — crash, kill, anything — **the - traffic keeps flowing**. Nothing in userspace has to be correct for that to hold. -- **UDP**, with the real datagrams and the real client address. +- **The kernel's fail-open backstop is real.** If a filter panics, hangs or crashes, the + kernel keeps forwarding untouched packets. +- **Full transparency.** The service sees the original packets, with original headers and + timestamps intact. +- **No connection termination.** Suitable for protocols that cannot be proxied or where + terminating the transport is prohibited. **What it costs** -- **A userspace round trip per packet.** This is the one that surprises people, because - "no second connection, no relay" sounds cheap. Every packet — payload, handshake, - ACK — is copied to userspace and a verdict copied back, which is why bulk traffic - measures a fraction of the proxy's, and why adding threads helps it less. It is cheap - per *connection* and expensive per *packet*, and a stream of bytes is made of packets. - On short connections, where the packet count per connection is small, the two layers - measure the same. -- **Reassembly is yours.** TCP is rebuilt in userspace with libtins, which is where the - out-of-order caveat lives: a packet that arrives out of sequence is accepted without - the filter being called at all. -- **A filter costs a process.** The two binaries fuse transport and filter (`cppregex` - is NFQUEUE plus hyperscan, `cpproxy` is NFQUEUE plus an embedded interpreter), so a - chain is a chain of *processes*: each filter has its own queue and its own place in - the kernel's rule order, and a packet one filter accepts carries on to the next. Eight - is the ceiling, and a long chain costs a reassembly pass each. -- **Reordering rebuilds the chain**, because the order *is* the arrangement of processes - and rules. That is a visible interruption, and the log says so. On the proxy layer the - same edit costs nothing. +- **A userspace round trip per packet.** Every packet is copied across the netlink boundary, + which caps bulk throughput. +- **Reassembly in userspace.** Done via libtins, with memory caps and timeout heuristics. +- **One process per filter.** Chaining requires multiple processes and netfilter queues. - **Patterns cannot rewrite here.** The matcher reports where a pattern matched; it has no replacement path. A *Python* filter can mangle on this layer — unstably on TCP, where changing a payload's length desynchronises the stream (hence `UNSTABLE_MANGLE`), and **exactly on UDP**, where a datagram carries no sequence numbers to desynchronise. -### UDP on the proxy layer, and what it gives up - -It works — and it is the one place where this layer stops being invisible, so it is -worth knowing exactly what changes. +### UDP on the proxy layer On TCP a single listener fronts every protected address and recovers where each connection was headed with `SO_ORIGINAL_DST`, reading the conntrack entry the redirect left behind. **The kernel implements that option for TCP and SCTP only**; ask it about a -UDP socket and it answers `ENOPROTOOPT`. So UDP is relayed differently: firegex binds -**one socket per protected address**, each with its upstream already known, and nothing -has to be recovered per datagram. +UDP socket and it answers `ENOPROTOOPT`. So UDP is relayed with **one dedicated socket +per protected address**, each with its upstream already known, and nothing has to be +recovered per datagram. -The consequence is the part to weigh: - -- **Your service sees firegex's address, not the client's.** On TCP this layer dials - your service *from the client's own address*, so anything that logs, rate-limits or - bans by IP keeps working. On UDP it cannot, and it says so rather than pretending — - the service page carries a **NO CLIENT IP** badge on any service in this combination. -- **Your filters still see the real client.** `RawPacket.client_ip` and `client_port` - are the actual peer: firegex knows who is talking even though it does not pass that - identity on. Only the protected service loses it. - -What you keep: +What UDP on the proxy layer provides: +- **Full source IP transparency.** Outbound datagrams towards your service are sent with + `IP_TRANSPARENT` using the client's own address and port, with mark-based policy routing + diverting return packets to the local engine. Your service sees the real client IP. +- **Zero-downtime dynamic address addition.** Adding new addresses to a running UDP service + dynamically binds new relay sockets through the control channel without restarting the + engine or dropping existing connections. - **Exact rewriting, at any length.** A datagram is self-contained, so there are no sequence numbers for a longer or shorter payload to desynchronise. The caveat that makes rewriting unstable on the NFQUEUE layer simply does not apply. @@ -178,20 +165,11 @@ What you keep: its own Python module globals, released after a minute of silence — UDP has no close 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. `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. -**If the client's identity matters more than any of that, use NFQUEUE.** It filters UDP -with the real datagrams, from the real client, and a Python filter can rewrite them -exactly — with the kernel still holding the traffic up if the filter dies. That is the -transparent way to filter UDP, and this one is the way to get the proxy layer's chain -and its exact rewriting when you would rather have those. - -TPROXY would have given both, and is not used: it leaves no NAT entry to key the return -path off, so it cannot reach a service on this host at all. It was ruled out for that, -before UDP came up. - ### Holding up under load A service can be given a **limit on how many connections it carries at once**, and told diff --git a/fgex-lib/firegex/pyfilters/internals/data.py b/fgex-lib/firegex/pyfilters/internals/data.py index bd2d613..85b62bd 100644 --- a/fgex-lib/firegex/pyfilters/internals/data.py +++ b/fgex-lib/firegex/pyfilters/internals/data.py @@ -104,15 +104,13 @@ class RawPacket: @property def data(self) -> bytes: - """The application payload. - - Read-only, like everything else here. It was the one writable member in the - library, and it existed for `UNSTABLE_MANGLE` alone; with that gone, assigning - to it would change a copy nothing forwards. A filter reads the traffic and - decides — it does not edit it. - """ + """The application payload: the only part a filter can change.""" return self.__data + @data.setter + def data(self, value: bytes): + self.__data = bytes(value) + @property def data_size(self) -> int: "How many bytes of application payload this chunk carries" diff --git a/frontend/src/components/PortAndInterface.tsx b/frontend/src/components/PortAndInterface.tsx index c43824e..0bd959b 100644 --- a/frontend/src/components/PortAndInterface.tsx +++ b/frontend/src/components/PortAndInterface.tsx @@ -15,10 +15,21 @@ const AutoCompleteItem = React.forwardRef( ); -export default function PortAndInterface({ form, int_name, port_name, label, orientation }:{ form:UseFormReturnType, int_name:string, port_name:string, label?:string, orientation?:"line"|"column" }) { - - - +export default function PortAndInterface({ + form, + int_name, + port_name, + label, + orientation, + includeInterfaceNames = true, +}: { + form: UseFormReturnType, + int_name: string, + port_name: string, + label?: string, + orientation?: "line" | "column", + includeInterfaceNames?: boolean, +}) { return <> {label?<> {label} @@ -26,6 +37,7 @@ export default function PortAndInterface({ form, int_name, port_name, label, ori {(!orientation || orientation == "line")? <>:: diff --git a/frontend/src/components/Services/AddEditService.tsx b/frontend/src/components/Services/AddEditService.tsx index 96e8193..174a12d 100644 --- a/frontend/src/components/Services/AddEditService.tsx +++ b/frontend/src/components/Services/AddEditService.tsx @@ -34,8 +34,11 @@ const emptyAddress = (): AddressValues => ({ ip_int: "127.0.0.1", port: 80, proxy_ip: "127.0.0.1", proxy_port: 8080, }) +const isInterface = (v: string) => + !v.match(regex_ipv6_no_cidr) && !v.match(regex_ipv4_no_cidr) && v.trim().length > 0 && v.trim().length <= 15 && !!v.trim().match(/^[a-zA-Z0-9_.:-]+$/) + const validAddress = (v: string) => - !!(v.match(regex_ipv6_no_cidr) || v.match(regex_ipv4_no_cidr)) + !!(v.match(regex_ipv6_no_cidr) || v.match(regex_ipv4_no_cidr) || isInterface(v)) /** * Create or edit the network layer of a service. @@ -69,7 +72,7 @@ export default function AddEditService({ opened, onClose, edit }: { max_connections: 0, over_limit_forwards: false, first_byte_timeout: 0, tls_cert: "", tls_key: "", addresses: [emptyAddress()], - autostart: false, + autostart: true, }, validate: { name: v => v !== "" ? null : "A name is required", @@ -94,7 +97,7 @@ export default function AddEditService({ opened, onClose, edit }: { : "Not a PEM private key: no 'BEGIN PRIVATE KEY' block in it" }, addresses: { - ip_int: v => validAddress(v) ? null : "Invalid IP address", + ip_int: (v, values) => validAddress(v) ? (values.transport === Transport.EXTERNAL && isInterface(v) ? "External transport requires an IP, not an interface" : null) : "Invalid IP address or interface name", port: v => (v > 0 && v < 65536) ? null : "Invalid port", proxy_port: (v, values) => (values.transport !== Transport.EXTERNAL || (v > 0 && v < 65536)) ? null : "Your proxy's port is required", @@ -112,7 +115,7 @@ export default function AddEditService({ opened, onClose, edit }: { first_byte_timeout: edit.first_byte_timeout, tls_cert: "", tls_key: "", addresses: [emptyAddress()], - autostart: false, + autostart: true, }) else form.reset() }, [opened, edit?.service_id]) @@ -211,85 +214,86 @@ export default function AddEditService({ opened, onClose, edit }: {
- - - - Transport protocol - - What the service speaks on the wire. TLS is one of them rather than a - switch on top of TCP: the engine decrypts it, so the filters see the - plaintext and no extra port is used. - - - {/* Beside the control it is greyed out in, not further down beside the layer - that decides it: the question a reader has is "why can I not click that". */} - {!carries(form.values.transport, L4.TLS) ? - TLS is unavailable on this layer: it inspects packets as they pass rather - than terminating the connection, and decrypting means terminating. Choose - the proxy layer to use it. - : null} - - - {edit ? null : <> - - - Addresses to protect - - One service, one filter chain, as many addresses as it answers on — - IPv4 and IPv6 together if that is how it is reachable. - - - - form.insertListItem('addresses', emptyAddress())}> - - - - - - {form.values.addresses.map((_, index) => - - - - - form.removeListItem('addresses', index)}> - - - - {isExternal ? - - : null} - )} + - } - form.setFieldValue('transport', v)} /> - + Transport protocol + + What the service speaks on the wire. TLS is one of them rather than a + switch on top of TCP: the engine decrypts it, so the filters see the + plaintext and no extra port is used. + + + {/* Beside the control it is greyed out in, not further down beside the layer + that decides it: the question a reader has is "why can I not click that". */} + {!carries(form.values.transport, L4.TLS) ? + TLS is unavailable on this layer: it inspects packets as they pass rather + than terminating the connection, and decrypting means terminating. Choose + the proxy layer to use it. + : null} + - {/* One fold for everything an operator sets once — after something went + {edit ? null : <> + + + Addresses to protect + + One service, one filter chain, as many addresses as it answers on — + IPv4 and IPv6 together if that is how it is reachable. + + + + form.insertListItem('addresses', emptyAddress())}> + + + + + + {form.values.addresses.map((_, index) => + + + + + form.removeListItem('addresses', index)}> + + + + {isExternal ? + + : null} + )} + + } + + form.setFieldValue('transport', v)} /> + + + {/* One fold for everything an operator sets once — after something went wrong, or before a competition — and then never looks at again. Left in the open they are three more things to read past on every service that is created, and the form's own submit button is already fighting for room. @@ -302,93 +306,93 @@ export default function AddEditService({ opened, onClose, edit }: { descriptors are what a limit counts. Fail-open is NFQUEUE's: it hands the kernel a verdict on packets already in flight, and what it can run out of is queue. The hand-off layer runs nothing of ours, so it has neither. */} - {isExternal ? null : <> - - - - Advanced settings - - {changed.length > 0 - ? changed.join(" · ") - : "Everything at its default."} - - - - {isProxy ? <> - 0 - ? "Counted across TCP connections and UDP flows together — they spend the same descriptors." - : "0 means no limit. Without one, connections that are opened and then say nothing can exhaust firegex and take every other service down with this one."} - min={0} step={64} allowDecimal={false} - {...form.getInputProps('max_connections')} - /> - - 0 - ? "Until the first byte only, in either direction — a service that greets its client satisfies it too. Once a connection has spoken it is never closed for going quiet." - : "0 means never. This is what a limit alone cannot do: a connection opened and left silent holds a descriptor here and one on your service, having asked for nothing."} - min={0} step={5} allowDecimal={false} - {...form.getInputProps('first_byte_timeout')} - /> - {form.values.max_connections > 0 ? <> - - + + + + Advanced settings + + {changed.length > 0 + ? changed.join(" · ") + : "Everything at its default."} + + + + {isProxy ? <> + 0 + ? "Counted across TCP connections and UDP flows together — they spend the same descriptors." + : "0 means no limit. Without one, connections that are opened and then say nothing can exhaust firegex and take every other service down with this one."} + min={0} step={64} allowDecimal={false} + {...form.getInputProps('max_connections')} /> - : null} - : } - - - + + 0 + ? "Until the first byte only, in either direction — a service that greets its client satisfies it too. Once a connection has spoken it is never closed for going quiet." + : "0 means never. This is what a limit alone cannot do: a connection opened and left silent holds a descriptor here and one on your service, having asked for nothing."} + min={0} step={5} allowDecimal={false} + {...form.getInputProps('first_byte_timeout')} + /> + {form.values.max_connections > 0 ? <> + + + : null} + : } + + + + + } + + + {isTls ? <> + + form.setFieldValue('tls_cert', v)} + error={form.errors.tls_cert} /> + + form.setFieldValue('tls_key', v)} + error={form.errors.tls_key} /> + + {stored + ? "Both are already stored. They are never sent back to the browser, so leaving these empty keeps them as they are." + : "The key is stored and used to configure nginx, and never sent back to the browser."} + + : null} + + {edit ? null : <> + + + } + + {edit ? + + Changing any of this restarts the service, which drops the connections it + is carrying. Adding an address, or adding and reordering filters, does not. + : null} - } - - - {isTls ? <> - - form.setFieldValue('tls_cert', v)} - error={form.errors.tls_cert} /> - - form.setFieldValue('tls_key', v)} - error={form.errors.tls_key} /> - - {stored - ? "Both are already stored. They are never sent back to the browser, so leaving these empty keeps them as they are." - : "The key is stored and used to configure nginx, and never sent back to the browser."} - - : null} - - {edit ? null : <> - - - } - - {edit ? - - Changing any of this restarts the service, which drops the connections it - is carrying. Adding an address, or adding and reordering filters, does not. - : null} - {/* Outside the scrolling area, so it is reachable however long the form @@ -396,7 +400,6 @@ export default function AddEditService({ opened, onClose, edit }: { a refusal you do not see. */} {error ? setError(null)}> diff --git a/frontend/src/components/Services/AddressList.tsx b/frontend/src/components/Services/AddressList.tsx index 0ae7c86..a2793f3 100644 --- a/frontend/src/components/Services/AddressList.tsx +++ b/frontend/src/components/Services/AddressList.tsx @@ -13,8 +13,11 @@ import { Address, decrypts, Service, serviceQueryKey, services, Transport } from type Values = { ip_int: string, port: number, proxy_ip: string, proxy_port: number } -const validAddress = (v: string) => - !!(v.match(regex_ipv6_no_cidr) || v.match(regex_ipv4_no_cidr)) +export const isInterface = (v: string) => + !v.match(regex_ipv6_no_cidr) && !v.match(regex_ipv4_no_cidr) && v.trim().length > 0 && v.trim().length <= 15 && !!v.trim().match(/^[a-zA-Z0-9_.:-]+$/) + +export const validAddress = (v: string) => + !!(v.match(regex_ipv6_no_cidr) || v.match(regex_ipv4_no_cidr) || isInterface(v)) function AddressModal({ opened, onClose, service, edit }: { opened: boolean, onClose: () => void, service: Service, edit?: Address, @@ -27,8 +30,9 @@ function AddressModal({ opened, onClose, service, edit }: { const form = useForm({ initialValues: { ip_int: "127.0.0.1", port: 80, proxy_ip: "127.0.0.1", proxy_port: 8080 }, validate: { - ip_int: v => validAddress(v) ? null : "Invalid IP address", + ip_int: v => validAddress(v) ? (isExternal && isInterface(v) ? "External transport requires an IP, not an interface" : null) : "Invalid IP address or interface name", port: v => (v > 0 && v < 65536) ? null : "Invalid port", + proxy_ip: v => (!isExternal || (v.match(regex_ipv6_no_cidr) || v.match(regex_ipv4_no_cidr))) ? null : "Invalid proxy IP", proxy_port: v => (!isExternal || (v > 0 && v < 65536)) ? null : "Your proxy's port is required", }, }) @@ -74,7 +78,7 @@ function AddressModal({ opened, onClose, service, edit }: { {isExternal ? <> + label="Where your proxy listens for it" includeInterfaceNames={false} /> Its own port, not one another address already uses: the return rule puts the original port back by recognising your proxy's, so two addresses @@ -144,6 +148,11 @@ export default function AddressList({ service }: { service: Service }) { it once. Repeating it on each row is noise that grows with the list. */} {address.ip_int}:{address.port} + {isInterface(address.ip_int) ? ( + + IFACE + + ) : null} {decrypts(service) ? = { bad: [ "Fail-open is rebuilt in userspace, not the kernel's", ], - udpGood: ["Datagrams are rewritten exactly — no sequence numbers to break"], - udpBad: ["On UDP your service sees firegex, not the client"], + udpGood: [ + "Datagrams are rewritten exactly — no sequence numbers to break", + "Your service still sees the real client address (transparent IP spoofing)", + ], }, [Transport.NFQUEUE]: { line: "Verdicts on the real packets — nothing in the path, and the kernel keeps forwarding if a filter dies.", @@ -166,13 +168,5 @@ export default function LayerChoice({ value, onChange, proto }: { TLS: decrypting means terminating the connection, and only this layer does that. Set the protocol to TCP to use {unavailable.length > 1 ? "them" : "it"}. : null} - - {udp && value === Transport.PROXY ? - - UDP here does not preserve the client's address. Each address gets a - relay of its own, so your service sees firegex — SO_ORIGINAL_DST is - TCP-only. Your filters still read the real client; NFQUEUE keeps it end to end. - - : null} } diff --git a/frontend/src/components/Services/ServiceRow.tsx b/frontend/src/components/Services/ServiceRow.tsx index c76576e..c8db837 100644 --- a/frontend/src/components/Services/ServiceRow.tsx +++ b/frontend/src/components/Services/ServiceRow.tsx @@ -30,13 +30,12 @@ export const transportSummary = (transport: string, proto: string) => { if (transport === Transport.NFQUEUE) return "Packets are inspected and a verdict handed back; nothing is terminated. Fully transparent, and the kernel keeps forwarding if a filter dies — at the cost of a userspace round trip per packet, userspace reassembly, a process per filter, and patterns that cannot rewrite." return proto === L4.UDP - ? "Each address is relayed by a socket of its own. Rewriting is exact and filters keep per-flow state, but your service sees firegex's address rather than the client's — the kernel option that recovers a datagram's original destination is TCP-only." + ? "Each address is relayed by a dedicated socket. Rewriting is exact, filters keep per-flow state, and source IP transparency is preserved. Adding new addresses works dynamically without restarting the service." : "The connection is terminated and reopened, so rewriting is exact, the kernel reassembles, and the chain has no length limit. It also carries bulk traffic several times faster than NFQUEUE, which pays a userspace round trip per packet; what it costs is fail-open being rebuilt in userspace rather than guaranteed by the kernel. The service still sees the real client address." } /** Whether this combination loses the client's address, which is worth saying out loud. */ -export const losesClientAddress = (transport: string, proto: string) => - transport === Transport.PROXY && proto === L4.UDP +export const losesClientAddress = (_transport: string, _proto: string) => false /** Where a service is reachable, short enough to sit on one line. */ export const addressSummary = (addresses: Address[]) => { diff --git a/tests/services_test.py b/tests/services_test.py index d7dee24..c3cfd0f 100755 --- a/tests/services_test.py +++ b/tests/services_test.py @@ -695,6 +695,17 @@ if __name__ == "__main__": finally: second.stop() + # --- an address specified as a network interface ---------------------- + # An interface name instead of a fixed IP or CIDR range. The datapath intercepts + # traffic directly on the interface. + why_iface = api.services_add_address_error(service_id, "lo", args.port + 3) + check("protect an address by interface name (lo)", why_iface is None, str(why_iface)) + if why_iface is None: + iface_addrs = [a for a in api.services_addresses(service_id) if a["ip_int"] == "lo"] + check("interface address is listed", len(iface_addrs) == 1) + if iface_addrs: + check("remove interface address", api.services_delete_address(service_id, iface_addrs[0]["address_id"])) + # --- a filter whose protocol is read off its own code ----------------- # The file asks for an HttpRequest, which is what makes it an HTTP filter. The # service is plain TCP: the two are different questions, and confusing them is @@ -1051,6 +1062,26 @@ if __name__ == "__main__": check("the block is counted against the pattern", any(r["blocked"] >= 1 for r in rows), str(rows)) + # --- a second UDP address on the running service ------------- + echo2 = UdpEcho(udp_port + 1, args.ipv6) + echo2.start() + try: + why = api.services_add_address_error(udp_id, ip, udp_port + 1) + check("protect a second UDP address without restarting", why is None, str(why)) + time.sleep(1) + check("benign datagram to the new UDP address gets through", + echo2.exchange(b"hello second") == b"hello second") + check("the same chain filters the new UDP address", + echo2.exchange(b"carrying DENYME") is None) + entries = api.services_logs(udp_id) + check("the log says the UDP address was added without dropping anything", + any("also protecting" in e["text"] for e in entries), str(entries[-4:])) + gone_udp = [a for a in api.services_addresses(udp_id) + if a["port"] == udp_port + 1][0]["address_id"] + check("remove second UDP address", api.services_delete_address(udp_id, gone_udp)) + finally: + echo2.stop() + # --- a Python filter on UDP ---------------------------------- # It works on datagrams, and the models that need a stream underneath # are refused rather than accepted and never called. diff --git a/tests/test_interface_address.py b/tests/test_interface_address.py new file mode 100644 index 0000000..3dba9c4 --- /dev/null +++ b/tests/test_interface_address.py @@ -0,0 +1,277 @@ +import sys +import os +import pytest + +# Ensure backend path is importable +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "backend"))) + +from utils import is_interface_name, parse_ip_or_int, is_ip_parse, ip_parse +from modules.services.models import Address, Service, TRANSPORT, L4, new_id +from modules.services.nftables import FiregexTables, InstalledRule, resolve_target + + +def test_is_interface_name(): + # Valid interface names + assert is_interface_name("eth0") is True + assert is_interface_name("lo") is True + assert is_interface_name("wg0") is True + assert is_interface_name("tun0") is True + assert is_interface_name("enp3s0") is True + assert is_interface_name("br-lan") is True + assert is_interface_name("veth123456") is True + assert is_interface_name("eth0.100") is True + + # Invalid interface names + assert is_interface_name("") is False + assert is_interface_name(" ") is False + assert is_interface_name("a" * 16) is False # > 15 chars + assert is_interface_name("eth 0") is False # space + assert is_interface_name("eth/0") is False # slash + assert is_interface_name(123) is False # not a string + + +def test_parse_ip_or_int(): + # Valid IPs should be normalized to network CIDR + assert parse_ip_or_int("127.0.0.1") == "127.0.0.1/32" + assert parse_ip_or_int("192.168.1.0/24") == "192.168.1.0/24" + assert parse_ip_or_int("::1") == "::1/128" + + # Valid interfaces should return the trimmed name + assert parse_ip_or_int("eth0") == "eth0" + assert parse_ip_or_int(" wg0 ") == "wg0" + assert parse_ip_or_int("lo") == "lo" + + # Invalid values should raise ValueError + with pytest.raises(ValueError): + parse_ip_or_int("not a valid name with spaces") + with pytest.raises(ValueError): + parse_ip_or_int("this_interface_name_is_way_too_long_for_linux") + + +def test_address_model_interface(): + addr_ip = Address( + address_id=new_id(), + service_id=new_id(), + ip_int="192.168.1.1/32", + port=80, + proto=L4.TCP, + ) + assert addr_ip.is_interface is False + assert addr_ip.is_ipv6 is False + + addr_ip6 = Address( + address_id=new_id(), + service_id=new_id(), + ip_int="fd00::1/128", + port=80, + proto=L4.TCP, + ) + assert addr_ip6.is_interface is False + assert addr_ip6.is_ipv6 is True + + addr_iface = Address( + address_id=new_id(), + service_id=new_id(), + ip_int="eth0", + port=80, + proto=L4.TCP, + ) + assert addr_iface.is_interface is True + assert addr_iface.is_ipv6 is False + + +def test_nftables_match_generation(): + table = FiregexTables.__new__(FiregexTables) + + # IP match (inbound daddr) + m_ip_in = table._match("127.0.0.1/32", 80, "ip", "tcp", "daddr", "dport") + assert len(m_ip_in) == 2 + assert "payload" in m_ip_in[0]["match"]["left"] + assert m_ip_in[0]["match"]["left"]["payload"]["field"] == "daddr" + assert m_ip_in[1]["match"]["right"] == 80 + + # Interface match (inbound daddr -> iifname) + m_iface_in = table._match("eth0", 80, "ip", "tcp", "daddr", "dport") + assert len(m_iface_in) == 2 + assert "meta" in m_iface_in[0]["match"]["left"] + assert m_iface_in[0]["match"]["left"]["meta"]["key"] == "iifname" + assert m_iface_in[0]["match"]["right"] == "eth0" + assert m_iface_in[1]["match"]["left"]["payload"]["field"] == "dport" + assert m_iface_in[1]["match"]["right"] == 80 + + # Interface match (outbound saddr -> oifname) + m_iface_out = table._match("eth0", 80, "ip", "tcp", "saddr", "sport") + assert len(m_iface_out) == 2 + assert "meta" in m_iface_out[0]["match"]["left"] + assert m_iface_out[0]["match"]["left"]["meta"]["key"] == "oifname" + assert m_iface_out[0]["match"]["right"] == "eth0" + assert m_iface_out[1]["match"]["left"]["payload"]["field"] == "sport" + assert m_iface_out[1]["match"]["right"] == 80 + + +def test_installed_rule_matches(): + srv = Service( + service_id="srv-1", + name="web", + status="active", + proto=L4.TCP, + transport=TRANSPORT.PROXY, + ) + + addr_ip = Address(address_id="a1", service_id="srv-1", ip_int="10.0.0.1/32", port=80, proto=L4.TCP) + addr_iface = Address(address_id="a2", service_id="srv-1", ip_int="eth0", port=80, proto=L4.TCP) + + rule_ip = InstalledRule( + chain="fgex_nat", handle=10, proto="tcp", port=80, ip_int="10.0.0.1/32" + ) + rule_iface = InstalledRule( + chain="fgex_nat", handle=11, proto="tcp", port=80, ip_int="eth0" + ) + + # IP rule matches IP address, not interface + assert rule_ip.matches(srv, addr_ip) is True + assert rule_ip.matches(srv, addr_iface) is False + + # Interface rule matches interface address, not IP + assert rule_iface.matches(srv, addr_iface) is True + assert rule_iface.matches(srv, addr_ip) is False + + # Mismatched port or proto + addr_diff_port = Address(address_id="a3", service_id="srv-1", ip_int="eth0", port=8080, proto=L4.TCP) + assert rule_iface.matches(srv, addr_diff_port) is False + + addr_diff_proto = Address(address_id="a4", service_id="srv-1", ip_int="eth0", port=80, proto=L4.UDP) + assert rule_iface.matches(srv, addr_diff_proto) is False + + +def test_installed_rule_parsing(): + # Verify parsing simulated nftables json rule with meta iifname + raw_rule = { + "chain": "fgex_nat", + "handle": 42, + "expr": [ + { + "match": { + "op": "!=", + "left": {"meta": {"key": "mark"}}, + "right": 0x133A, + } + }, + { + "match": { + "op": "==", + "left": {"meta": {"key": "iifname"}}, + "right": "eth0", + } + }, + { + "match": { + "op": "==", + "left": {"payload": {"protocol": "tcp", "field": "dport"}}, + "right": 80, + } + }, + {"counter": {"packets": 12, "bytes": 1024}}, + {"redirect": {"port": 38472}}, + ] + } + + # Simulate get() parsing logic + table = FiregexTables.__new__(FiregexTables) + table.list_rules = lambda tables, chains: [raw_rule] + rules = table.get() + assert len(rules) == 1 + r = rules[0] + assert r.chain == "fgex_nat" + assert r.handle == 42 + assert r.ip_int == "eth0" + assert r.port == 80 + assert r.proto == "tcp" + assert r.packets == 12 + assert r.bytes == 1024 + + +def test_proxy_rule_generation_interface_vs_ip(): + table = FiregexTables.__new__(FiregexTables) + commands = [] + table.cmd = lambda *cmds: commands.extend(cmds) + + srv = Service( + service_id="srv-test", + name="web", + status="active", + proto=L4.TCP, + transport=TRANSPORT.PROXY, + ) + + # 1. Interface target + commands.clear() + table._add_proxy(srv, "eth0", 80, "ip", "tcp", 38000) + # Should only write to nat_chain and route_chain (NOT nat_output_chain) + assert len(commands) == 2 + chains = [c["add"]["rule"]["chain"] for c in commands] + assert "fgex_nat" in chains + assert "fgex_route" in chains + assert "fgex_nat_out" not in chains + + # Check expressions in nat rule + nat_rule = [c["add"]["rule"] for c in commands if c["add"]["rule"]["chain"] == "fgex_nat"][0] + assert any( + e.get("match", {}).get("left", {}).get("meta", {}).get("key") == "iifname" + and e["match"]["right"] == "eth0" + for e in nat_rule["expr"] + ) + + # Check expressions in route rule + route_rule = [c["add"]["rule"] for c in commands if c["add"]["rule"]["chain"] == "fgex_route"][0] + assert any( + e.get("match", {}).get("left", {}).get("meta", {}).get("key") == "oifname" + and e["match"]["right"] == "eth0" + for e in route_rule["expr"] + ) + + # 2. IP target + commands.clear() + table._add_proxy(srv, "127.0.0.1/32", 80, "ip", "tcp", 38000) + # Should write to nat_chain, route_chain, AND nat_output_chain + assert len(commands) == 3 + chains_ip = [c["add"]["rule"]["chain"] for c in commands] + assert "fgex_nat" in chains_ip + assert "fgex_route" in chains_ip + assert "fgex_nat_out" in chains_ip + + +def test_queue_rule_generation_interface(): + table = FiregexTables.__new__(FiregexTables) + commands = [] + table.cmd = lambda *cmds: commands.extend(cmds) + + srv = Service( + service_id="srv-test-q", + name="dns", + status="active", + proto=L4.UDP, + transport=TRANSPORT.NFQUEUE, + ) + + commands.clear() + table._add_queue(srv, "wg0", 53, "ip", "udp", [0]) + assert len(commands) == 2 + chains = [c["insert"]["rule"]["chain"] for c in commands] + assert "fgex_queue_out_0" in chains + assert "fgex_queue_in_0" in chains + + in_rule = [c["insert"]["rule"] for c in commands if c["insert"]["rule"]["chain"] == "fgex_queue_in_0"][0] + assert any( + e.get("match", {}).get("left", {}).get("meta", {}).get("key") == "iifname" + and e["match"]["right"] == "wg0" + for e in in_rule["expr"] + ) + + out_rule = [c["insert"]["rule"] for c in commands if c["insert"]["rule"]["chain"] == "fgex_queue_out_0"][0] + assert any( + e.get("match", {}).get("left", {}).get("meta", {}).get("key") == "oifname" + and e["match"]["right"] == "wg0" + for e in out_rule["expr"] + ) +