feat: decide who may connect at all, by source address

mirror has no accounts, so on a real network the address a connection arrives
from is not a nicety — it is the entire access control. Before this change the
answer was "everyone who can open a TCP connection", which is fine for a laptop
demo and alarming for a box someone leaves running.

`mirror serve --allow-cidr 10.2.0.0/22` (or MIRROR_ALLOW_CIDRS) now restricts
pages, the JSON API and the signaling socket to the networks named, and it is
strict where it needs to be: a malformed CIDR exits 2 with one line instead of
starting a server whose list silently failed to parse, because the two
outcomes of that typo are "nobody can connect" and "everybody can connect".
Empty remains the only permissive state, on purpose.

The rules that matter live in src/mirror/access.py and are the ones that get
either outcome:

  * X-Forwarded-For is believed only from MIRROR_TRUSTED_PROXIES (loopback by
    default), and only its left-most value. An untrusted peer that sends the
    header gets judged on its own address — there is a test for exactly that,
    because it is the shape of a bypass.
  * The socket is closed before it accepts, so a refused address never creates
    a roster entry to clean up behind it.
  * An address that cannot be parsed fails closed, and a v6 address is never
    "inside" a v4 network (comparing the two raises TypeError, which would be
    a 500 on a dual-stack host).
  * /healthz is exempt: the container health probe comes from loopback, which
    a production list otherwise excludes.

The refused client is told why, and which address it was judged as — "403" in
front of a proxy is undiagnosable otherwise. Browsers get a page, fetches get
JSON, and the page is self-contained because a refused address cannot load
/static/app.css any more than it can load a page. net.js keys its own notice on
the /api/config 403 and stops: without that the pill reads "offline,
reconnecting in 32s" forever for a decision that will never change.

The static files drop their only inline script and style attribute (the landing
page's counter is /static/home.js, its fine print is .fineprint) so the
deployment can send `script-src 'self'; style-src 'self'` with no
'unsafe-inline'. That policy was verified through a real nginx edge carrying the
exact header set — capture through SRTP to decoded, painted frames at
1280x720, eight distinct, zero securitypolicyviolation events — which also
caught the edge sending `Cache-Control: no-store, no-store` and needed
proxy_hide_header, since nginx appends rather than replaces.

24 tests added; 176 pass, 2 skipped, pyright and ruff clean.

LLM-Generated: true

tests/e2e/conftest.py

@@ -13,6 +13,7 @@ import socket
 import subprocess
 import sys
 import time
+from collections.abc import Iterator, Sequence
 from pathlib import Path
 
 import pytest
@@ -40,9 +41,8 @@ def _free_port() -> int:
         return probe.getsockname()[1]
 
 
-@pytest.fixture(scope="session")
-def base_url() -> str:
-    """A real `mirror serve` process on a real port — not a test double."""
+def _start(extra_args: Sequence[str] = ()) -> tuple[str, subprocess.Popen[str]]:
+    """Spawn a real `mirror serve` on a free port and wait for it to answer."""
     import httpx
 
     port = _free_port()
@@ -58,6 +58,7 @@ def base_url() -> str:
             str(port),
             "--log-level",
             "warning",
+            *extra_args,
         ],
         cwd=REPO_ROOT,
         stdout=subprocess.PIPE,
@@ -79,17 +80,18 @@ def base_url() -> str:
             output = process.stdout.read() if process.stdout else ""
             raise RuntimeError(f"mirror serve exited early ({process.returncode}):\n{output}")
         try:
+            # /healthz is the readiness probe *and* the one route an address gate
+            # never refuses, so it works for a restricted server too.
             if httpx.get(f"{url}/healthz", timeout=1.0).status_code == 200:
-                break
+                return url, process
         except Exception as exc:  # still booting
             last_error = exc
         time.sleep(0.15)
-    else:
-        process.terminate()
-        raise RuntimeError(f"mirror serve never came up: {last_error}")
+    process.terminate()
+    raise RuntimeError(f"mirror serve never came up: {last_error}")
 
-    yield url
 
+def _stop(process: subprocess.Popen[str]) -> None:
     process.terminate()
     try:
         process.wait(timeout=5)
@@ -97,6 +99,26 @@ def base_url() -> str:
         process.kill()
 
 
