Sovran_Systems/Sovran_SystemsOS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 5 Wiki Activity

Security hardening: fix DDNS injection, Nix injection, reboot auth, support key, sudo rules

Browse Source 
Co-authored-by: naturallaw777 <99053422+naturallaw777@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-08-11 10:44:26 +00:00
committed by GitHub
co-authored by naturallaw777
parent 590ed134b7
commit f2ad9c1f17
3 changed files with 637 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/sovran_systemsos_web/server.py
+221 -80
View File
@@ -82,6 +82,23 @@ 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: "npub1" followed by exactly 58 bech32 characters
NPUB_RE = re.compile(r"^npub1[023456789acdefghjklmnpqrstuvwxyz]{58}$")
# Accepted SSH public-key algorithms for support sessions
_SSH_PUBKEY_ALGORITHMS = frozenset([
"ssh-ed25519",
"ecdsa-sha2-nistp256",
"ecdsa-sha2-nistp384",
"ecdsa-sha2-nistp521",
"sk-ssh-ed25519@openssh.com",
])
# DDNS URL: HTTPS only, no credentials, no control chars, max 2048 bytes
_DDNS_URL_MAX_LEN = 2048
_DDNS_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f]")
# Systemd service that rewrites the Sovran-managed /etc/hosts loopback block
SOVRAN_HOSTS_SERVICE = "sovran-hosts-update.service"
@@ -123,7 +140,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,9 +157,6 @@ 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"
# Dedicated restricted support user (non-root) for wallet privacy
@@ -514,6 +528,97 @@ _DICEWARE_WORDS = [
]
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
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 present and not a raw IP address.
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")
hostname = parsed.hostname or ""
if not hostname:
raise ValueError("DDNS URL must contain a hostname")
# Reject raw IP addresses — DDNS providers use hostnames
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
return url
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}")
# Validate base64 payload
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
def _generate_diceware_password() -> str:
"""Generate a human-readable diceware-style passphrase: word-word-word-N."""
import secrets as _secrets
@@ -1847,11 +1952,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 "")
@@ -1949,19 +2054,10 @@ 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 support user's authorized_keys first
"""Check if a per-session support key is currently installed."""
try:
with open(SUPPORT_USER_AUTH_KEYS, "r") as f:
if SUPPORT_KEY_COMMENT in f.read():
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
return bool(f.read().strip())
except FileNotFoundError:
return False
@@ -2116,12 +2212,16 @@ def _get_wallet_unlock_info() -> dict:
return {}
def _enable_support() -> bool:
"""Add the Sovran support public key to the restricted support user's authorized_keys.
def _enable_support(pubkey: str) -> bool:
"""Install a per-session SSH public key for the restricted support user.
Falls back to root's authorized_keys if the support user cannot be created.
The key is written only to the ``sovran-support`` account's
``authorized_keys``; 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()
@@ -2129,7 +2229,7 @@ def _enable_support() -> bool:
if use_restricted_user:
os.makedirs(SUPPORT_USER_SSH_DIR, mode=0o700, exist_ok=True)
with open(SUPPORT_USER_AUTH_KEYS, "w") as f:
f.write(SOVRAN_SUPPORT_PUBKEY + "\n")
f.write(pubkey.strip() + "\n")
os.chmod(SUPPORT_USER_AUTH_KEYS, 0o600)
try:
pw = pwd.getpwnam(SUPPORT_USER)
@@ -2138,23 +2238,9 @@ def _enable_support() -> bool:
except Exception:
pass
else:
# Fallback: add key to root's authorized_keys
os.makedirs("/root/.ssh", mode=0o700, exist_ok=True)
with open(SUPPORT_KEY_FILE, "w") as f:
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)
# Support user could not be created; fail closed rather than
# falling back to root's authorized_keys.
return False
acl_applied = _apply_wallet_acls() if use_restricted_user else False
wallet_paths = _get_existing_wallet_paths()
@@ -2182,7 +2268,7 @@ def _enable_support() -> bool:
def _disable_support() -> bool:
"""Remove the Sovran support public key and revoke all wallet access."""
"""Remove the per-session support key and revoke all wallet access."""
try:
# Remove from support user's authorized_keys
try:
@@ -2190,18 +2276,7 @@ def _disable_support() -> bool:
except FileNotFoundError:
pass
# Remove from root's authorized_keys (fallback / legacy)
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
# Remove the dedicated key file (legacy path, best-effort)
try:
os.remove(SUPPORT_KEY_FILE)
except FileNotFoundError:
@@ -2229,19 +2304,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:
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
return True
# ── Routes ───────────────────────────────────────────────────────
@@ -3855,10 +3925,24 @@ async def api_support_status():
}
class SupportEnableRequest(BaseModel):
ssh_public_key: str
@app.post("/api/support/enable")
async def api_support_enable():
"""Add the Sovran support SSH key to allow remote tech support.
Requires the sshd feature to be enabled first."""
async def api_support_enable(req: SupportEnableRequest):
"""Install a per-session SSH public key for the restricted support account.
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 +3953,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 +4296,8 @@ async def api_features_toggle(req: FeatureToggleRequest):
if req.feature == "haven":
npub = (req.extra or {}).get("nostr_npub", "").strip()
if npub:
if not NPUB_RE.fullmatch(npub):
raise HTTPException(status_code=400, detail="Invalid Nostr npub (must be npub1 followed by 58 bech32 characters)")
nostr_npub = npub
elif not nostr_npub:
raise HTTPException(status_code=400, detail="nostr_npub is required for Haven")
@@ -4227,6 +4313,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 NPUB_RE.fullmatch(new_npub):
raise HTTPException(status_code=400, detail="Invalid Nostr npub (must be npub1 followed by 58 bech32 characters)")
nostr_npub = new_npub
try:
os.makedirs(os.path.dirname(NOSTR_NPUB_FILE), exist_ok=True)
@@ -4402,22 +4490,71 @@ def _ensure_njalla_script() -> None:
pass
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)
fd, tmp = tempfile.mkstemp(dir=njalla_dir, prefix=".ddns_urls_tmp")
try:
with os.fdopen(fd, "w") as f:
json.dump(urls, f)
os.replace(tmp, NJALLA_DDNS_URLS_FILE)
except Exception:
try:
os.unlink(tmp)
except OSError:
pass
raise
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.
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 cron job (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)
try:
subprocess.run(
["bash", NJALLA_SCRIPT], timeout=30, check=False,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
ip_result = subprocess.run(
["dig", "@resolver4.opendns.com", "myip.opendns.com", "+short", "-4"],
capture_output=True, text=True, timeout=10, check=False,
)
public_ip = ip_result.stdout.strip()
except Exception:
pass
public_ip = ""
for raw_url in urls:
try:
# Replace the placeholder with the resolved IP (safe string replacement)
url = raw_url.replace("${IP}", public_ip) if public_ip else raw_url
subprocess.run(
["curl", "--silent", "--max-time", "15", "--fail", url],
timeout=20, check=False,
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
)
except Exception:
pass
def _reload_caddy_for_domain_change() -> None:
@@ -4550,25 +4687,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
# 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')
# Validate URL strictly — reject injection attempts before persisting
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
modules/core/tech-support.nix
+14 -7
View File
@@ -43,17 +43,24 @@
];
# ── Scoped sudo rules for support staff ───────────────────────────────────
# Grants only the minimum privileges needed for a support session.
# Support staff cannot stop/disable/mask services or access wallet files.
# Grants only the minimum privileges needed for diagnostic support.
# 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.
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/nano /etc/nixos/configuration.nix"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/nixos-rebuild switch --flake /etc/nixos"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/systemctl restart *"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/journalctl *"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/systemctl restart sovran-hub.service"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/systemctl restart caddy.service"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/systemctl restart bitcoind.service"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/systemctl restart lnd.service"; options = [ "NOPASSWD" ]; }
{ command = "/run/current-system/sw/bin/systemctl status sovran-hub.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" ]; }
{ command = "/run/current-system/sw/bin/journalctl *"; options = [ "NOPASSWD" ]; }
];
}
];
tests/test_security.py
+402
View File
@@ -0,0 +1,402 @@
"""Security regression tests for Sovran Hub server helpers.
Tests cover the concrete payload classes described in the security review:
- DDNS URL validation (injection payloads)
- Nix string escaping (injection into generated Nix source)
- Nostr npub validation (Nix injection via nostr_npub)
- SSH public-key validation (support-key handling)
- /api/reboot auth-exemption removal
These tests use only the Python standard library so they run without installing
the full application dependency tree. The helper functions are replicated from
server.py to allow isolated unit testing.
"""
import base64
import ipaddress
import re
import sys
import types
import unittest
import urllib.parse
# ---------------------------------------------------------------------------
# Replicate the helpers under test so we can test them without importing the
# full FastAPI application (which is not available in CI).
# ---------------------------------------------------------------------------
# ── _nix_escape ──────────────────────────────────────────────────────────────
def _nix_escape(value: str) -> str:
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
# ── NPUB_RE ───────────────────────────────────────────────────────────────────
NPUB_RE = re.compile(r"^npub1[023456789acdefghjklmnpqrstuvwxyz]{58}$")
# ── _validate_ddns_url ────────────────────────────────────────────────────────
_DDNS_URL_MAX_LEN = 2048
_DDNS_CONTROL_RE = re.compile(r"[\x00-\x1f\x7f]")
def _validate_ddns_url(url: str) -> str:
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")
hostname = parsed.hostname or ""
if not hostname:
raise ValueError("DDNS URL must contain a hostname")
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
return url
# ── _validate_ssh_pubkey ──────────────────────────────────────────────────────
_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:
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
# ---------------------------------------------------------------------------
# Test cases
# ---------------------------------------------------------------------------
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):
result = _nix_escape("${pkgs.bash}")
# Nix interpolation is prevented by the leading backslash; the raw
# result must contain "\\${" (backslash then ${), not bare "${".
self.assertIn("\\${", result)
# The result must not start with "${" (unescaped)
self.assertFalse(result.startswith("${"))
def test_newline_escaped(self):
result = _nix_escape("foo\nbar")
self.assertNotIn("\n", result)
self.assertIn("\\n", result)
def test_carriage_return_escaped(self):
result = _nix_escape("foo\rbar")
self.assertNotIn("\r", result)
def test_tab_escaped(self):
result = _nix_escape("foo\tbar")
self.assertNotIn("\t", result)
def test_semicolons_unchanged(self):
# Semicolons are safe inside Nix string literals
self.assertEqual(_nix_escape("a;b"), "a;b")
def test_valid_timezone(self):
# Typical timezone value must pass through unchanged
self.assertEqual(_nix_escape("Europe/London"), "Europe/London")
def test_injection_payload_quotes_and_interpolation(self):
payload = '"; import <nixpkgs/nixos/tests/keymap.nix> { ${builtins.readFile "/etc/shadow"} }'
result = _nix_escape(payload)
# Unescaped double-quotes must not appear in the result
self.assertNotIn('"', result.replace('\\"', ""))
# All ${...} sequences must be preceded by backslash
self.assertIn("\\${", result)
# No bare ${ that is not preceded by backslash
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)
# The result should not contain unescaped quotes
self.assertNotIn('"', result.replace('\\"', ""))
class TestNpubValidation(unittest.TestCase):
"""NPUB_RE must accept valid npubs and reject injection payloads."""
# A real npub (58 bech32 chars after "npub1")
VALID_NPUB = "npub1" + "q" * 58 # synthetic but format-correct
def test_valid_npub_accepted(self):
self.assertIsNotNone(NPUB_RE.fullmatch(self.VALID_NPUB))
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_injection_with_quote_rejected(self):
payload = 'npub1aaa"; extraUsers.evil.isNormalUser = true; #'
self.assertIsNone(NPUB_RE.fullmatch(payload))
def test_injection_with_interpolation_rejected(self):
payload = "npub1" + "${" + "q" * 52
self.assertIsNone(NPUB_RE.fullmatch(payload))
def test_injection_with_newline_rejected(self):
payload = "npub1" + "q" * 30 + "\n" + "q" * 28
self.assertIsNone(NPUB_RE.fullmatch(payload))
def test_injection_with_semicolon_rejected(self):
payload = "npub1" + "q" * 30 + ";" + "q" * 27
self.assertIsNone(NPUB_RE.fullmatch(payload))
def test_injection_with_backslash_rejected(self):
payload = "npub1" + "q" * 30 + "\\" + "q" * 27
self.assertIsNone(NPUB_RE.fullmatch(payload))
def test_uppercase_letters_rejected(self):
# bech32 is lower-case only (uppercase I, O, B, 1 excluded)
self.assertIsNone(NPUB_RE.fullmatch("npub1" + "Q" * 58))
def test_empty_rejected(self):
self.assertIsNone(NPUB_RE.fullmatch(""))
class TestDdnsUrlValidation(unittest.TestCase):
"""_validate_ddns_url must reject injection payloads."""
VALID_URL = "https://njal.la/update/?h=test.example.com&k=TOKEN&a=${IP}"
def test_valid_url_accepted(self):
result = _validate_ddns_url(self.VALID_URL)
self.assertEqual(result, self.VALID_URL)
def test_http_scheme_rejected(self):
with self.assertRaises(ValueError):
_validate_ddns_url("http://njal.la/update/?h=test&k=TOKEN")
def test_ftp_scheme_rejected(self):
with self.assertRaises(ValueError):
_validate_ddns_url("ftp://njal.la/update/?h=test&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#fragment")
def test_raw_ip_host_rejected(self):
with self.assertRaises(ValueError):
_validate_ddns_url("https://1.2.3.4/update/?k=TOKEN")
def test_control_character_newline_rejected(self):
with self.assertRaises(ValueError):
_validate_ddns_url("https://njal.la/update/?k=TOKEN\nmalicious")
def test_control_character_null_rejected(self):
with self.assertRaises(ValueError):
_validate_ddns_url("https://njal.la/update/?k=TOKEN\x00evil")
def test_semicolon_in_query_accepted(self):
# Semicolons are valid in query strings
result = _validate_ddns_url("https://njal.la/update/?h=test&k=abc;def")
self.assertIsNotNone(result)
def test_quote_injection_in_url_accepted_as_url(self):
# A quote character is valid URL data (percent-encoded in practice),
# but even unencoded it is safe because _validate_ddns_url validates
# structure not characters. The important thing is the URL passes
# through to curl as a single argument — shell injection is impossible.
url = 'https://njal.la/update/?k=abc"def'
result = _validate_ddns_url(url)
self.assertEqual(result, url)
def test_backtick_in_url_accepted_as_url(self):
url = "https://njal.la/update/?k=abc`id`"
result = _validate_ddns_url(url)
self.assertEqual(result, url)
def test_dollar_sign_in_url_accepted_as_url(self):
url = "https://njal.la/update/?k=$(cat /etc/passwd)"
result = _validate_ddns_url(url)
self.assertEqual(result, url)
def test_empty_url_rejected(self):
with self.assertRaises(ValueError):
_validate_ddns_url("")
def test_too_long_rejected(self):
with self.assertRaises(ValueError):
_validate_ddns_url("https://njal.la/?" + "x" * 2050)
class TestSshPubkeyValidation(unittest.TestCase):
"""_validate_ssh_pubkey must accept valid keys and reject injections."""
# Minimal valid ed25519 key (32-byte payload, base64-encoded)
_PAYLOAD = base64.b64encode(b"\x00" * 64).decode()
VALID_KEY = f"ssh-ed25519 {_PAYLOAD} user@host"
def test_valid_ed25519_accepted(self):
result = _validate_ssh_pubkey(self.VALID_KEY)
self.assertEqual(result, self.VALID_KEY)
def test_unsupported_algorithm_rejected(self):
payload = base64.b64encode(b"\x00" * 40).decode()
with self.assertRaises(ValueError):
_validate_ssh_pubkey(f"ssh-rsa {payload} user@host")
def test_dss_algorithm_rejected(self):
payload = base64.b64encode(b"\x00" * 40).decode()
with self.assertRaises(ValueError):
_validate_ssh_pubkey(f"ssh-dss {payload} user@host")
def test_multiline_injection_rejected(self):
with self.assertRaises(ValueError):
_validate_ssh_pubkey(f"{self.VALID_KEY}\nssh-ed25519 AAAA second-key")
def test_options_prefix_not_accepted(self):
# OpenSSH key options ("command=..." etc.) should be rejected as
# the algorithm will not match.
with self.assertRaises(ValueError):
_validate_ssh_pubkey(f'command="evil" {self.VALID_KEY}')
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(f"ssh-ed25519 {self._PAYLOAD}\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 = base64.b64encode(b"\x00" * 5).decode()
with self.assertRaises(ValueError):
_validate_ssh_pubkey(f"ssh-ed25519 {short} user@host")
class TestAuthExemptPaths(unittest.TestCase):
"""/api/reboot and status endpoints must not be in the auth-exempt set."""
def _get_exempt_paths(self):
"""Extract _AUTH_EXEMPT_PATHS from server.py without importing it."""
import re
src = open(
__file__.replace("tests/test_security.py", "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):
exempt = self._get_exempt_paths()
self.assertNotIn("/api/reboot", exempt)
def test_updates_status_not_exempt(self):
exempt = self._get_exempt_paths()
self.assertNotIn("/api/updates/status", exempt)
def test_rebuild_status_not_exempt(self):
exempt = self._get_exempt_paths()
self.assertNotIn("/api/rebuild/status", exempt)
def test_login_still_exempt(self):
exempt = self._get_exempt_paths()
self.assertIn("/api/login", exempt)
def test_ping_still_exempt(self):
exempt = self._get_exempt_paths()
self.assertIn("/api/ping", exempt)
class TestTechSupportSudoRules(unittest.TestCase):
"""tech-support.nix must not grant Nix-edit or unrestricted rebuild/systemctl."""
def _get_nix_content(self):
import os
path = os.path.join(
os.path.dirname(__file__), "..", "modules", "core", "tech-support.nix"
)
with open(os.path.normpath(path)) as f:
return f.read()
def test_no_nano_custom_nix(self):
content = self._get_nix_content()
self.assertNotIn("nano /etc/nixos/custom.nix", content)
def test_no_nano_configuration_nix(self):
content = self._get_nix_content()
self.assertNotIn("nano /etc/nixos/configuration.nix", content)
def test_no_unrestricted_nixos_rebuild(self):
content = self._get_nix_content()
self.assertNotIn("nixos-rebuild switch", content)
def test_no_wildcard_systemctl_restart(self):
content = self._get_nix_content()
self.assertNotIn("systemctl restart *", content)
if __name__ == "__main__":
unittest.main()