diff --git a/src/netmover/crypto.py b/src/netmover/crypto.py new file mode 100644 index 0000000..4f42303 --- /dev/null +++ b/src/netmover/crypto.py @@ -0,0 +1,88 @@ +"""Geheimnisse (Fallstrick 7): Fernet-Verschlüsselung, SSH-Schlüssel, Schwärzung.""" + +from __future__ import annotations + +import os +import stat +from collections.abc import Iterable +from pathlib import Path + +from cryptography.fernet import Fernet, InvalidToken +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from netmover.logging_setup import event, get_logger + +log = get_logger(__name__) + +REDACTED = "••••••" + + +class SecretError(RuntimeError): + pass + + +def load_or_create_key(path: Path) -> bytes: + """Liest den Fernet-Schlüssel oder legt ihn beim ersten Start mit Modus 0600 an.""" + path.parent.mkdir(parents=True, exist_ok=True) + try: + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + except FileExistsError: + mode = stat.S_IMODE(path.stat().st_mode) + if mode & 0o077: + path.chmod(0o600) + event(log, "secret_key.mode_fixed", path=str(path), old_mode=oct(mode)) + key = path.read_bytes().strip() + try: + Fernet(key) + except ValueError as exc: + raise SecretError(f"{path} enthält keinen gültigen Schlüssel") from exc + return key + key = Fernet.generate_key() + with os.fdopen(fd, "wb") as fh: + fh.write(key + b"\n") + fh.flush() + os.fsync(fh.fileno()) + event(log, "secret_key.created", path=str(path)) + return key + + +class SecretBox: + def __init__(self, key: bytes) -> None: + self._fernet = Fernet(key) + + def encrypt(self, plaintext: str) -> str: + return self._fernet.encrypt(plaintext.encode()).decode() + + def decrypt(self, token: str) -> str: + try: + return self._fernet.decrypt(token.encode()).decode() + except InvalidToken as exc: + raise SecretError( + "Gespeichertes Geheimnis lässt sich nicht entschlüsseln – " + "wurde secret.key ausgetauscht?" + ) from exc + + +def generate_ssh_keypair(comment: str = "netmover") -> tuple[str, str]: + """Erzeugt ein Ed25519-Schlüsselpaar: (privat im OpenSSH-Format, öffentlich).""" + key = Ed25519PrivateKey.generate() + private = key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.OpenSSH, + serialization.NoEncryption(), + ).decode() + public = ( + key.public_key() + .public_bytes(serialization.Encoding.OpenSSH, serialization.PublicFormat.OpenSSH) + .decode() + ) + return private, f"{public} {comment}" + + +def redact(text: str, secrets: Iterable[str | None]) -> str: + """Ersetzt bekannte Geheimnisse, bevor Ausgaben protokolliert oder angezeigt werden.""" + # Längste zuerst, damit Teilstrings eines längeren Geheimnisses nicht stehen bleiben. + for secret in sorted({s for s in secrets if s and len(s) >= 3}, key=len, reverse=True): + text = text.replace(secret, REDACTED) + return text diff --git a/src/netmover/timeouts.py b/src/netmover/timeouts.py new file mode 100644 index 0000000..17d7441 --- /dev/null +++ b/src/netmover/timeouts.py @@ -0,0 +1,58 @@ +"""Dateizugriffe mit Zeitlimit (Fallstrick 13). + +Ein Zugriff auf einen hängenden ``hard``-NFS-Mount blockiert im Kernel und lässt +sich weder abbrechen noch beenden. Deshalb läuft jeder solche Zugriff in einem +eigenen Daemon-Thread: Läuft das Zeitlimit ab, geben wir auf und lassen den +Thread zurück. Er blockiert nur sich selbst, nie den Scheduler oder andere +Regeln. Bewusst kein Thread-Pool – hängende Threads würden ihn auffüllen. +""" + +from __future__ import annotations + +import threading +from collections.abc import Callable + +from netmover.logging_setup import event, get_logger + +log = get_logger(__name__) + +_abandoned = 0 +_abandoned_lock = threading.Lock() + + +class OperationTimeout(TimeoutError): + def __init__(self, what: str, seconds: float) -> None: + super().__init__(f"{what}: keine Antwort nach {seconds:g} s") + self.what = what + self.seconds = seconds + + +def abandoned_threads() -> int: + """Anzahl zurückgelassener (vermutlich hängender) Threads – für die Statusseite.""" + return _abandoned + + +def run_with_timeout[T](fn: Callable[[], T], seconds: float, what: str) -> T: + result: list[T] = [] + error: list[BaseException] = [] + done = threading.Event() + + def target() -> None: + try: + result.append(fn()) + except BaseException as exc: + error.append(exc) + finally: + done.set() + + thread = threading.Thread(target=target, name=f"nm-timeout:{what}", daemon=True) + thread.start() + if not done.wait(seconds): + global _abandoned + with _abandoned_lock: + _abandoned += 1 + event(log, "operation.timeout", level=30, what=what, seconds=seconds) + raise OperationTimeout(what, seconds) + if error: + raise error[0] + return result[0] diff --git a/src/netmover/validation.py b/src/netmover/validation.py new file mode 100644 index 0000000..a1a3071 --- /dev/null +++ b/src/netmover/validation.py @@ -0,0 +1,245 @@ +"""Validierung aller Eingaben, die in einen ``mount``-Aufruf oder einen Pfad fließen. + +Fallstrick 5 (Shell-Injection): Server, Freigabe und Zusatzoptionen kommen aus +einem Webformular. Sie werden hier gegen strikte Muster geprüft; Optionen nur +gegen eine Allowlist erlaubter Schlüssel. Zusätzlich ruft das Backend ``mount`` +ausschließlich mit Argumentliste auf – beides zusammen, nicht eins von beiden. +""" + +from __future__ import annotations + +import ipaddress +import re +import unicodedata +from dataclasses import dataclass +from pathlib import PurePosixPath + +from netmover.models import MountType + + +class ValidationError(ValueError): + """Eingabe ungültig. ``field`` benennt das Formularfeld, ``str(exc)`` den Grund.""" + + def __init__(self, field: str, message: str) -> None: + super().__init__(message) + self.field = field + + +_HOSTNAME_RE = re.compile( + r"^(?=.{1,253}$)[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?" + r"(?:\.[A-Za-z0-9](?:[A-Za-z0-9-]{0,61}[A-Za-z0-9])?)*$" +) +# Freigabename, optional mit Unterpfad: "scans" oder "scans/eingang" +_CIFS_SHARE_RE = re.compile(r"^[A-Za-z0-9._-]+(?:/[A-Za-z0-9._-]+)*$") +# Absoluter Exportpfad bzw. entfernter Basispfad: "/volume1/scans" +_ABS_PATH_RE = re.compile(r"^/(?:[A-Za-z0-9._-]+/?)*$") +_OPT_KEY_RE = re.compile(r"^[a-z][a-z0-9_]*$") +_OPT_VALUE_RE = re.compile(r"^[A-Za-z0-9._:/@-]+$") +_MODE_RE = re.compile(r"^0?[0-7]{3}$") +_USERNAME_RE = re.compile(r"^[A-Za-z0-9._@$\\-]{1,255}$") + +# Schlüssel, die ausschließlich das Tool setzt. +FORBIDDEN_OPTION_KEYS = frozenset( + {"credentials", "cred", "username", "user", "password", "password2", "pass", "domain"} +) +# Schlüssel mit eigenem Formularfeld. +DEDICATED_OPTION_KEYS = frozenset({"uid", "gid", "file_mode", "dir_mode"}) + +_COMMON_FLAGS = frozenset({"ro", "rw", "noexec", "nosuid", "nodev", "hard", "soft", "noatime"}) + +ALLOWED_OPTIONS: dict[MountType, frozenset[str]] = { + MountType.CIFS: _COMMON_FLAGS + | { + "vers", "sec", "iocharset", "noperm", "nounix", "serverino", "noserverino", + "cache", "actimeo", "rsize", "wsize", "seal", "nobrl", "mfsymlinks", + "forceuid", "forcegid", "noacl", "echo_interval", "handlecache", + "nohandlecache", "nocase", "port", "max_credits", "retrans", + }, + MountType.NFS4: _COMMON_FLAGS + | { + "vers", "proto", "port", "timeo", "retrans", "rsize", "wsize", "ac", "noac", + "actimeo", "lookupcache", "nconnect", "sec", "softerr", "fsc", "clientaddr", + }, + MountType.NFS3: _COMMON_FLAGS + | { + "proto", "port", "timeo", "retrans", "rsize", "wsize", "ac", "noac", + "actimeo", "lookupcache", "nconnect", "sec", "nolock", "local_lock", + "mountport", "mountproto", "softerr", "fsc", + }, + MountType.SFTP: frozenset(), +} # fmt: skip + +_NFS4_VERSIONS = frozenset({"4", "4.0", "4.1", "4.2"}) + + +def validate_server(value: str) -> str: + value = value.strip() + if not value: + raise ValidationError("server", "Server fehlt.") + try: + ip = ipaddress.ip_address(value) + except ValueError: + ip = None + if ip is not None: + if ip.version == 6: + raise ValidationError( + "server", "IPv6-Adressen werden nicht unterstützt – bitte Hostname verwenden." + ) + return value + if not _HOSTNAME_RE.fullmatch(value): + raise ValidationError( + "server", "Ungültiger Servername – erlaubt sind Hostname oder IPv4-Adresse." + ) + return value + + +def _no_dot_segments(field: str, value: str) -> None: + if any(part in {".", ".."} for part in value.split("/")): + raise ValidationError(field, "„.“ und „..“ sind als Pfadbestandteil nicht erlaubt.") + + +def validate_share(mount_type: MountType, value: str) -> str: + value = value.strip() + if mount_type is MountType.CIFS: + value = value.strip("/") + if not _CIFS_SHARE_RE.fullmatch(value): + raise ValidationError( + "share", + "Ungültige Freigabe – erlaubt sind Buchstaben, Ziffern, „.“, „_“, „-“ und " + "„/“ als Trenner (z. B. „scans“ oder „scans/eingang“).", + ) + else: + if len(value) > 1: + value = value.rstrip("/") + if not _ABS_PATH_RE.fullmatch(value): + raise ValidationError( + "share", + "Ungültiger Pfad – absolut, nur Buchstaben, Ziffern, „.“, „_“, „-“ " + "(z. B. „/volume1/scans“).", + ) + _no_dot_segments("share", value) + return value + + +def validate_mode(field: str, value: str | None) -> str | None: + if value is None or value.strip() == "": + return None + value = value.strip() + if not _MODE_RE.fullmatch(value): + raise ValidationError(field, "Modus bitte oktal angeben, z. B. „0644“.") + return value.rjust(4, "0") + + +def validate_id(field: str, value: int | str | None) -> int | None: + if value is None or (isinstance(value, str) and value.strip() == ""): + return None + try: + num = int(value) + except (TypeError, ValueError): + raise ValidationError(field, "Bitte eine ganze Zahl angeben.") from None + if not 0 <= num <= 2**32 - 2: + raise ValidationError(field, "Wert außerhalb des gültigen Bereichs.") + return num + + +def validate_port(value: int | str | None) -> int | None: + num = validate_id("port", value) + if num is not None and not 1 <= num <= 65535: + raise ValidationError("port", "Port muss zwischen 1 und 65535 liegen.") + return num + + +def validate_username(value: str) -> str: + value = value.strip() + if not _USERNAME_RE.fullmatch(value): + raise ValidationError("username", "Ungültiger Benutzername.") + return value + + +def validate_secret(value: str) -> str: + """Passwörter dürfen alles enthalten außer Zeilenumbrüchen und NUL – sie landen + zeilenweise in der Credentials-Datei von ``mount.cifs``.""" + if any(c in value for c in "\r\n\0"): + raise ValidationError("password", "Passwort darf keine Zeilenumbrüche enthalten.") + return value + + +@dataclass(frozen=True) +class MountOption: + key: str + value: str | None = None + + def render(self) -> str: + return self.key if self.value is None else f"{self.key}={self.value}" + + +def parse_extra_options(mount_type: MountType, raw: str) -> list[MountOption]: + """Zerlegt „vers=3.0,noserverino“ und prüft jede Option gegen die Allowlist.""" + options: list[MountOption] = [] + seen: set[str] = set() + for chunk in raw.replace("\n", ",").split(","): + chunk = chunk.strip() + if not chunk: + continue + key, sep, value = chunk.partition("=") + key = key.strip().lower() + val: str | None = value.strip() if sep else None + if not _OPT_KEY_RE.fullmatch(key): + raise ValidationError("extra_options", f"Ungültiger Optionsname: {key!r}") + if key in FORBIDDEN_OPTION_KEYS: + raise ValidationError( + "extra_options", + f"Option „{key}“ ist gesperrt – Zugangsdaten setzt NetMover selbst.", + ) + if key in DEDICATED_OPTION_KEYS: + raise ValidationError("extra_options", f"Option „{key}“ bitte im eigenen Feld angeben.") + if key not in ALLOWED_OPTIONS[mount_type]: + raise ValidationError( + "extra_options", + f"Option „{key}“ ist für {mount_type.value} nicht erlaubt.", + ) + if val is not None and not _OPT_VALUE_RE.fullmatch(val): + raise ValidationError( + "extra_options", + f"Ungültiger Wert für „{key}“ – erlaubt sind A–Z, 0–9 und . _ : / @ -", + ) + if key == "vers" and mount_type is MountType.NFS4 and val not in _NFS4_VERSIONS: + raise ValidationError("extra_options", "Für nfs4 ist nur vers=4, 4.0, 4.1, 4.2 gültig.") + if key in seen: + raise ValidationError("extra_options", f"Option „{key}“ ist doppelt angegeben.") + seen.add(key) + options.append(MountOption(key, val)) + return options + + +def slugify(name: str) -> str: + """Erzeugt den Verzeichnisnamen unter dem Mount-Basisverzeichnis.""" + text = name.replace("ß", "ss") + text = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode() + text = re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") + text = text[:48].strip("-") + if not text: + raise ValidationError( + "name", "Name muss mindestens einen Buchstaben oder eine Ziffer haben." + ) + return text + + +def safe_relpath(field: str, value: str) -> str: + """Relativer Unterpfad für Regeln. Leerzeichen sind erlaubt (die Pfade gehen nie + durch eine Shell), Ausbrüche per ``..`` oder absolute Pfade nicht.""" + value = value.strip().strip("/") + if "\0" in value: + raise ValidationError(field, "Ungültiges Zeichen im Pfad.") + if value == "": + return "" + _no_dot_segments(field, value) + return str(PurePosixPath(value)) + + +def safe_abspath(field: str, value: str) -> str: + value = value.strip() + if "\0" in value or not value.startswith("/"): + raise ValidationError(field, "Bitte einen absoluten Pfad angeben.") + _no_dot_segments(field, value) + return str(PurePosixPath(value)) diff --git a/tests/test_crypto.py b/tests/test_crypto.py new file mode 100644 index 0000000..9372640 --- /dev/null +++ b/tests/test_crypto.py @@ -0,0 +1,67 @@ +"""Fallstrick 7: Schlüsseldatei, Verschlüsselung, Schwärzung, SSH-Schlüssel.""" + +from __future__ import annotations + +import io +import stat +from pathlib import Path + +import paramiko +import pytest +from cryptography.fernet import Fernet + +from netmover.crypto import ( + REDACTED, + SecretBox, + SecretError, + generate_ssh_keypair, + load_or_create_key, + redact, +) + + +def test_key_created_with_0600_and_reused(tmp_path: Path) -> None: + path = tmp_path / "lib" / "secret.key" + key1 = load_or_create_key(path) + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + assert load_or_create_key(path) == key1 + + +def test_key_mode_is_tightened(tmp_path: Path) -> None: + path = tmp_path / "secret.key" + path.write_bytes(Fernet.generate_key()) + path.chmod(0o644) + load_or_create_key(path) + assert stat.S_IMODE(path.stat().st_mode) == 0o600 + + +def test_invalid_key_file(tmp_path: Path) -> None: + path = tmp_path / "secret.key" + path.write_text("kaputt") + path.chmod(0o600) + with pytest.raises(SecretError): + load_or_create_key(path) + + +def test_roundtrip_and_wrong_key() -> None: + box = SecretBox(Fernet.generate_key()) + token = box.encrypt("geheim") + assert "geheim" not in token + assert box.decrypt(token) == "geheim" + with pytest.raises(SecretError, match=r"secret\.key"): + SecretBox(Fernet.generate_key()).decrypt(token) + + +def test_redact() -> None: + text = "mount error: user scanner pw Sommer2026! failed (Sommer2026)" + out = redact(text, ["Sommer2026!", "Sommer2026", None, ""]) + assert "Sommer2026" not in out + assert out.count(REDACTED) == 2 + + +def test_ssh_keypair_usable_by_paramiko() -> None: + private, public = generate_ssh_keypair("netmover@test") + key = paramiko.Ed25519Key.from_private_key(io.StringIO(private)) + assert public.startswith("ssh-ed25519 ") + assert public.split()[1] == key.get_base64() + assert public.endswith(" netmover@test") diff --git a/tests/test_validation.py b/tests/test_validation.py new file mode 100644 index 0000000..5103085 --- /dev/null +++ b/tests/test_validation.py @@ -0,0 +1,159 @@ +"""Fallstrick 5 (Shell-Injection) und 6 (Mountpoints) – Eingabeprüfung.""" + +from __future__ import annotations + +import pytest + +from netmover.models import MountType +from netmover.validation import ( + ValidationError, + parse_extra_options, + safe_abspath, + safe_relpath, + slugify, + validate_mode, + validate_secret, + validate_server, + validate_share, +) + +INJECTION = [ + "nas;rm -rf /", + "nas && reboot", + "nas|id", + "$(id)", + "`id`", + "nas\nid", + "nas,credentials=/etc/shadow", + "nas name", + "-oProxyCommand=x", + "", + "nas/../etc", +] + + +@pytest.mark.parametrize("value", INJECTION) +def test_server_rejects_injection(value: str) -> None: + with pytest.raises(ValidationError) as exc: + validate_server(value) + assert exc.value.field == "server" + + +@pytest.mark.parametrize("value", ["192.168.1.10", "nas", "nas.fritz.box", "ds-920"]) +def test_server_accepts_hosts(value: str) -> None: + assert validate_server(value) == value + + +def test_server_rejects_ipv6_with_clear_message() -> None: + with pytest.raises(ValidationError, match="IPv6"): + validate_server("fe80::1") + + +@pytest.mark.parametrize( + "value", + ["scans,uid=0", "scans $HOME", "scans`id`", "scans\nfoo", "../etc", "sc ans", "a/../b", "C$"], +) +def test_cifs_share_rejects(value: str) -> None: + with pytest.raises(ValidationError): + validate_share(MountType.CIFS, value) + + +def test_cifs_share_normalises_slashes() -> None: + assert validate_share(MountType.CIFS, "/scans/eingang/") == "scans/eingang" + + +@pytest.mark.parametrize("value", ["volume1/scans", "/volume1/scans,rw", "/a b", "/a/../b", "/$x"]) +def test_nfs_export_rejects(value: str) -> None: + with pytest.raises(ValidationError): + validate_share(MountType.NFS4, value) + + +def test_nfs_export_accepts() -> None: + assert validate_share(MountType.NFS3, "/volume1/scans/") == "/volume1/scans" + + +@pytest.mark.parametrize("key", ["credentials", "username", "password", "pass", "user", "domain"]) +def test_forbidden_option_keys(key: str) -> None: + with pytest.raises(ValidationError, match="gesperrt"): + parse_extra_options(MountType.CIFS, f"{key}=x") + + +@pytest.mark.parametrize("key", ["uid", "gid", "file_mode", "dir_mode"]) +def test_dedicated_option_keys(key: str) -> None: + with pytest.raises(ValidationError, match="eigenen Feld"): + parse_extra_options(MountType.CIFS, f"{key}=1000") + + +@pytest.mark.parametrize( + "raw", + [ + "vers=3.0;id", + "vers=$(id)", + "vers=`id`", + "vers=3 0", + "iocharset=utf8,x-systemd.automount", + "exec", # nicht in der Allowlist + "VERS=3.0=1", + ], +) +def test_option_values_and_unknown_keys(raw: str) -> None: + with pytest.raises(ValidationError): + parse_extra_options(MountType.CIFS, raw) + + +def test_options_allowlist_per_type() -> None: + with pytest.raises(ValidationError, match="nicht erlaubt"): + parse_extra_options(MountType.NFS4, "nolock") + assert [o.render() for o in parse_extra_options(MountType.NFS3, "nolock, timeo=600")] == [ + "nolock", + "timeo=600", + ] + + +def test_options_parse_ok_and_duplicates() -> None: + opts = parse_extra_options(MountType.CIFS, "vers=3.0, noserverino,\niocharset=utf8") + assert [o.render() for o in opts] == ["vers=3.0", "noserverino", "iocharset=utf8"] + with pytest.raises(ValidationError, match="doppelt"): + parse_extra_options(MountType.CIFS, "vers=3.0,vers=2.1") + + +def test_nfs4_vers_restricted() -> None: + assert parse_extra_options(MountType.NFS4, "vers=4.1")[0].value == "4.1" + with pytest.raises(ValidationError): + parse_extra_options(MountType.NFS4, "vers=3") + + +def test_mode() -> None: + assert validate_mode("file_mode", "644") == "0644" + assert validate_mode("file_mode", "") is None + with pytest.raises(ValidationError): + validate_mode("file_mode", "0999") + + +def test_secret_rejects_newlines() -> None: + assert validate_secret("p@ss,w=rd;$`") == "p@ss,w=rd;$`" + with pytest.raises(ValidationError): + validate_secret("pass\nusername=root") + + +def test_slugify() -> None: + assert slugify("Synology Scans (Büro)") == "synology-scans-buro" + assert slugify("Straße/../..") == "strasse" + with pytest.raises(ValidationError): + slugify("../..") + + +def test_relpath() -> None: + assert safe_relpath("p", "/Scans Max/eingang/") == "Scans Max/eingang" + assert safe_relpath("p", "") == "" + for bad in ["../x", "a/../../b", "a/./b", "a\0b"]: + with pytest.raises(ValidationError): + safe_relpath("p", bad) + + +def test_abspath() -> None: + assert safe_abspath("p", "/opt/consume/") == "/opt/consume" + with pytest.raises(ValidationError): + safe_abspath("p", "opt/consume") + with pytest.raises(ValidationError): + safe_abspath("p", "/opt/../etc")