+@pytest.fixture(scope="session")
+def base_url() -> Iterator[str]:
+    """A real `mirror serve` process on a real port — not a test double."""
+    url, process = _start()
+    yield url
+    _stop(process)
+
+
+@pytest.fixture(scope="session")
+def restricted_url() -> Iterator[str]:
+    """A server that admits only 203.0.113.0/24 — so it refuses this machine.
+
+    That range is reserved for documentation by RFC 5737, which is why it is safe
+    to use as "everybody else": nothing real is ever locked out by it.
+    """
+    url, process = _start(["--allow-cidr", "203.0.113.0/24"])
+    yield url
+    _stop(process)
+
+
 @pytest.fixture(scope="session")
 def browser():
     with sync_playwright() as playwright:

tests/e2e/test_live_mirror.py

@@ -267,3 +267,36 @@ class TestRosterTruthfulness:
         mirror = payload["mirrors"][0]
         assert mirror["width"] and mirror["height"]
         assert mirror["displaySurface"] in ("monitor", "window", "browser")
+
+
+class TestAddressGate:
+    """`--allow-cidr` in a real process, refused by a real browser.
+
+    ``tests/test_app.py::TestAddressGate`` covers the JSON refusal the pages key
+    their own notice on, the proxy unwrapping, and the socket closing before it
+    accepts. This covers what a person actually gets when a server refuses them:
+    an explanation naming the address it judged, not an empty wall that looks
+    like a bug in the tool.
+    """
+
+    @pytest.mark.parametrize("path", ["/mirror", "/watch"])
+    def test_a_refused_address_sees_why_and_none_of_the_app(self, browser, restricted_url, path):
+        context = browser.new_context(viewport={"width": 1100, "height": 720})
+        page = context.new_page()
+        try:
+            response = page.goto(f"{restricted_url}{path}")
+            assert response.status == 403
+            expect(page.locator("body")).to_contain_text("not open to your address")
+            expect(page.locator("body")).to_contain_text("203.0.113.0/24")
+            # Which address it judged — without this, "you are not allowed" in
+            # front of a proxy is undiagnosable.
+            expect(page.locator("body")).to_contain_text("127.0.0.1")
+            # None of the app shell leaked past the gate.
+            expect(page.locator("#start, #grid, #notice")).to_have_count(0)
+        finally:
+            context.close()
+
+    def test_the_health_probe_is_not_gated(self, restricted_url):
+        import httpx
+
+        assert httpx.get(f"{restricted_url}/healthz", timeout=5).status_code == 200

tests/test_access.py

