ducoterra
7988da2b93
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
121 lines
5.0 KiB
Python
121 lines
5.0 KiB
Python
"""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"
|