diff --git a/SECURITY.md b/SECURITY.md index 12559a7..fd8f8aa 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -41,6 +41,15 @@ network (private, link-local, and VPN addresses), even when ports 80 and 443 are forwarded to this computer for public services. Other IPv4 clients get the connection closed. IPv6 global addresses are not filtered. +The Hub also checks every client itself, before it shows a login page. It runs +as root, so it answers only this computer and the local network (loopback, +private, VPN and link-local addresses) and turns everyone else away, however +they reached it. Global IPv6 addresses are turned away too: a laptop on your +network and a stranger on the internet look the same by address alone. If your +devices use addresses outside the local ranges, list their networks in +`sovran_systemsOS.hub.extraLanNetworks` in `custom.nix`; +`sovran_systemsOS.hub.lanOnly = false` turns the check off. + ### Public services and your home IP address Server + Desktop publishes services under your own domain. The Dynamic DNS diff --git a/app/sovran_systemsos_web/security_helpers.py b/app/sovran_systemsos_web/security_helpers.py index ff46e52..8107d14 100644 --- a/app/sovran_systemsos_web/security_helpers.py +++ b/app/sovran_systemsos_web/security_helpers.py @@ -381,6 +381,83 @@ class LoginThrottle: return len(self._failures) +# ── Local-network client policy ─────────────────────────────────────────────── +# +# The Hub runs as root: it can display stored credentials, reboot the machine +# and rebuild the system. It answers this computer and the local network and +# nobody else. Whether a packet may reach its port is the firewall's and the +# router's business; this is the second lock, applied by the application itself +# so that a port forward, a firewall mistake, or a machine that has a public +# address does not put the login page in front of the internet. +# +# The Hub listens on IPv4 only (see sovran-hub.nix), so IPv6 clients never +# reach it directly and the IPv6 ranges below only matter if that bind is ever +# widened. Global IPv6 addresses (2000::/3) are deliberately not listed: a +# global address belonging to a laptop on the LAN cannot be told apart from a +# stranger's by the address alone, and allowing the range would let the whole +# IPv6 internet through. +LAN_ONLY_IPV4 = ( + "127.0.0.0/8", # this computer + "10.0.0.0/8", + "172.16.0.0/12", + "192.168.0.0/16", + "100.64.0.0/10", # Tailscale and other VPN/CGNAT ranges + "169.254.0.0/16", # link-local +) +LAN_ONLY_IPV6 = ( + "::1/128", + "fc00::/7", # unique-local (covers fd00::/8) + "fe80::/10", # link-local +) + + +class LanPolicy: + """Decides whether a client address counts as local. + + ``extra_networks`` are CIDR blocks an operator has declared local in + addition to the built-in ranges, of either address family. ``enabled=False`` + turns the check off entirely; it is the one explicit way to do that. + """ + + def __init__(self, extra_networks=(), enabled=True): + self.enabled = bool(enabled) + nets = [ipaddress.ip_network(c, strict=False) + for c in LAN_ONLY_IPV4 + LAN_ONLY_IPV6] + for cidr in (extra_networks or ()): + try: + net = ipaddress.ip_network(cidr, strict=False) + except ValueError: + # A malformed entry must never widen the policy. Ignore it and + # stay at the strictest interpretation. + continue + if net.prefixlen == 0: + # 0.0.0.0/0 and ::/0 are "everyone". That is lan_only = false, + # and it should be asked for by name, not arrive as a "network". + continue + nets.append(net) + self._nets = tuple(nets) + + def allows(self, ip): + """Return True if *ip* may reach the service.""" + if not self.enabled: + return True + if not ip: + return False + try: + addr = ipaddress.ip_address(ip) + except ValueError: + return False + # A dual-stack socket reports IPv4 clients as ::ffff:a.b.c.d. The + # address that matters is the IPv4 one inside it. + if addr.version == 6 and addr.ipv4_mapped is not None: + addr = addr.ipv4_mapped + return any(addr.version == n.version and addr in n for n in self._nets) + + @property + def networks(self): + return self._nets + + # ── Persistent Hub session store ───────────────────────────────────────────── def load_session_store(path: str) -> dict[str, float]: diff --git a/app/sovran_systemsos_web/server.py b/app/sovran_systemsos_web/server.py index 5f3a3fd..cff1299 100644 --- a/app/sovran_systemsos_web/server.py +++ b/app/sovran_systemsos_web/server.py @@ -56,6 +56,7 @@ from .security_helpers import ( load_session_store, save_session_store, LoginThrottle, + LanPolicy, LOGIN_FAIL_DELAY, LOGIN_FAIL_MAX_DELAY, LOGIN_FAIL_WINDOW, @@ -821,8 +822,61 @@ class AuthMiddleware(BaseHTTPMiddleware): return await call_next(request) +# ── Local-network middleware ─────────────────────────────────── +# +# The Hub runs as root. Whether a packet may reach its port is up to the +# firewall and the router; this is the second lock, so a port forward or a +# firewall mistake does not put the login page in front of the internet. It +# runs before authentication: a client that is not on the local network never +# sees the login page at all. +# +# Built from the Nix-generated config. lan_only defaults to True, so a Hub built +# without the key still turns off-network clients away rather than failing open. +_hub_cfg = load_config() +_lan_policy = LanPolicy( + enabled=bool(_hub_cfg.get("lan_only", True)), + extra_networks=tuple(_hub_cfg.get("lan_extra_networks") or ()), +) + + +class LanOnlyMiddleware(BaseHTTPMiddleware): + """Refuse clients that are not on this computer or the local network.""" + + # Each refused address is logged once, and the list is capped: a scanner + # must not be able to fill the journal or the process's memory. + _MAX_LOGGED = 256 + + def __init__(self, app, policy): + super().__init__(app) + self._policy = policy + self._logged: set = set() + + async def dispatch(self, request: Request, call_next): + client_ip = request.client.host if request.client else None + if not self._policy.allows(client_ip): + self._note_refusal(client_ip) + return JSONResponse( + {"detail": "Not available from this network"}, status_code=403, + ) + return await call_next(request) + + def _note_refusal(self, client_ip): + if client_ip in self._logged or len(self._logged) >= self._MAX_LOGGED: + return + self._logged.add(client_ip) + logger.warning( + "Refused a Hub request from %r: not this computer or a local " + "network. If that address is yours, add its network to " + "sovran_systemsOS.hub.extraLanNetworks.", + client_ip, + ) + + app.add_middleware(AuthMiddleware) app.add_middleware(NoCacheMiddleware) +# Registered last so it runs outermost: a client that is not on the local +# network is turned away before authentication is considered at all. +app.add_middleware(LanOnlyMiddleware, policy=_lan_policy) _ICONS_DIR = os.environ.get( "SOVRAN_HUB_ICONS", diff --git a/modules/core/roles.nix b/modules/core/roles.nix index 7d2da4b..44f3ce3 100755 --- a/modules/core/roles.nix +++ b/modules/core/roles.nix @@ -61,6 +61,46 @@ sshd = lib.mkEnableOption "SSH remote access"; }; + # ── Hub ─────────────────────────────────────────────────── + hub = { + lanOnly = lib.mkOption { + type = lib.types.bool; + default = true; + description = '' + Refuse Hub requests from clients that are not on this computer or on + the local network: loopback, private (10.0.0.0/8, 172.16.0.0/12, + 192.168.0.0/16), VPN/CGNAT (100.64.0.0/10) and link-local addresses, + plus anything listed in sovran_systemsOS.hub.extraLanNetworks. + + The Hub runs as root and can display stored credentials and reboot + the machine. Whether a packet may reach its port is up to the + firewall and your router; this check is the second lock, so that a + port forward or a firewall mistake does not put the Hub's login page + in front of the internet. + + Set it to false only if this computer sits on a network that hands + out public addresses to your own devices and you would rather not + list them. + ''; + }; + + extraLanNetworks = lib.mkOption { + type = lib.types.listOf lib.types.str; + default = [ ]; + example = [ "203.0.113.0/28" ]; + description = '' + Extra networks, in CIDR notation, that the Hub should treat as local + in addition to the built-in ranges. Needed only if devices on your + local network use addresses outside the private ranges, for example a + public IPv4 block your provider routes onto your LAN. + + Keep each entry as narrow as you can: every address inside it is let + through. To let everything through, set sovran_systemsOS.hub.lanOnly + to false instead; 0.0.0.0/0 and ::/0 are not accepted here. + ''; + }; + }; + # ── Web exposure (controls Caddy vhosts) ────────────────── web = { btcpayserver = lib.mkOption { diff --git a/modules/core/sovran-hub.nix b/modules/core/sovran-hub.nix index 5e9c6d1..2ce41ab 100644 --- a/modules/core/sovran-hub.nix +++ b/modules/core/sovran-hub.nix @@ -115,6 +115,15 @@ let else if cfg.roles.node then "node" else "server_plus_desktop"; + # IPv4 a.b.c.d[/0-32] or IPv6 [/0-128], and never a /0 (that is "everyone"). + octet = "(25[0-5]|2[0-4][0-9]|1[0-9][0-9]|[1-9]?[0-9])"; + lanNetworkOk = p: + ( + builtins.match "${octet}(\\.${octet}){3}(/(3[0-2]|[12]?[0-9]))?" p != null + || builtins.match "[0-9a-fA-F:]*:[0-9a-fA-F:]*(/(12[0-8]|1[01][0-9]|[1-9]?[0-9]))?" p != null + ) + && builtins.match ".*/0" p == null; + generatedConfig = pkgs.writeText "sovran-hub-config.json" (builtins.toJSON { refresh_interval = 5; @@ -122,6 +131,9 @@ let role = activeRole; services = monitoredServices; feature_manager = true; + # Read by LanOnlyMiddleware in server.py. + lan_only = cfg.hub.lanOnly; + lan_extra_networks = cfg.hub.extraLanNetworks; feature_states = { bitcoin-tor-gossip = cfg.features.bitcoin-tor-gossip; }; @@ -503,6 +515,22 @@ in }; config = { + # Catch a typo'd network at build time. The Hub ignores an entry it cannot + # parse (it must never widen its policy by guessing), so without this the + # only symptom would be a client that is refused for no visible reason. + assertions = [ + { + assertion = builtins.all lanNetworkOk cfg.hub.extraLanNetworks; + message = '' + sovran_systemsOS.hub.extraLanNetworks must be a list of IPv4 or IPv6 + networks in CIDR notation, for example [ "203.0.113.0/28" ]. A /0 + prefix is not accepted; set sovran_systemsOS.hub.lanOnly = false to + let every client through. Got: + ${builtins.toJSON cfg.hub.extraLanNetworks} + ''; + } + ]; + systemd.services.sovran-hub-web = { description = "Sovran_SystemsOS Hub Web Interface"; wantedBy = [ "multi-user.target" ]; diff --git a/tests/test_hub_lan_only.py b/tests/test_hub_lan_only.py new file mode 100644 index 0000000..3cb42ff --- /dev/null +++ b/tests/test_hub_lan_only.py @@ -0,0 +1,54 @@ +"""Guards for the Hub checking its own clients. + +The Hub runs as root. Whether a packet may reach its port is up to the firewall +and the router; the application adds a second lock by answering only this +computer and the local network. These read the modules as text, like the other +nix-file checks: nothing is run and nothing touches the network. +""" + +import os +import re +import unittest + +_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) + + +def _read(*parts): + with open(os.path.join(_ROOT, *parts), encoding="utf-8") as f: + return f.read() + + +def _option(src, name): + m = re.search(name + r"\s*=\s*lib\.mkOption\s*\{(.*?)\n \};", src, re.S) + return m.group(1) if m else None + + +class HubChecksItsOwnClients(unittest.TestCase): + + def test_lan_only_option_exists_and_defaults_on(self): + body = _option(_read("modules", "core", "roles.nix"), "lanOnly") + self.assertIsNotNone(body, "hub.lanOnly option not found") + self.assertRegex(body, r"default\s*=\s*true") + + def test_extra_networks_option_exists_and_defaults_empty(self): + body = _option(_read("modules", "core", "roles.nix"), "extraLanNetworks") + self.assertIsNotNone(body, "hub.extraLanNetworks option not found") + self.assertRegex(body, r"default\s*=\s*\[\s*\]") + + def test_policy_is_baked_into_the_generated_config(self): + hub = _read("modules", "core", "sovran-hub.nix") + self.assertRegex(hub, r"lan_only\s*=\s*cfg\.hub\.lanOnly\s*;") + self.assertRegex(hub, r"lan_extra_networks\s*=\s*cfg\.hub\.extraLanNetworks\s*;") + + def test_a_typo_is_caught_at_build_time(self): + hub = _read("modules", "core", "sovran-hub.nix") + self.assertIn("builtins.all lanNetworkOk cfg.hub.extraLanNetworks", hub) + + def test_the_hub_enforces_it_in_its_own_middleware(self): + server = _read("app", "sovran_systemsos_web", "server.py") + self.assertIn("class LanOnlyMiddleware(BaseHTTPMiddleware)", server) + self.assertIn("app.add_middleware(LanOnlyMiddleware", server) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_lan_policy.py b/tests/test_lan_policy.py new file mode 100644 index 0000000..378f8ee --- /dev/null +++ b/tests/test_lan_policy.py @@ -0,0 +1,230 @@ +"""Tests for the Hub's local-network policy and the middleware that enforces it. + +The Hub runs as root, so it answers this computer and the local network and +nobody else: LanPolicy in security_helpers decides, LanOnlyMiddleware in +server.py enforces it before authentication is considered. + +LanPolicy is exercised directly. The middleware is exercised over real HTTP +where the environment allows it; server.py cannot be imported from this repo +(it needs sovran_nwc from the Sovran_Bitcoin flake), so those tests skip rather +than fail, and the wiring is additionally asserted from source so it is always +checked. +""" + +import logging +import os +import sys +import unittest + +_REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) +_APP_PARENT = os.path.join(_REPO_ROOT, "app") +if _APP_PARENT not in sys.path: + sys.path.insert(0, _APP_PARENT) + +from sovran_systemsos_web.security_helpers import ( # noqa: E402 + LanPolicy, + LAN_ONLY_IPV4, + LAN_ONLY_IPV6, +) + +_LOCAL = ( + "127.0.0.1", "10.0.0.1", "172.16.0.1", "172.31.255.254", "192.168.1.10", + "100.64.0.1", "169.254.1.1", + "::1", "fd12:3456::1", "fc00::1", "fe80::1", +) + +_REMOTE = ( + "8.8.8.8", "1.1.1.1", "203.0.113.9", "9.255.255.255", + "172.15.255.255", "172.32.0.1", "192.169.0.1", "100.63.255.255", + # public IPv6 — every address in 2000::/3 is on the internet + "2001:4860:4860::8888", "2606:4700:4700::1111", + "2a00:1450:4001::1", "2400:cb00::1", +) + + +class LanPolicyMatrix(unittest.TestCase): + + def test_local_addresses_are_allowed(self): + policy = LanPolicy() + for address in _LOCAL: + with self.subTest(local=address): + self.assertTrue(policy.allows(address)) + + def test_remote_addresses_are_refused(self): + policy = LanPolicy() + for address in _REMOTE: + with self.subTest(remote=address): + self.assertFalse(policy.allows(address)) + + def test_ipv6_global_is_not_whitelisted(self): + # 2000::/3 is the whole IPv6 global unicast space: allowing it would + # let every public IPv6 address through. + joined = " ".join(LAN_ONLY_IPV4 + LAN_ONLY_IPV6) + self.assertNotIn("2000::/3", joined) + for net in LanPolicy().networks: + if net.version == 6: + with self.subTest(range=str(net)): + self.assertTrue(str(net).startswith(("::1", "fc00", "fe80")), + f"{net} is not a local-only IPv6 range") + + +class LanPolicyConfiguration(unittest.TestCase): + + def test_declared_networks_are_allowed(self): + policy = LanPolicy(extra_networks=["203.0.113.0/28", "2001:db8:abcd::/48"]) + self.assertTrue(policy.allows("203.0.113.9")) + self.assertFalse(policy.allows("203.0.113.16")) + self.assertTrue(policy.allows("2001:db8:abcd::5")) + self.assertFalse(policy.allows("2001:db8:abce::5")) + + def test_a_bare_address_is_a_single_host(self): + policy = LanPolicy(extra_networks=["203.0.113.9"]) + self.assertTrue(policy.allows("203.0.113.9")) + self.assertFalse(policy.allows("203.0.113.10")) + + def test_disabled_allows_everything(self): + policy = LanPolicy(enabled=False) + for address in _REMOTE: + with self.subTest(remote=address): + self.assertTrue(policy.allows(address)) + + def test_malformed_network_does_not_widen_the_policy(self): + # A typo must fail closed, not open the Hub to everything. + policy = LanPolicy(extra_networks=["not-a-network", "203.0.113.0/28"]) + self.assertTrue(policy.allows("203.0.113.9")) + self.assertFalse(policy.allows("8.8.8.8")) + + def test_a_zero_length_prefix_is_not_a_network(self): + # 0.0.0.0/0 and ::/0 mean "everyone". That is lan_only = false and it + # has to be asked for by name rather than arrive as a "network". + policy = LanPolicy(extra_networks=["0.0.0.0/0", "::/0"]) + for address in _REMOTE: + with self.subTest(remote=address): + self.assertFalse(policy.allows(address)) + + def test_missing_or_unparseable_client_is_refused(self): + policy = LanPolicy() + for address in (None, "", "testclient", "not-an-ip"): + with self.subTest(client=address): + self.assertFalse(policy.allows(address)) + + def test_a_dual_stack_socket_does_not_hide_the_ipv4_client(self): + # With an IPv6 listener, IPv4 clients arrive as ::ffff:a.b.c.d. The + # address that counts is the IPv4 one inside it, both ways round. + policy = LanPolicy() + for address in ("::ffff:192.168.1.5", "::ffff:127.0.0.1", "::ffff:10.1.2.3"): + with self.subTest(local=address): + self.assertTrue(policy.allows(address)) + for address in ("::ffff:8.8.8.8", "::ffff:203.0.113.9"): + with self.subTest(remote=address): + self.assertFalse(policy.allows(address)) + + +# ── Middleware ─────────────────────────────────────────────────────────────── + +try: + from fastapi import FastAPI # noqa: E402 + from fastapi.testclient import TestClient # noqa: E402 + from sovran_systemsos_web.server import LanOnlyMiddleware # noqa: E402 + HAVE_MIDDLEWARE = True +except Exception: # fastapi / sovran_nwc unavailable from this repo + HAVE_MIDDLEWARE = False + + +def _app_with(policy): + app = FastAPI() + + @app.get("/ping") + async def ping(): + return {"ok": True} + + app.add_middleware(LanOnlyMiddleware, policy=policy) + return app + + +@unittest.skipUnless(HAVE_MIDDLEWARE, "server.py is not importable here") +class LanOnlyMiddlewareOverHttp(unittest.TestCase): + + def _status(self, policy, client_ip): + client = TestClient(_app_with(policy), client=(client_ip, 51234)) + return client.get("/ping").status_code + + def test_local_client_is_served(self): + for address in ("127.0.0.1", "192.168.1.10", "10.0.0.1"): + with self.subTest(local=address): + self.assertEqual(self._status(LanPolicy(), address), 200) + + def test_remote_client_is_refused(self): + for address in ("203.0.113.9", "8.8.8.8", "2001:4860:4860::8888"): + with self.subTest(remote=address): + self.assertEqual(self._status(LanPolicy(), address), 403) + + def test_refusal_says_nothing_about_the_configuration(self): + # An outsider learns that the answer is no, not why or what to change. + client = TestClient(_app_with(LanPolicy()), client=("203.0.113.9", 51234)) + response = client.get("/ping") + self.assertEqual(response.status_code, 403) + self.assertEqual(response.json(), {"detail": "Not available from this network"}) + + def test_disabled_policy_admits_remote_clients(self): + self.assertEqual(self._status(LanPolicy(enabled=False), "203.0.113.9"), 200) + + def test_a_refused_address_is_logged_once(self): + # The operator whose own device is refused needs to find out why; a + # scanner must not be able to fill the journal. + client = TestClient(_app_with(LanPolicy()), client=("203.0.113.9", 51234)) + with self.assertLogs("sovran_systemsos_web.server", level="WARNING") as seen: + for _ in range(5): + client.get("/ping") + self.assertEqual(len(seen.records), 1) + message = seen.records[0].getMessage() + self.assertIn("203.0.113.9", message) + self.assertIn("sovran_systemsOS.hub.extraLanNetworks", message) + + def test_a_served_client_is_not_logged(self): + records = [] + + class _Collect(logging.Handler): + def emit(self, record): + records.append(record) + + logger = logging.getLogger("sovran_systemsos_web.server") + handler = _Collect(level=logging.WARNING) + logger.addHandler(handler) + try: + client = TestClient(_app_with(LanPolicy()), client=("192.168.1.10", 51234)) + client.get("/ping") + finally: + logger.removeHandler(handler) + self.assertEqual(records, []) + + +# ── Wiring, checked from source so it always runs ───────────────────────────── + +def _server_source(): + with open(os.path.join(_APP_PARENT, "sovran_systemsos_web", "server.py"), + encoding="utf-8") as f: + return f.read() + + +class LanOnlyWiring(unittest.TestCase): + + def test_middleware_is_registered_outermost(self): + # Starlette makes the last-registered middleware the outermost one, so + # an off-network client is turned away before auth is considered. + src = _server_source() + auth = src.index("app.add_middleware(AuthMiddleware)") + nocache = src.index("app.add_middleware(NoCacheMiddleware)") + lan = src.index("app.add_middleware(LanOnlyMiddleware") + self.assertLess(auth, nocache) + self.assertLess(nocache, lan) + + def test_policy_comes_from_the_generated_config(self): + src = _server_source() + self.assertIn("LanPolicy(", src) + self.assertIn('_hub_cfg.get("lan_only", True)', src) + self.assertIn('_hub_cfg.get("lan_extra_networks")', src) + + +if __name__ == "__main__": + unittest.main()