@@ -0,0 +1,120 @@
+"""The source-address gate: who may use this server at all."""
+
+from __future__ import annotations
+
+import pytest
+
+from mirror.access import (
+    DEFAULT_TRUSTED_PROXIES,
+    ConfigError,
+    Gate,
+    format_cidrs,
+    parse_cidrs,
+)
+
+
+class TestParseCidrs:
+    def test_empty_means_no_networks(self):
+        assert parse_cidrs("") == ()
+        assert parse_cidrs(None) == ()
+        assert parse_cidrs("  ,  ") == ()
+
+    def test_networks_addresses_and_both_families(self):
+        networks = parse_cidrs("10.2.0.0/22, 192.168.4.7, fd00::/64")
+        assert [str(network) for network in networks] == [
+            "10.2.0.0/22",
+            "192.168.4.7/32",
+            "fd00::/64",
+        ]
+
+    def test_separators_people_actually_type(self):
+        assert len(parse_cidrs("10.2.0.0/22; 10.3.0.0/22\n10.4.0.0/22")) == 3
+
+    def test_a_host_part_that_does_not_fit_its_mask_becomes_the_network_meant(self):
+        # Someone pasting "10.2.1.5/22" means the /22 they live in, not nothing.
+        assert str(parse_cidrs("10.2.1.5/22")[0]) == "10.2.0.0/22"
+
+    def test_duplicates_collapse(self):
+        assert parse_cidrs("10.2.0.0/22,10.2.0.0/22") == parse_cidrs("10.2.0.0/22")
+
+    @pytest.mark.parametrize("raw", ["laptop", "10.2.0.0/33", "10.2.0.0/22/x", "999.1.1.1"])
+    def test_nonsense_is_refused_at_startup_not_at_request_time(self, raw):
+        with pytest.raises(ConfigError, match="MIRROR_ALLOW_CIDRS"):
+            parse_cidrs(raw)
+
+    def test_the_error_names_the_source_so_two_settings_are_distinguishable(self):
+        with pytest.raises(ConfigError, match="MIRROR_TRUSTED_PROXIES"):
+            parse_cidrs("nope", source="MIRROR_TRUSTED_PROXIES")
+        with pytest.raises(ConfigError, match="--allow-cidr"):
+            parse_cidrs("nope", source="--allow-cidr")
+
+    def test_formatting_round_trips_for_a_status_line(self):
+        assert format_cidrs(parse_cidrs("10.2.0.0/22,fd00::/64")) == "10.2.0.0/22, fd00::/64"
+
+
+class TestGatePermits:
+    def test_no_networks_listed_means_the_gate_is_open(self):
+        gate = Gate()
+        assert gate.enabled is False
+        assert gate.permits("8.8.8.8")
+        assert gate.permits("10.2.1.5")
+
+    def test_the_listed_network_is_in_and_everything_else_is_out(self):
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22"))
+        assert gate.enabled is True
+        assert gate.permits("10.2.3.255")  # last address of the /22
+        assert not gate.permits("10.2.4.0")  # one past it
+        assert not gate.permits("10.1.99.99")
+
+    def test_an_address_we_cannot_parse_fails_closed(self):
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22"))
+        assert not gate.permits(None)
+        assert not gate.permits("testclient")
+        assert not gate.permits("")
+
+    def test_a_v6_address_is_never_inside_a_v4_network(self):
+        # ipaddress raises TypeError on a cross-family `in`, which must not
+        # become a 500 on a dual-stack host.
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22"))
+        assert not gate.permits("fd00::5")
+
+    def test_each_family_is_judged_on_its_own_list(self):
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22,fd00::/64"))
+        assert gate.permits("10.2.1.10")
+        assert gate.permits("fd00::1")
+        assert not gate.permits("2001:db8::1")
+
+
+class TestGateClientAddress:
+    def test_the_socket_peer_is_the_answer_without_a_proxy(self):
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22"))
+        assert gate.client_address("10.2.1.10") == "10.2.1.10"
+
+    def test_an_untrusted_peer_cannot_claim_to_be_someone_else(self):
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22"), trusted_proxies=parse_cidrs("127.0.0.0/8"))
+        assert gate.client_address("203.0.113.9", "10.2.1.5") == "203.0.113.9"
+
+    def test_a_trusted_proxy_speaks_for_the_client_and_leftmost_wins(self):
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22"), trusted_proxies=parse_cidrs("127.0.0.0/8"))
+        assert gate.client_address("127.0.0.1", "10.2.1.5") == "10.2.1.5"
+        assert gate.client_address("127.0.0.1", "10.2.1.5, 10.9.9.9") == "10.2.1.5"
+
+    def test_a_broken_header_falls_back_to_the_peer_instead_of_500ing(self):
+        gate = Gate(allow=parse_cidrs("127.0.0.0/8"), trusted_proxies=parse_cidrs("127.0.0.0/8"))
+        assert gate.client_address("127.0.0.1", "unknown") == "127.0.0.1"
+        assert gate.client_address("127.0.0.1", "") == "127.0.0.1"
+
+    def test_check_returns_the_verdict_and_the_address_it_judged(self):
+        gate = Gate(allow=parse_cidrs("10.2.0.0/22"), trusted_proxies=parse_cidrs("127.0.0.0/8"))
+        assert gate.check("127.0.0.1", "10.2.1.5") == (True, "10.2.1.5")
+        assert gate.check("127.0.0.1", "8.8.8.8") == (False, "8.8.8.8")
+
+
+class TestDefaults:
+    def test_loopback_is_the_default_trusted_proxy(self):
+        gate = Gate(
+            allow=parse_cidrs("10.2.0.0/22"),
+            trusted_proxies=parse_cidrs(DEFAULT_TRUSTED_PROXIES),
+        )
+        assert gate.client_address("127.0.0.1", "10.2.1.5") == "10.2.1.5"
+        assert gate.client_address("10.1.1.27", "10.2.1.5") == "10.1.1.27"

tests/test_app.py

@@ -9,8 +9,22 @@ import subprocess
 from pathlib import Path
 
 import pytest
+from fastapi.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
+
+from mirror.access import parse_cidrs
+from mirror.app import create_app
+from mirror.config import Settings
 
 STATIC = Path(__file__).resolve().parent.parent / "src/mirror/static"
+
+
+def gated(*allow, client_address="203.0.113.9", **kwargs):
+    """A client speaking as ``client_address`` to a server restricted to ``allow``."""
+    settings = Settings(allow_cidrs=parse_cidrs(",".join(allow)), **kwargs)
+    return TestClient(create_app(settings), client=(client_address, 44444))
+
+
 EXTERNAL_REF = re.compile(r"""(?:src|href)\s*=\s*["']https?://""", re.IGNORECASE)
 
 
@@ -166,6 +180,99 @@ class TestStylesheet:
         assert offenders == []
 
 
