ssh: don't open port 22 for the loopback-only sshd

sshd-localhost.nix turns sshd on for every role so that "ssh root@localhost"
works, and binds it to 127.0.0.1 only ("zero network exposure", as its
own comment says, and as sshd.nix expects: the sshd feature "extends this
to 0.0.0.0 and opens port 22 on the firewall when the user enables remote
SSH").

It never turned off services.openssh.openFirewall, which NixOS defaults
to true and applies whether or not sshd listens on the ports. So port 22
was open in the firewall on every role, Desktop Only included, with
nothing behind it. No listener means no live exposure today; it does mean
the firewall was not saying what the documentation says, and the day
something does bind 22 on a wider address (a listenAddresses change, a
second daemon) it would be reachable without anyone having opened it.

openFirewall is now mkDefault false there. Nothing that wants SSH
published loses it: sshd.nix (feature sshd) and remote-deploy.nix both
add 22 explicitly, and only when enabled. Found by evaluating the real
module set; grepping for allowedTCPPorts does not see a NixOS default.

Evaluated with nix eval, TCP firewall ports:

                            before     after
  Server + Desktop          22 80 443 3051 8937   80 443 3051 8937
  Bitcoin Node Only         22 3051 8937 60847    3051 8937 60847
  Desktop Only              22                    (none)
  Desktop + features.sshd   22                    22
  Desktop + deploy.enable   22 3389               22 3389

and sshd's listen addresses are unchanged: 127.0.0.1 by default,
127.0.0.1 and 0.0.0.0 with the sshd feature. Desktop Only now opens no
TCP port at all; UDP 5353 (mDNS) is the only port open there, and
SECURITY.md says so.

Add tests/test_ssh_exposure.py.

README.md

@@ -316,8 +316,8 @@ the tools of your selected mode already in place.
 Prefer to keep using Windows, macOS, Linux, Android, or iOS? Install
 Sovran_SystemsOS on a separate computer and let it run quietly on your local
 network, with or without a monitor. From any other device on the same network,
-open a browser, visit `http://sovransystemsos.local`, and manage everything
-from [The Sovran Hub](#the-sovran-hub).
+open a browser, visit `http://sovransystemsos.local:8937`, and manage
+everything from [The Sovran Hub](#the-sovran-hub).
 
 Your existing devices stay familiar. Sovran_SystemsOS provides the independent
 infrastructure behind them.
@@ -354,7 +354,7 @@ From one place, the Hub helps you:
           │                      │                      │
     Windows laptop         Phone or tablet          Mac or Linux
           │                      │                      │
-          └──────── Browser: sovransystemsos.local ────┘
+          └─────── Browser: sovransystemsos.local:8937 ─┘
                                  │
                                  ▼
                     ┌──────────────────────────┐
@@ -376,9 +376,10 @@ Keep using the devices you already own. Sovran_SystemsOS becomes the private
 Bitcoin and digital infrastructure behind them.
 
 > **Local access:** the Hub is available at
-> `http://sovransystemsos.local` to devices connected to the same local
-> network. It is protected by authentication and is not automatically exposed
-> to the public internet.
+> `http://sovransystemsos.local:8937` to devices connected to the same local
+> network (not on Desktop, which publishes nothing). It is protected by
+> authentication, answers only your local network, and is not automatically
+> exposed to the public internet.
 
 ---
 
@@ -536,18 +537,20 @@ Open the Hub directly from the Sovran_SystemsOS desktop, or from any other
 device on the same local network at:
 
 ```text
-http://sovransystemsos.local
+http://sovransystemsos.local:8937
 ```
 
-Sign in with your Sovran_SystemsOS credentials.
+Sign in with your Sovran_SystemsOS credentials. Desktop does not publish the Hub
+on the network, so in that mode open it from the desktop.
 
 <details>
-<summary><strong>If sovransystemsos.local does not open</strong></summary>
+<summary><strong>If sovransystemsos.local:8937 does not open</strong></summary>
 
 1. Make sure the Sovran_SystemsOS machine is powered on, and allow it a few
    minutes to finish starting.
 2. Make sure both devices are connected to the same local network, and that
-   you entered the full address `http://sovransystemsos.local`.
+   you entered the full address `http://sovransystemsos.local:8937`,
+   including the `:8937`.
 3. Avoid guest Wi-Fi networks, which may prevent devices from seeing one
    another.
 4. Temporarily disconnect any VPN that may interfere with local-network

SECURITY.md

@@ -35,11 +35,42 @@ external networks and are outside a “fully offline” model.
 The local Hub currently uses HTTP. Authentication does not encrypt local network
 traffic, so use a trusted LAN and avoid public or guest Wi-Fi.
 
-Caddy serves the Hub (`sovransystemsos.local`), Ride The Lightning (port 3051),
-and Mempool (port 60847) only to this computer and to clients on your local
-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 is served on port 8937, on its own: Caddy does not front it. Server +
+Desktop and Bitcoin Node Only open that port in the firewall, so other devices
+on your local network reach the Hub at `http://sovransystemsos.local:8937`.
+Forwarding ports 80 and 443 for public services does not put the Hub in front of
+the internet, because the only thing Caddy answers on those ports is the public
+sites.
+
+On Desktop Only the Hub is not published at all. It is reachable only from the
+machine itself, through the desktop application window on localhost. Desktop
+Only is the role most likely to be used away from home, and a root-capable admin
+UI has no business listening on a coffee-shop network. The firewall there opens
+no TCP port at all; the only port open is UDP 5353, for mDNS.
+`sovran_systemsOS.hub.directPort = true` in `custom.nix` opens port 8937 if you
+do want to reach a Desktop Only Hub from another device.
+
+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. The Hub listens on IPv4 only, so IPv6 clients do not reach it
+at all; if that ever changes, global IPv6 addresses would be turned away,
+because 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.
+
+The check goes by the address a connection comes from. A router that rewrites
+that address when it forwards a port makes an outsider look local, so the check
+is a second lock and not a reason to forward port 8937: don't.
+
+Ride The Lightning (port 3051) and Mempool (port 60847) listen on loopback only,
+and Caddy is how your local network reaches them. Caddy does not filter them by
+client address: forwarding ports 80 and 443 for public services does not reach
+them, because they answer on ports of their own, which nothing asks you to
+forward. Do not forward 3051 or 60847. If you do, Ride The Lightning still asks
+for its own random password and locks out repeated failures, and Mempool shows
+public blockchain data, but neither should face the internet.
 
 ### Public services and your home IP address
 

app/sovran_systemsos_web/security_helpers.py

@@ -15,6 +15,7 @@ import json
 import os
 import re
 import tempfile
+import threading
 import time
 import urllib.parse
 
@@ -245,6 +246,218 @@ def _validate_ssh_pubkey(key: str) -> str:
     return key
 
 
+# ── Login throttling ──────────────────────────────────────────────────────────
+#
+# Delays applied after each failed login, and the lockout that follows once an
+# address has tripped LOGIN_FAIL_MAX inside the window.
+#
+# LOGIN_FAIL_WINDOW has to be longer than the time it takes to reach
+# LOGIN_FAIL_MAX failures under the ramping delay: with a 2s ramp capped at
+# LOGIN_FAIL_MAX_DELAY, 10 attempts take about 80 seconds, so a 60 second
+# window would silently expire the earliest failures and the counter could
+# never reach the limit. 900s (15 minutes) keeps the whole ramp inside it.
+LOGIN_FAIL_DELAY = 2.0         # base delay; the nth failure waits n x this
+LOGIN_FAIL_MAX_DELAY = 10.0    # ceiling for a single delay
+LOGIN_FAIL_WINDOW = 900.0      # rolling window failures are counted in
+LOGIN_FAIL_MAX = 10            # failures in the window that trigger a lockout
+LOGIN_LOCKOUT_SECONDS = 300.0  # how long the lockout lasts
+
+# Cap on how many addresses are tracked, so a distributed sweep cannot grow
+# the table without bound.
+_LOGIN_THROTTLE_MAX_IPS = 4096
+
+
+class LoginThrottle:
+    """Per-address failed-login tracking with a ramping delay and a lockout.
+
+    The delay ramps so a script hammering the login form slows down as it goes,
+    and once LOGIN_FAIL_MAX failures land inside the window the address is
+    refused outright for LOGIN_LOCKOUT_SECONDS. A successful login clears the
+    address so a legitimate user who fumbles a password is not penalised later.
+
+    ``sleep`` and ``clock`` are injectable so tests run without waiting.
+    """
+
+    def __init__(
+        self,
+        fail_delay=LOGIN_FAIL_DELAY,
+        max_delay=LOGIN_FAIL_MAX_DELAY,
+        window=LOGIN_FAIL_WINDOW,
+        max_failures=LOGIN_FAIL_MAX,
+        lockout=LOGIN_LOCKOUT_SECONDS,
+        max_tracked_ips=_LOGIN_THROTTLE_MAX_IPS,
+        sleep=None,
+        clock=None,
+    ):
+        self._fail_delay = float(fail_delay)
+        self._max_delay = float(max_delay)
+        self._window = float(window)
+        self._max_failures = int(max_failures)
+        self._lockout = float(lockout)
+        self._max_tracked_ips = int(max_tracked_ips)
+        self._sleep = sleep if sleep is not None else time.sleep
+        self._clock = clock if clock is not None else time.monotonic
+        self._lock = threading.Lock()
+        self._failures: dict[str, list[float]] = {}
+
+    # ── internals ────────────────────────────────────────────────────────────
+
+    def _prune(self, ip, now):
+        """Drop timestamps outside the window; return what is left."""
+        keep = [t for t in self._failures.get(ip, ()) if now - t < self._window]
+        if keep:
+            self._failures[ip] = keep
+        else:
+            self._failures.pop(ip, None)
+        return keep
+
+    def _evict(self, now):
+        """Forget addresses that can no longer affect anything."""
+        horizon = max(self._window, self._lockout)
+        for ip in [i for i, ts in self._failures.items()
+                   if ts and now - max(ts) > horizon]:
+            self._failures.pop(ip, None)
+        while len(self._failures) > self._max_tracked_ips:
+            oldest = min(self._failures, key=lambda i: max(self._failures[i]))
+            self._failures.pop(oldest, None)
+
+    # ── public API ───────────────────────────────────────────────────────────
+
+    def delay_for(self, count):
+        """Return the delay owed after *count* failures in the current window."""
+        if count <= 0:
+            return 0.0
+        return min(self._fail_delay * count, self._max_delay)
+
+    def failure_count(self, ip):
+        """Return the failures currently counted against *ip*."""
+        with self._lock:
+            return len(self._prune(ip, self._clock()))
+
+    def is_locked_out(self, ip):
+        """Return True while *ip* is inside a lockout."""
+        now = self._clock()
+        with self._lock:
+            failures = self._prune(ip, now)
+            if len(failures) < self._max_failures:
+                return False
+            return (now - failures[-1]) < self._lockout
+
+    def remaining_lockout(self, ip):
+        """Return the seconds left in *ip*'s lockout, or 0.0 if not locked out."""
+        now = self._clock()
+        with self._lock:
+            failures = self._prune(ip, now)
+            if len(failures) < self._max_failures:
+                return 0.0
+            return max(0.0, self._lockout - (now - failures[-1]))
+
+    def record_failure(self, ip):
+        """Record a failure for *ip* and serve out the delay it has earned.
+
+        Returns the delay that was applied. The lock is never held across the
+        sleep, so one slow client cannot stall every other login.
+        """
+        now = self._clock()
+        with self._lock:
+            failures = list(self._prune(ip, now))
+            failures.append(now)
+            self._failures[ip] = failures
+            count = len(failures)
+            self._evict(now)
+        delay = self.delay_for(count)
+        if delay > 0:
+            self._sleep(delay)
+        return delay
+
+    def clear(self, ip):
+        """Forget *ip*, e.g. after a successful login."""
+        with self._lock:
+            self._failures.pop(ip, None)
+
+    def tracked_addresses(self):
+        """Return how many addresses are currently being tracked."""
+        with self._lock:
+            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]:

app/sovran_systemsos_web/server.py

@@ -55,6 +55,13 @@ from .security_helpers import (
     _bech32_convertbits_decode,
     load_session_store,
     save_session_store,
+    LoginThrottle,
+    LanPolicy,
+    LOGIN_FAIL_DELAY,
+    LOGIN_FAIL_MAX_DELAY,
+    LOGIN_FAIL_WINDOW,
+    LOGIN_FAIL_MAX,
+    LOGIN_LOCKOUT_SECONDS,
 )
 from .update_state import effective_update_status
 
@@ -175,11 +182,19 @@ _sessions_lock = Lock()
 _SESSION_PERSIST_MIN_INTERVAL = 30.0  # seconds
 _sessions_last_persist = 0.0
 
-# Failed login tracking: ip → list of failure timestamps
-_login_failures: dict[str, list[float]] = {}
-LOGIN_FAIL_DELAY  = 2.0   # seconds to sleep after a failed attempt
-LOGIN_FAIL_WINDOW = 60.0  # rolling window (seconds) for counting failures
-LOGIN_FAIL_MAX    = 10    # max failures in window before extra delay
+# Failed login tracking.
+#
+# LOGIN_FAIL_MAX used to be declared here and never read anywhere: the only
+# thing a failed attempt cost an attacker was a flat 2 second delay, and there
+# was no lockout, no escalation and no ban. The throttling now lives in
+# security_helpers.LoginThrottle, which ramps the delay and refuses an address
+# outright once it has tripped LOGIN_FAIL_MAX inside the window.
+#
+# The window moved from 60s to 900s. With the ramping delay, reaching
+# LOGIN_FAIL_MAX takes about 80 seconds, so a 60 second window expired the
+# earliest failures before the limit could ever be reached — the old constant
+# could not have worked even if it had been wired up.
+_login_throttle = LoginThrottle()
 
 # Public paths that are accessible without a valid session
 _AUTH_EXEMPT_PATHS = {"/login", "/api/login", "/auto-login", "/api/ping"}
@@ -770,19 +785,20 @@ def _ensure_onboarding_reopened_for_migration() -> None:
         logger.warning("Could not clear onboarding flag for migration flow: %s", exc)
 
 
-def _record_failure(client_ip: str) -> None:
-    """Record a failed login attempt and apply a rate-limit delay.
+def _record_failure(client_ip: str) -> float:
+    """Record a failed login attempt and apply the throttling delay.
 
     Must always be called via loop.run_in_executor() so that the blocking
     time.sleep() does not stall the asyncio event loop.
+
+    Returns the delay that was applied.
     """
-    now = time.time()
-    failures = _login_failures.setdefault(client_ip, [])
-    # Prune old entries outside the window
-    _login_failures[client_ip] = [t for t in failures if now - t < LOGIN_FAIL_WINDOW]
-    _login_failures[client_ip].append(now)
-    # Sleep in the thread-pool thread to slow brute-force without blocking the loop
-    time.sleep(LOGIN_FAIL_DELAY)
+    return _login_throttle.record_failure(client_ip)
+
+
+def _is_locked_out(client_ip: str) -> bool:
+    """Return True while *client_ip* is inside a lockout."""
+    return _login_throttle.is_locked_out(client_ip)
 
 
 # ── Authentication middleware ─────────────────────────────────────
@@ -806,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",
@@ -2638,10 +2707,24 @@ async def api_login(req: LoginRequest, request: Request):
     """Validate the Hub password and issue a session cookie."""
     client_ip = request.client.host if request.client else "unknown"
     loop = asyncio.get_event_loop()
+
+    # Refuse outright while the address is locked out. This runs before the
+    # scrypt hash, so a locked-out client costs almost nothing to reject.
+    if _is_locked_out(client_ip):
+        remaining = int(_login_throttle.remaining_lockout(client_ip) // 60) + 1
+        raise HTTPException(
+            status_code=429,
+            detail=f"Too many failed attempts. Try again in about {remaining} minute(s).",
+        )
+
     ok = await loop.run_in_executor(None, _check_password, req.password)
     if not ok:
         await loop.run_in_executor(None, _record_failure, client_ip)
         raise HTTPException(status_code=401, detail="Incorrect password")
+
+    # A real login clears the address, so fumbling a password once in a while
+    # does not accumulate towards a lockout.
+    _login_throttle.clear(client_ip)
     token = _create_session()
     response = JSONResponse({"ok": True})
     response.set_cookie(

app/sovran_systemsos_web/templates/index.html

@@ -264,7 +264,7 @@
                   <h3>Network</h3>
                   <div class="net-row"><span class="net-k">LAN</span><span class="ip-value" id="ip-internal">…</span></div>
                   <div class="net-row"><span class="net-k">WAN</span><span class="ip-value" id="ip-external">…</span></div>
-                  <div class="sub">sovransystemsos.local</div>
+                  <div class="sub">sovransystemsos.local:8937</div>
                 </div>
               </div>
               <div class="welcome-dyn" id="wc-more"></div>
@@ -520,7 +520,7 @@
       <div class="security-reset-password-box" id="security-reset-new-password">&nbsp;</div>
       <p class="security-reset-password-warning">
         ✍️ <strong>Write this down now.</strong><br />
-        You will need it to log in to your computer<br />and the Sovran Hub at <em>sovransystemsos.local</em>.
+        You will need it to log in to your computer<br />and the Sovran Hub at <em>sovransystemsos.local:8937</em>.
       </p>
       <button class="security-reset-reboot-btn" id="security-reset-reboot-btn" disabled>
         I have written down my new password — Restart Entire System

iso/installer.py

@@ -54,9 +54,14 @@ DICEWARE_WORDS = [
 ]
 
 def generate_diceware_password():
-    words = [secrets.choice(DICEWARE_WORDS) for _ in range(3)]
-    digit = secrets.randbelow(10)
-    return "-".join(words) + f"-{digit}"
+    # 4 words from a 96 word list plus 2 digits: 96^4 x 100 = ~8.5e9, about
+    # 33 bits. The old 3 words plus 1 digit was 96^3 x 10 = ~8.8e6, about 23
+    # bits, for a password that is simultaneously the desktop login, the
+    # 'free' account password and the only thing in front of a Hub that runs
+    # as root and hands out every stored credential.
+    words = [secrets.choice(DICEWARE_WORDS) for _ in range(4)]
+    digits = f"{secrets.randbelow(100):02d}"
+    return "-".join(words) + f"-{digits}"
 
 try:
     logfile = open(LOG, "a")

modules/core/caddy.nix

@@ -14,13 +14,54 @@ let
     || config.sovran_systemsOS.features.haven
     || config.sovran_systemsOS.features."nwc-wallets"
     || config.sovran_systemsOS.features.element-calling;
+
+  # RTL and Mempool listen on loopback only: Sovran_Bitcoin binds them to
+  # 127.0.0.1, and RTL's unit is sandboxed to loopback besides. Caddy is how
+  # the local network reaches them (:3051 and :60847), so it has to run
+  # wherever they do. That includes Bitcoin Node Only, which has no
+  # domain-based service and so no other reason to run Caddy.
+  #
+  # The Hub is not one of these. It listens on 0.0.0.0:8937 itself, so it is
+  # served on its own port rather than through Caddy: the one service that
+  # runs as root has nothing in front of it that it does not need, and the
+  # public sites on ports 80/443 cannot be asked for it by Host header.
+  servesRtl = config.sovran_systemsOS.services.bitcoin;
+  servesMempool = servesRtl && config.sovran_systemsOS.features.mempool;
+
+  caddyEnabled = needsHttpsPorts || extraVhosts != "" || servesRtl;
+
+  # Sites for the local network, one per loopback-only service. Written after
+  # the public domain sites; each exists only where its service does.
+  #
+  # They do not filter by client address. A request can only reach them on
+  # their own ports, which no setup step asks you to forward, so forwarding
+  # 80/443 for public services does not expose them (a Host header on those
+  # ports cannot select a site that listens elsewhere). RTL has its own
+  # password and lockout, and Mempool shows public chain data. An address
+  # check here could not be made right for IPv6 anyway: a laptop's global
+  # address on the LAN looks exactly like a stranger's.
+  bitcoinUiSites =
+    lib.optionalString servesRtl ''
+
+:3051 {
+  reverse_proxy :3050
+  encode gzip zstd
+}
+''
+    + lib.optionalString servesMempool ''
+
+:60847 {
+  reverse_proxy :60845
+  encode gzip zstd
+}
+'';
 in
 {
   services.caddy = {
-    # Only enable Caddy when at least one domain-based service needs it or
-    # the operator has defined custom vhosts.  This prevents Caddy from
-    # running on Desktop Only installs that have no web services configured.
-    enable = needsHttpsPorts || extraVhosts != "";
+    # Caddy runs when a domain-based service needs it, when the operator has
+    # defined custom vhosts, or when it is the way to reach RTL and Mempool.
+    # Desktop Only has none of those, so Caddy stays off there.
+    enable = caddyEnabled;
     user = "caddy";
     group = "root";
   };
@@ -89,26 +130,6 @@ EOF
 EOF
       ''}
 
-      # ── LAN-only guard ──────────────────────────────
-      # The Hub, RTL and Mempool sites below are meant for this home network
-      # only. Forwarding ports 80/443 on the router also lets other clients
-      # reach Caddy, so these sites check where a request comes from, not just
-      # which Host it asks for. Anyone else gets the connection closed.
-      #   private_ranges   10/8, 172.16/12, 192.168/16, 127/8, fd00::/8, ::1
-      #   100.64.0.0/10    Tailscale and other VPN addresses
-      #   169.254.0.0/16, fe80::/10, fc00::/7   link-local and unique-local
-      #   2000::/3         IPv6 global addresses. Computers on this network
-      #                    often connect over their own global address, which
-      #                    looks the same as one from the internet, so IPv6
-      #                    global addresses are not filtered.
-      cat >> /run/caddy/Caddyfile <<'EOF'
-
-(sovran_lan_only) {
-  @outside not remote_ip private_ranges 100.64.0.0/10 169.254.0.0/16 fe80::/10 fc00::/7 2000::/3
-  abort @outside
-}
-EOF
-
       # ── Matrix ──────────────────────────────────────
       if [ -n "$MATRIX" ]; then
         if [ -f /run/caddy/element-calling.snippet ]; then
@@ -222,40 +243,11 @@ $LIGHTNING {
 EOF
       fi
 
-      # ── Sovran Hub (LAN access via mDNS) ────────────
-      cat >> /run/caddy/Caddyfile <<EOF
-
-http://sovransystemsos.local {
-  import sovran_lan_only
-  reverse_proxy localhost:8937
-  header {
-    Clear-Site-Data "\"cache\""
-    Cache-Control "no-store, no-cache, must-revalidate, max-age=0"
-    Pragma "no-cache"
-    Expires "0"
-  }
-}
-EOF
-
-      # ── RTL (LAN access) ────────────────────────────
-      cat >> /run/caddy/Caddyfile <<EOF
-
-:3051 {
-  import sovran_lan_only
-  reverse_proxy :3050
-  encode gzip zstd
-}
-EOF
-
-      # ── Mempool (LAN access) ────────────────────────
-      cat >> /run/caddy/Caddyfile <<EOF
-
-:60847 {
-  import sovran_lan_only
-  reverse_proxy :60845
-  encode gzip zstd
-}
-EOF
+      # ── RTL and Mempool (local network) ─────────────
+      # Only where those services run; see bitcoinUiSites above.
+      cat >> /run/caddy/Caddyfile <<'LAN_SITES_EOF'
+${bitcoinUiSites}
+LAN_SITES_EOF
 
       # ── Custom vhosts from custom.nix ──────────────
       cat >> /run/caddy/Caddyfile <<'CUSTOM_VHOSTS_EOF'

modules/core/roles.nix

@@ -61,6 +61,67 @@
       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.
+        '';
+      };
+
+      directPort = lib.mkOption {
+        type = lib.types.bool;
+        default = !config.sovran_systemsOS.roles.desktop;
+        defaultText = lib.literalExpression "!config.sovran_systemsOS.roles.desktop";
+        description = ''
+          Open port 8937 on the firewall, so that other devices on the local
+          network can reach the Hub at http://sovransystemsos.local:8937.
+
+          On by default for Server + Desktop and Bitcoin Node Only. Off on
+          Desktop Only, the role most likely to be used away from home: there
+          nothing is published, and the Hub is reachable only from this
+          computer, through the desktop application window on localhost. Set it
+          to true in custom.nix if you do want to reach a Desktop Only Hub from
+          another device.
+
+          The Hub runs as root, so it checks every client itself (see
+          sovran_systemsOS.hub.lanOnly); the firewall opening only decides
+          whether a packet may reach it at all.
+        '';
+      };
+
+      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 {

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;
       };
@@ -475,6 +487,12 @@ os.environ["SOVRAN_HUB_ICONS"]  = os.path.join("$out", "share", "sovran-hub", "i
 import uvicorn
 uvicorn.run(
     "sovran_systemsos_web.server:app",
+    # IPv4 only, on purpose. The desktop launcher uses "localhost", which
+    # falls back to 127.0.0.1, and other devices reach the Hub over IPv4 too.
+    # An IPv6 listener would admit clients whose global addresses the Hub's own
+    # check cannot tell from a stranger's (see LanPolicy). Which devices may
+    # connect is up to the firewall (hub.directPort) and that check
+    # (hub.lanOnly), not this bind.
     host="0.0.0.0",
     port=8937,
     log_level="info",
@@ -503,6 +521,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" ];
@@ -571,7 +605,14 @@ in
 
     environment.systemPackages = [ sovran-hub-web ];
 
-    networking.firewall.allowedTCPPorts = [ 8937 60847 ];
+    # The Hub is served on its own port, not through Caddy (see caddy.nix).
+    # Nothing here filters by client address: that is the Hub's own check
+    # (sovran_systemsOS.hub.lanOnly), and which networks can route to this
+    # computer at all is the router's call.
+    # 60847 is where Caddy serves Mempool, so it is open only when Mempool is.
+    networking.firewall.allowedTCPPorts =
+      lib.optionals cfg.hub.directPort [ 8937 ]
+      ++ lib.optionals (cfg.services.bitcoin && cfg.features.mempool) [ 60847 ];
 
     # ── Auto-launch Hub in browser on login ───────────────────────
     environment.etc."xdg/autostart/sovran-hub-autolaunch.desktop".text = ''

modules/core/sshd-localhost.nix

@@ -9,6 +9,13 @@
 
   services.openssh = {
     enable = true;
+    # sshd listens on 127.0.0.1 only here, so there is nothing for the firewall
+    # to let in. NixOS opens sshd's ports by default (openFirewall = true)
+    # whether or not sshd listens on them, which left port 22 open on every
+    # role, Desktop Only included. The roles that do publish SSH open it
+    # themselves: the sshd feature (sshd.nix) and remote deploy
+    # (remote-deploy.nix) both add 22 explicitly.
+    openFirewall = lib.mkDefault false;
     listenAddresses = lib.mkDefault [
       { addr = "127.0.0.1"; port = 22; }
     ];

modules/credentials.nix

@@ -91,7 +91,7 @@ in
       SECRET_FILE="/var/lib/secrets/root-password"
       if [ ! -f "$SECRET_FILE" ]; then
         mkdir -p /var/lib/secrets
-        # Generate a diceware-style passphrase: word-word-word-N
+        # Generate a diceware-style passphrase: word-word-word-word-NN
         WORDS="apple barn brook cabin cedar cloud coral crane delta eagle ember \
                fern field flame flora flint frost grove haven hedge holly heron \
                jade juniper kelp larch lemon lilac linden loch lotus maple marsh \
@@ -106,8 +106,9 @@ in
         W1=''${WORD_ARRAY[$((RANDOM % COUNT))]}
         W2=''${WORD_ARRAY[$((RANDOM % COUNT))]}
         W3=''${WORD_ARRAY[$((RANDOM % COUNT))]}
-        DIGIT=$((RANDOM % 10))
-        ROOT_PASS="$W1-$W2-$W3-$DIGIT"
+        W4=''${WORD_ARRAY[$((RANDOM % COUNT))]}
+        DIGIT=$(printf '%02d' $((RANDOM % 100)))
+        ROOT_PASS="$W1-$W2-$W3-$W4-$DIGIT"
         echo "$ROOT_PASS" > "$SECRET_FILE"
         chmod 600 "$SECRET_FILE"
       fi
@@ -170,7 +171,7 @@ in
       fi
 
       mkdir -p /var/lib/secrets
-      # Generate a diceware-style passphrase: word-word-word-N
+      # Generate a diceware-style passphrase: word-word-word-word-NN
       WORDS="apple barn brook cabin cedar cloud coral crane delta eagle ember \
              fern field flame flora flint frost grove haven hedge holly heron \
              jade juniper kelp larch lemon lilac linden loch lotus maple marsh \
@@ -185,8 +186,9 @@ in
       W1=''${WORD_ARRAY[$((RANDOM % COUNT))]}
       W2=''${WORD_ARRAY[$((RANDOM % COUNT))]}
       W3=''${WORD_ARRAY[$((RANDOM % COUNT))]}
-      DIGIT=$((RANDOM % 10))
-      FREE_PASS="$W1-$W2-$W3-$DIGIT"
+      W4=''${WORD_ARRAY[$((RANDOM % COUNT))]}
+      DIGIT=$(printf '%02d' $((RANDOM % 100)))
+      FREE_PASS="$W1-$W2-$W3-$W4-$DIGIT"
       echo "$FREE_PASS" > "$SECRET_FILE"
       chmod 600 "$SECRET_FILE"
       echo "free:$FREE_PASS" | chpasswd
@@ -229,8 +231,9 @@ in
       W1=''${WORD_ARRAY[$((RANDOM % COUNT))]}
       W2=''${WORD_ARRAY[$((RANDOM % COUNT))]}
       W3=''${WORD_ARRAY[$((RANDOM % COUNT))]}
-      DIGIT=$((RANDOM % 10))
-      FREE_PASS="$W1-$W2-$W3-$DIGIT"
+      W4=''${WORD_ARRAY[$((RANDOM % COUNT))]}
+      DIGIT=$(printf '%02d' $((RANDOM % 100)))
+      FREE_PASS="$W1-$W2-$W3-$W4-$DIGIT"
 
       printf '%s\n' "$FREE_PASS" > "$SECRET_FILE"
       chmod 600 "$SECRET_FILE"

modules/sovran-bitcoin-integration.nix

@@ -105,9 +105,11 @@ in {
     '';
   };
 
-  # ── 5. Firewall — Hub management port ──────────────────────────
+  # ── 5. Firewall — RTL ──────────────────────────────────────────
+  # RTL is a web app served by Caddy over TCP on 3051. The matching UDP rule
+  # that used to sit here was carried over from the TCP line and opened a port
+  # nothing listens on.
   networking.firewall.allowedTCPPorts = lib.mkIf cfg.services.bitcoin [ 3051 ];
-  networking.firewall.allowedUDPPorts = lib.mkIf cfg.services.bitcoin [ 3051 ];
 
   # ── 6. NWC / LNURL — Sovran Hub integration ───────────────────
   # Sovran_Bitcoin's albyhub.nix and lnurl.nix handle the base services.

tests/test_caddy_lan_only.py

@@ -1,110 +0,0 @@
-"""Guards for the LAN-only Caddy sites.
-
-The Hub (http://sovransystemsos.local), Ride The Lightning (:3051) and Mempool
-(:60847) sites are for the home network. Caddy has to check where a request comes
-from because forwarding ports 80/443 on the router lets other clients reach it too.
-The domain sites for the operator's own public services must stay public.
-
-These read modules/core/caddy.nix like the nix-file checks in test_security.py:
-nothing is run and nothing touches the network.
-"""
-
-import ipaddress
-import os
-import re
-import unittest
-
-_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
-
-# What Caddy's "private_ranges" shortcut expands to (see the remote_ip matcher docs).
-_PRIVATE_RANGES = ["192.168.0.0/16", "172.16.0.0/12", "10.0.0.0/8",
-                   "127.0.0.1/8", "fd00::/8", "::1"]
-
-_LAN_SITES = ("http://sovransystemsos.local", ":3051", ":60847")
-
-
-def _read(*parts):
-    with open(os.path.join(_ROOT, *parts), encoding="utf-8") as f:
-        return f.read()
-
-
-def _site(src, address):
-    m = re.search(r"^" + re.escape(address) + r" \{\n(.*?)^\}$", src, re.S | re.M)
-    return m.group(1) if m else None
-
-
-def _snippet(src):
-    m = re.search(r"^\(sovran_lan_only\) \{\n(.*?)^\}$", src, re.S | re.M)
-    return m.group(1) if m else None
-
-
-def _allowed_networks(snippet):
-    m = re.search(r"^\s*@outside not remote_ip (.+)$", snippet, re.M)
-    assert m, "the @outside matcher is missing"
-    nets = []
-    for token in m.group(1).split():
-        for cidr in (_PRIVATE_RANGES if token == "private_ranges" else [token]):
-            nets.append(ipaddress.ip_network(cidr, strict=False))
-    return nets
-
-
-def _is_allowed(nets, address):
-    ip = ipaddress.ip_address(address)
-    return any(ip.version == n.version and ip in n for n in nets)
-
-
-class LanOnlySites(unittest.TestCase):
-
-    @classmethod
-    def setUpClass(cls):
-        cls.src = _read("modules", "core", "caddy.nix")
-
-    def test_lan_sites_use_the_guard_before_they_proxy(self):
-        for address in _LAN_SITES:
-            with self.subTest(site=address):
-                body = _site(self.src, address)
-                self.assertIsNotNone(body, f"{address} site not found")
-                self.assertIn("import sovran_lan_only", body)
-                self.assertLess(body.index("import sovran_lan_only"),
-                                body.index("reverse_proxy"))
-
-    def test_only_the_lan_sites_use_the_guard(self):
-        self.assertEqual(self.src.count("import sovran_lan_only"), len(_LAN_SITES))
-        public = re.findall(r"^\$[A-Z]+ \{\n(.*?)^\}$", self.src, re.S | re.M)
-        self.assertGreaterEqual(len(public), 6, "domain sites not found")
-        for body in public:
-            self.assertNotIn("sovran_lan_only", body)
-        # the Matrix site that Element calling writes instead of the plain one
-        self.assertNotIn("sovran_lan_only", _read("modules", "element-calling.nix"))
-
-    def test_guard_closes_the_connection_for_everyone_else(self):
-        snippet = _snippet(self.src)
-        self.assertIsNotNone(snippet, "(sovran_lan_only) snippet not found")
-        self.assertRegex(snippet, r"(?m)^\s*abort @outside\s*$")
-        # defined before the first site that imports it
-        self.assertLess(self.src.index("(sovran_lan_only) {"),
-                        self.src.index("import sovran_lan_only"))
-
-    def test_guard_ranges(self):
-        nets = _allowed_networks(_snippet(self.src))
-        for address in ("127.0.0.1", "::1", "10.0.0.1", "172.16.0.1", "172.31.255.254",
-                        "192.168.1.10", "100.64.0.1", "100.127.255.254", "169.254.1.1",
-                        "fd12:3456::1", "fe80::1", "fc00::1"):
-            with self.subTest(allowed=address):
-                self.assertTrue(_is_allowed(nets, address))
-        for address in ("8.8.8.8", "1.1.1.1", "203.0.113.9", "9.255.255.255", "11.0.0.1",
-                        "172.15.255.255", "172.32.0.1", "192.169.0.1", "100.63.255.255",
-                        "100.128.0.1", "169.253.255.255"):
-            with self.subTest(refused=address):
-                self.assertFalse(_is_allowed(nets, address))
-
-    def test_ipv6_global_addresses_are_not_filtered(self):
-        # Computers on the home network often connect over their own global IPv6
-        # address, which cannot be told apart from the internet's by address alone.
-        # If this is ever tightened, LAN clients on IPv6 networks lose the Hub.
-        nets = _allowed_networks(_snippet(self.src))
-        self.assertTrue(_is_allowed(nets, "2001:db8::5"))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_hub_direct.py

@@ -0,0 +1,193 @@
+"""Guards for serving the Hub on its own port instead of through Caddy.
+
+The Hub is the one service that runs as root. It listens on 0.0.0.0:8937
+itself, so Caddy adds nothing it needs: not TLS (the site was plain http), not
+authentication, not cache headers (the app sets its own). What it did add was a
+second door: with ports 80/443 forwarded for public services, a Host header on
+those ports reached the Hub. The Hub is therefore served on port 8937 only, and
+Caddy keeps the two services it is actually needed for, because they listen on
+loopback only: Ride The Lightning (:3051) and Mempool (:60847).
+
+Like the other nix-file checks these read the modules as text: 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 _without_comments(src):
+    """The Nix source with `#` comment lines removed."""
+    return "\n".join(l for l in src.splitlines() if not l.lstrip().startswith("#"))
+
+
+def _binding(src, name):
+    """The right-hand side of a top-level `name = ...;` binding in a let/attrset."""
+    m = re.search(r"^\s*" + re.escape(name) + r"\s*=\s*(?P<v>.*?);\s*$", src, re.M | re.S)
+    assert m, f"{name} not found"
+    return m.group("v")
+
+
+class HubIsNotACaddySite(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.caddy = _read("modules", "core", "caddy.nix")
+        cls.code = _without_comments(cls.caddy)
+
+    def test_there_is_no_site_for_the_hub(self):
+        self.assertNotRegex(self.code, r"sovransystemsos\.local")
+        self.assertNotIn("8937", self.code)
+
+    def test_the_public_ports_still_belong_to_the_public_sites(self):
+        # Caddy now also runs to bridge RTL and Mempool (even on Node Only),
+        # which must not open 80/443 by itself: those follow the domain-based
+        # services and nothing else.
+        self.assertRegex(
+            self.code,
+            r"networking\.firewall\.allowedTCPPorts\s*=\s*lib\.mkIf\s+needsHttpsPorts\s*\[\s*80\s+443\s*\]",
+        )
+
+
+class CaddyDoesNoAddressFiltering(unittest.TestCase):
+    """Caddy is a bridge for RTL and Mempool and a TLS front for public sites.
+
+    It used to carry a client-address guard (sovran_lan_only). That guard was
+    never aimed at these two sites: the bug was a Host header on ports 80/443
+    reaching the Hub, and RTL and Mempool sit on ports of their own. It also
+    could not be made right for IPv6, where a laptop's global address on the
+    LAN is indistinguishable from a stranger's.
+    """
+
+    @classmethod
+    def setUpClass(cls):
+        cls.caddy = _read("modules", "core", "caddy.nix")
+        cls.code = _without_comments(cls.caddy)
+
+    def test_there_is_no_address_filter(self):
+        # (private_ranges is deliberately not on this list: the Nextcloud site
+        # uses it for trusted_proxies, which is not a filter on who may connect.)
+        for needle in ("sovran_lan_only", "remote_ip", "abort @"):
+            with self.subTest(needle=needle):
+                self.assertNotIn(needle, self.code)
+
+    def test_the_bitcoin_sites_are_plain_proxies(self):
+        for site, upstream in ((":3051", ":3050"), (":60847", ":60845")):
+            with self.subTest(site=site):
+                m = re.search(r"^" + re.escape(site) + r" \{\n(.*?)^\}$", self.code, re.S | re.M)
+                self.assertIsNotNone(m, f"{site} site not found")
+                directives = [l.strip() for l in m.group(1).splitlines() if l.strip()]
+                self.assertEqual(directives, [f"reverse_proxy {upstream}", "encode gzip zstd"])
+
+    def test_the_options_for_a_declared_prefix_are_gone(self):
+        # Never needed once Caddy stops guessing: neither the option nor its
+        # build-time assertion may linger half-wired.
+        roles = _read("modules", "core", "roles.nix")
+        self.assertNotIn("lanIPv6Prefixes", roles)
+        self.assertNotIn("lanIPv6Prefixes", self.caddy)
+
+
+class CaddyRunsWhereItIsNeeded(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.caddy = _read("modules", "core", "caddy.nix")
+        cls.code = _without_comments(cls.caddy)
+
+    def test_it_runs_for_domains_vhosts_or_the_bitcoin_uis(self):
+        self.assertRegex(
+            self.code,
+            r"caddyEnabled\s*=\s*needsHttpsPorts\s*\|\|\s*extraVhosts\s*!=\s*\"\"\s*\|\|\s*servesRtl\s*;",
+        )
+        self.assertRegex(self.code, r"enable\s*=\s*caddyEnabled\s*;")
+
+    def test_rtl_and_mempool_follow_their_services(self):
+        self.assertRegex(self.code,
+                         r"servesRtl\s*=\s*config\.sovran_systemsOS\.services\.bitcoin\s*;")
+        self.assertRegex(
+            self.code,
+            r"servesMempool\s*=\s*servesRtl\s*&&\s*config\.sovran_systemsOS\.features\.mempool\s*;",
+        )
+
+    def test_each_site_exists_only_where_its_service_does(self):
+        self.assertRegex(self.code, r"lib\.optionalString\s+servesRtl\s*''\s*\n+:3051 \{")
+        self.assertRegex(self.code, r"lib\.optionalString\s+servesMempool\s*''\s*\n+:60847 \{")
+        # ... and they are written into the Caddyfile from that one place
+        self.assertIn("${bitcoinUiSites}", self.caddy)
+
+    def test_rtl_and_mempool_still_proxy_to_their_loopback_ports(self):
+        self.assertRegex(self.code, r":3051 \{[^}]*reverse_proxy :3050")
+        self.assertRegex(self.code, r":60847 \{[^}]*reverse_proxy :60845")
+
+
+class HubPortExposure(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        cls.hub = _read("modules", "core", "sovran-hub.nix")
+        cls.roles = _read("modules", "core", "roles.nix")
+
+    def _firewall_value(self):
+        m = re.search(
+            r"^    networking\.firewall\.allowedTCPPorts =\s*(?P<value>.*?);\s*$",
+            self.hub, re.S | re.M,
+        )
+        self.assertIsNotNone(m, "networking.firewall.allowedTCPPorts not found")
+        return m.group("value")
+
+    def test_the_hub_port_is_never_opened_unconditionally(self):
+        # Regression: this used to be `allowedTCPPorts = [ 8937 60847 ]` with
+        # no mkIf and no option gate, on every role, Desktop Only included.
+        value = self._firewall_value()
+        self.assertNotRegex(value, r"^\s*\[")
+        self.assertRegex(value, r"lib\.optionals\s+cfg\.hub\.directPort\s+\[\s*8937\s*\]")
+
+    def test_the_mempool_port_follows_mempool(self):
+        self.assertRegex(
+            self._firewall_value(),
+            r"lib\.optionals\s+\(cfg\.services\.bitcoin\s*&&\s*cfg\.features\.mempool\)\s+\[\s*60847\s*\]",
+        )
+
+    def test_nothing_else_is_opened_here(self):
+        ports = re.findall(r"\[\s*(\d+)\s*\]", self._firewall_value())
+        self.assertEqual(sorted(ports), ["60847", "8937"])
+
+    def test_direct_port_is_on_for_the_server_roles_and_off_for_desktop_only(self):
+        m = re.search(r"directPort\s*=\s*lib\.mkOption\s*\{(.*?)\n      \};", self.roles, re.S)
+        self.assertIsNotNone(m, "hub.directPort option not found")
+        self.assertRegex(m.group(1), r"default\s*=\s*!config\.sovran_systemsOS\.roles\.desktop\s*;")
+
+    def test_the_bind_is_ipv4_only_on_purpose(self):
+        # IPv6 clients cannot reach the Hub, so the question of which IPv6
+        # addresses are "local" never comes up. Widening the bind reopens it.
+        self.assertIn('host="0.0.0.0"', self.hub)
+        self.assertNotRegex(self.hub, r'host="::"')
+        self.assertNotRegex(self.hub, r"both IPv4 and IPv6")
+
+
+class TheHubIsDocumentedAtItsPort(unittest.TestCase):
+
+    def test_no_document_still_sends_people_to_port_80(self):
+        for name in (("README.md",), ("SECURITY.md",),
+                     ("app", "sovran_systemsos_web", "templates", "index.html")):
+            with self.subTest(file=name[-1]):
+                text = _read(*name)
+                self.assertNotRegex(text, r"sovransystemsos\.local(?!:8937)(?![a-z])",
+                                    f"{name[-1]} sends people to sovransystemsos.local without :8937")
+
+    def test_the_documents_name_the_port(self):
+        self.assertIn("http://sovransystemsos.local:8937", _read("README.md"))
+        self.assertIn("http://sovransystemsos.local:8937", _read("SECURITY.md"))
+        self.assertIn("hub.directPort", _read("SECURITY.md"))
+
+
+if __name__ == "__main__":
+    unittest.main()

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()

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()

tests/test_login_throttle.py

@@ -0,0 +1,196 @@
+"""Tests for the Hub's login throttling.
+
+These exercise the exact production implementation in
+sovran_systemsos_web.security_helpers.LoginThrottle. The clock and the sleep are
+injected, so the tests cover hours of lockout behaviour instantly.
+
+No network access, no filesystem writes, no real delays.
+"""
+
+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
+    LoginThrottle,
+    LOGIN_FAIL_DELAY,
+    LOGIN_FAIL_MAX_DELAY,
+    LOGIN_FAIL_WINDOW,
+    LOGIN_FAIL_MAX,
+    LOGIN_LOCKOUT_SECONDS,
+)
+
+
+class FakeClock:
+    """A clock that only moves when the test says so."""
+
+    def __init__(self):
+        self.now = 1000.0
+
+    def __call__(self):
+        return self.now
+
+    def advance(self, seconds):
+        self.now += seconds
+
+
+class FakeSleeper:
+    """Records the delays it was asked to apply instead of sleeping."""
+
+    def __init__(self, clock):
+        self.clock = clock
+        self.calls = []
+
+    def __call__(self, seconds):
+        self.calls.append(seconds)
+        self.clock.advance(seconds)
+
+
+def _make(**kwargs):
+    clock = kwargs.pop("clock", None) or FakeClock()
+    sleep = kwargs.pop("sleep", None) or FakeSleeper(clock)
+    return LoginThrottle(clock=clock, sleep=sleep, **kwargs), clock, sleep
+
+
+def _trip(throttle, ip="203.0.113.9"):
+    """Fail LOGIN_FAIL_MAX times. The fake sleeper advances the clock for us."""
+    for _ in range(LOGIN_FAIL_MAX):
+        throttle.record_failure(ip)
+
+
+class DelayRamp(unittest.TestCase):
+
+    def test_delay_ramps_with_the_failure_count(self):
+        throttle, _, _ = _make()
+        self.assertEqual(throttle.delay_for(0), 0.0)
+        self.assertEqual(throttle.delay_for(1), LOGIN_FAIL_DELAY)
+        self.assertEqual(throttle.delay_for(3), LOGIN_FAIL_DELAY * 3)
+
+    def test_delay_is_capped(self):
+        # Unbounded ramping would let a single client park a thread-pool worker
+        # for minutes at a time.
+        throttle, _, _ = _make()
+        self.assertLessEqual(throttle.delay_for(999), LOGIN_FAIL_MAX_DELAY)
+        self.assertEqual(throttle.delay_for(999), LOGIN_FAIL_MAX_DELAY)
+
+    def test_first_failure_is_not_delayed_much(self):
+        throttle, _, sleep = _make()
+        delay = throttle.record_failure("203.0.113.9")
+        self.assertEqual(delay, LOGIN_FAIL_DELAY)
+        self.assertEqual(sleep.calls, [LOGIN_FAIL_DELAY])
+
+
+class Lockout(unittest.TestCase):
+
+    def test_not_locked_out_initially(self):
+        throttle, _, _ = _make()
+        self.assertFalse(throttle.is_locked_out("203.0.113.9"))
+        self.assertEqual(throttle.remaining_lockout("203.0.113.9"), 0.0)
+
+    def test_reaching_the_limit_locks_the_address_out(self):
+        throttle, _, _ = _make()
+        _trip(throttle)
+        self.assertTrue(throttle.is_locked_out("203.0.113.9"))
+
+    def test_the_limit_is_reachable_inside_the_window(self):
+        # Regression guard for the old 60s window: with a ramping delay it
+        # takes ~80s to reach LOGIN_FAIL_MAX, so a 60s window expired the
+        # earliest failures first and the lockout could never fire.
+        throttle, clock, _ = _make()
+        start = clock.now
+        _trip(throttle)
+        self.assertLess(clock.now - start, LOGIN_FAIL_WINDOW)
+        self.assertEqual(throttle.failure_count("203.0.113.9"), LOGIN_FAIL_MAX)
+        self.assertTrue(throttle.is_locked_out("203.0.113.9"))
+
+    def test_one_failure_short_of_the_limit_is_not_a_lockout(self):
+        throttle, _, _ = _make()
+        for _ in range(LOGIN_FAIL_MAX - 1):
+            throttle.record_failure("203.0.113.9")
+        self.assertFalse(throttle.is_locked_out("203.0.113.9"))
+
+    def test_lockout_expires(self):
+        throttle, clock, _ = _make()
+        _trip(throttle)
+        self.assertTrue(throttle.is_locked_out("203.0.113.9"))
+        clock.advance(LOGIN_LOCKOUT_SECONDS + 1)
+        self.assertFalse(throttle.is_locked_out("203.0.113.9"))
+
+    def test_remaining_lockout_counts_down(self):
+        throttle, clock, _ = _make()
+        _trip(throttle)
+        full = throttle.remaining_lockout("203.0.113.9")
+        # the final record_failure applied a delay, which the fake clock has
+        # already advanced, so what is left is the lockout minus that delay
+        self.assertAlmostEqual(full, LOGIN_LOCKOUT_SECONDS,
+                               delta=LOGIN_FAIL_MAX_DELAY + 1.0)
+        clock.advance(full / 2)
+        self.assertLess(throttle.remaining_lockout("203.0.113.9"), full)
+        self.assertGreater(throttle.remaining_lockout("203.0.113.9"), 0.0)
+
+    def test_further_failures_while_locked_out_extend_it(self):
+        throttle, clock, _ = _make()
+        _trip(throttle)
+        clock.advance(LOGIN_LOCKOUT_SECONDS - 1)
+        throttle.record_failure("203.0.113.9")
+        self.assertTrue(throttle.is_locked_out("203.0.113.9"))
+
+
+class Isolation(unittest.TestCase):
+
+    def test_one_address_does_not_lock_out_another(self):
+        throttle, _, _ = _make()
+        _trip(throttle, "203.0.113.9")
+        self.assertTrue(throttle.is_locked_out("203.0.113.9"))
+        self.assertFalse(throttle.is_locked_out("198.51.100.7"))
+
+    def test_successful_login_clears_the_address(self):
+        throttle, _, _ = _make()
+        for _ in range(LOGIN_FAIL_MAX - 1):
+            throttle.record_failure("203.0.113.9")
+        throttle.clear("203.0.113.9")
+        self.assertEqual(throttle.failure_count("203.0.113.9"), 0)
+        self.assertFalse(throttle.is_locked_out("203.0.113.9"))
+
+    def test_old_failures_age_out_of_the_window(self):
+        throttle, clock, _ = _make()
+        throttle.record_failure("203.0.113.9")
+        clock.advance(LOGIN_FAIL_WINDOW + 1)
+        self.assertEqual(throttle.failure_count("203.0.113.9"), 0)
+
+
+class BoundedMemory(unittest.TestCase):
+
+    def test_tracked_addresses_are_evicted(self):
+        throttle, clock, _ = _make(max_tracked_ips=8)
+        for i in range(64):
+            throttle.record_failure(f"198.51.100.{i}")
+        clock.advance(LOGIN_FAIL_WINDOW + LOGIN_LOCKOUT_SECONDS + 1)
+        throttle.record_failure("203.0.113.9")
+        self.assertLessEqual(throttle.tracked_addresses(), 8)
+
+    def test_sleep_is_never_called_under_the_lock(self):
+        # If the lock were held across the sleep, one slow client would stall
+        # every other login — a self-inflicted DoS.
+        throttle, clock, _ = _make()
+        order = []
+
+        def spy(seconds):
+            order.append("sleep:start")
+            clock.advance(seconds)
+            order.append("sleep:end")
+
+        throttle._sleep = spy
+        throttle.record_failure("203.0.113.9")
+        self.assertEqual(order, ["sleep:start", "sleep:end"])
+        # A second address can still be recorded while the first is "sleeping".
+        self.assertEqual(throttle.failure_count("198.51.100.7"), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ssh_exposure.py

@@ -0,0 +1,64 @@
+"""Guards for when port 22 is open in the firewall.
+
+sshd-localhost.nix gives every role "ssh root@localhost" by listening on
+127.0.0.1 only. NixOS opens sshd's ports in the firewall by default whether or
+not sshd listens on them, which left port 22 open on every role, Desktop Only
+included, with nothing behind it. The roles that really publish SSH open it
+explicitly, so the default has to stay off.
+
+Like the other nix-file checks these read the modules as text: 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 _without_comments(src):
+    return "\n".join(l for l in src.splitlines() if not l.lstrip().startswith("#"))
+
+
+class LocalhostSshdDoesNotOpenThePort(unittest.TestCase):
+
+    def test_the_firewall_is_not_opened_by_default(self):
+        code = _without_comments(_read("modules", "core", "sshd-localhost.nix"))
+        self.assertRegex(code, r"openFirewall\s*=\s*lib\.mkDefault\s+false\s*;")
+
+    def test_it_still_listens_on_loopback_only(self):
+        code = _without_comments(_read("modules", "core", "sshd-localhost.nix"))
+        self.assertRegex(code, r'addr\s*=\s*"127\.0\.0\.1"')
+        self.assertNotIn("0.0.0.0", code)
+
+
+class PublishedSshOpensItsOwnPort(unittest.TestCase):
+    """Turning the default off must not close the roles that want SSH open."""
+
+    def test_the_sshd_feature_opens_22_and_only_when_enabled(self):
+        src = _without_comments(_read("modules", "sshd.nix"))
+        self.assertRegex(src, r"lib\.mkIf\s+config\.sovran_systemsOS\.features\.sshd")
+        self.assertRegex(src, r"networking\.firewall\.allowedTCPPorts\s*=\s*\[\s*22\s*\]")
+
+    def test_remote_deploy_opens_22_and_only_when_enabled(self):
+        src = _without_comments(_read("modules", "core", "remote-deploy.nix"))
+        self.assertRegex(src, r"lib\.mkIf\s+cfg\.enable")
+        self.assertRegex(src, r"networking\.firewall\.allowedTCPPorts\s*=\s*\[\s*22\s*\]")
+
+
+class DesktopOnlyDocumentsWhatItOpens(unittest.TestCase):
+
+    def test_security_policy_says_desktop_opens_no_tcp_port(self):
+        text = " ".join(_read("SECURITY.md").split())  # the file is line-wrapped
+        self.assertIn("opens no TCP port", text)
+        self.assertIn("UDP 5353", text)
+
+
+if __name__ == "__main__":
+    unittest.main()