feat: Eingabeprüfung, Verschlüsselung und Zeitlimits
Strikte Validierung von Server, Freigabe, Pfaden und Mount-Optionen (Allowlist, gesperrte Zugangsdaten-Schlüssel), Fernet-Schlüssel mit 0600, Ed25519-Schlüsselerzeugung, Schwärzung von Geheimnissen und Dateizugriffe mit Zeitlimit in Daemon-Threads. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Florian Wilde
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
278180c541
commit
689ddf0d94
@@ -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
|
||||
@@ -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]
|
||||
@@ -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))
|
||||
@@ -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")
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user