+class TestAddressGate:
+    """mirror has no accounts; the address is the access control."""
+
+    ALLOW = "10.2.0.0/22"
+
+    def test_unset_allows_everyone_which_is_the_development_default(self):
+        with gated(client_address="203.0.113.9") as client:
+            assert client.get("/watch").status_code == 200
+
+    @pytest.mark.parametrize("path", ["/", "/mirror", "/watch", "/api/streams", "/static/app.css"])
+    def test_a_refused_address_gets_no_page_no_roster_no_assets(self, path):
+        with gated(self.ALLOW) as client:
+            assert client.get(path).status_code == 403
+
+    def test_the_probe_stays_answerable_from_anywhere(self):
+        """The container health check comes from loopback, which the list excludes."""
+        with gated(self.ALLOW) as client:
+            response = client.get("/healthz")
+            assert response.status_code == 200
+            assert response.json()["status"] == "ok"
+
+    def test_a_browser_gets_an_explanation_rather_than_json(self):
+        with gated(self.ALLOW) as client:
+            response = client.get("/mirror", headers={"accept": "text/html"})
+            assert response.status_code == 403
+            assert "text/html" in response.headers["content-type"]
+            body = response.text
+            assert self.ALLOW in body
+            assert "203.0.113.9" in body  # which address was judged, or this is undiagnosable
+
+    def test_the_refusal_page_cannot_depend_on_assets_the_client_cannot_fetch(self):
+        with gated(self.ALLOW) as client:
+            body = client.get("/", headers={"accept": "text/html"}).text
+            assert not EXTERNAL_REF.search(body)
+            assert "/static/" not in body
+
+    def test_a_fetch_gets_the_machine_readable_refusal(self):
+        with gated(self.ALLOW) as client:
+            response = client.get("/api/config")
+            assert response.status_code == 403
+            assert response.json() == {
+                "error": "forbidden",
+                "message": "This mirror only accepts connections from 10.2.0.0/22.",
+                "you": "203.0.113.9",
+            }
+
+    def test_a_refusal_is_never_cacheable(self):
+        with gated(self.ALLOW) as client:
+            for headers in ({"accept": "text/html"}, {}):
+                response = client.get("/api/config", headers=headers)
+                assert response.headers["cache-control"] == "no-store"
+
+    def test_the_socket_closes_before_it_accepts(self):
+        with (
+            gated(self.ALLOW) as client,
+            pytest.raises(WebSocketDisconnect),
+            client.websocket_connect("/ws") as socket,
+        ):
+            socket.send_json({"type": "hello", "role": "viewer"})
+
+    def test_the_real_address_is_in_and_connects(self):
+        with gated(self.ALLOW, client_address="10.2.1.10") as client:
+            assert client.get("/watch").status_code == 200
+            with client.websocket_connect("/ws") as socket:
+                assert handshake(socket, "viewer")
+
+    def test_a_trusted_local_proxy_carries_the_client_through(self):
+        """The fleet shape: nginx on the same host, X-Forwarded-For set by nginx."""
+        with gated(self.ALLOW, client_address="127.0.0.1") as client:
+            assert client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"}).status_code == 200
+            with client.websocket_connect("/ws", headers={"x-forwarded-for": "10.2.1.5"}) as socket:
+                assert handshake(socket, "viewer")
+
+    def test_a_trusted_proxy_carrying_a_foreign_client_is_still_refused(self):
+        with gated(self.ALLOW, client_address="127.0.0.1") as client:
+            assert client.get("/watch", headers={"x-forwarded-for": "8.8.8.8"}).status_code == 403
+            with (
+                pytest.raises(WebSocketDisconnect),
+                client.websocket_connect("/ws", headers={"x-forwarded-for": "8.8.8.8"}),
+            ):
+                pass
+
+    def test_an_untrusted_peer_cannot_write_its_own_pass(self):
+        with gated(self.ALLOW, client_address="203.0.113.9") as client:
+            response = client.get("/watch", headers={"x-forwarded-for": "10.2.1.5"})
+            assert response.status_code == 403
+            assert response.json()["you"] == "203.0.113.9"
+
+    def test_both_address_families_can_be_admitted_side_by_side(self):
+        with gated(f"{self.ALLOW}, fd00::/64", client_address="fd00::1234") as client:
+            assert client.get("/watch").status_code == 200
+
+
 class TestJsonApi:
     def test_config_advertises_the_media_defaults(self, client):
         body = client.get("/api/config").json()

tests/test_config.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import pytest
 
 from mirror import __main__ as entry
+from mirror.access import ConfigError, parse_cidrs
 from mirror.config import DEFAULT_PORT, Settings, lan_addresses, parse_ice_servers
 from mirror.debug import maybe_enable_debugpy
 
@@ -30,6 +31,8 @@ class TestSettings:
             "MIRROR_ICE_SERVERS",
             "MIRROR_MAX_BITRATE",
             "MIRROR_MAX_MIRRORS",
+            "MIRROR_ALLOW_CIDRS",
+            "MIRROR_TRUSTED_PROXIES",
         ):
             monkeypatch.delenv(name, raising=False)
         settings = Settings.from_env()
@@ -37,6 +40,13 @@ class TestSettings:
         assert settings.host == "0.0.0.0"
         assert settings.ice_servers == []
         assert settings.max_streams == 0
+        # Unrestricted until told otherwise, and only loopback may speak for a client.
+        assert settings.allow_cidrs == ()
+        assert [str(network) for network in settings.trusted_proxies] == [
+            "127.0.0.0/8",
+            "::1/128",
+        ]
+        assert settings.gate().enabled is False
 
     def test_environment_wins(self, monkeypatch):
         monkeypatch.setenv("MIRROR_HOST", "127.0.0.1")
@@ -44,8 +54,12 @@ class TestSettings:
         monkeypatch.setenv("MIRROR_MAX_BITRATE", "250000")
         monkeypatch.setenv("MIRROR_MAX_MIRRORS", "3")
         monkeypatch.setenv("MIRROR_ICE_SERVERS", '[{"urls": "stun:box:3478"}]')
+        monkeypatch.setenv("MIRROR_ALLOW_CIDRS", "10.2.0.0/22, fd00::/64")
+        monkeypatch.setenv("MIRROR_TRUSTED_PROXIES", "10.1.1.27")
         settings = Settings.from_env()
         assert (settings.host, settings.port) == ("127.0.0.1", 9123)
+        assert [str(network) for network in settings.allow_cidrs] == ["10.2.0.0/22", "fd00::/64"]
+        assert [str(network) for network in settings.trusted_proxies] == ["10.1.1.27/32"]
         assert settings.max_bitrate_bps == 250_000
         assert settings.max_streams == 3
         assert settings.ice_servers == [{"urls": "stun:box:3478"}]
@@ -60,6 +74,19 @@ class TestSettings:
         config = Settings.from_env().client_config()
         assert set(config) == {"iceServers", "maxBitrateBps", "pingIntervalMs"}
 
+    def test_a_bad_cidr_refuses_to_start_rather_than_failing_open(self, monkeypatch):
+        monkeypatch.setenv("MIRROR_ALLOW_CIDRS", "office")
+        with pytest.raises(ConfigError, match="MIRROR_ALLOW_CIDRS"):
+            Settings.from_env()
+
+    def test_the_gate_carries_both_halves_of_the_setting(self):
+        settings = Settings(
+            allow_cidrs=parse_cidrs("10.2.0.0/22"), trusted_proxies=parse_cidrs("127.0.0.1")
+        )
+        gate = settings.gate()
+        assert gate.enabled is True
+        assert gate.check("127.0.0.1", "10.2.1.5") == (True, "10.2.1.5")
+
     def test_lan_addresses_are_real_addresses(self):
         for address in lan_addresses():
             assert not address.startswith("127.")
@@ -74,6 +101,28 @@ class TestCli:
         assert args.command == "serve"
         assert args.port == 8123
 
+    def test_allow_cidr_is_repeatable_and_replaces_the_environment(self, monkeypatch):
+        monkeypatch.setenv("MIRROR_ALLOW_CIDRS", "192.168.0.0/16")
+        args = entry.build_parser().parse_args(
+            ["serve", "--allow-cidr", "10.2.0.0/22", "--allow-cidr", "fd00::/64"]
+        )
+        assert args.allow_cidr == ["10.2.0.0/22", "fd00::/64"]
+        # cmd_serve replaces rather than merges, so the environment cannot leave a
+        # second, wider network open by accident.
+        assert [str(network) for network in parse_cidrs(",".join(args.allow_cidr))] == [
+            "10.2.0.0/22",
+            "fd00::/64",
+        ]
+
+    def test_a_bad_allow_cidr_exits_with_a_message_not_a_traceback(self, capsys):
+        # Config is parsed before anything binds or imports, so this cannot stall
+        # on a stray DEBUGPY in the operator's shell.
+        code = entry.cmd_serve(
+            entry.build_parser().parse_args(["serve", "--allow-cidr", "nonsense"])
+        )
+        assert code == 2
+        assert "--allow-cidr" in capsys.readouterr().err
+
     def test_port_validation_rejects_nonsense(self):
         parser = entry.build_parser()
         with pytest.raises(SystemExit):