diff --git a/src/netmover/admin.py b/src/netmover/admin.py new file mode 100644 index 0000000..a1e4444 --- /dev/null +++ b/src/netmover/admin.py @@ -0,0 +1,543 @@ +"""Verwaltung von Anbindungen und Regeln – Prüfen und Speichern von Formulardaten. + +Unabhängig vom Web-Framework: nimmt ein ``Mapping[str, str]`` (Formular) und +liefert entweder gespeicherte Objekte oder ein Fehler-Dict ``{feld: meldung}``. + +Passwörter: leer lassen heißt „unverändert“; sie werden nie zurückgegeben. +""" + +from __future__ import annotations + +import re +from collections.abc import Callable, Mapping +from dataclasses import dataclass, field + +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Session + +from netmover.crypto import SecretBox, generate_ssh_keypair +from netmover.models import ( + CollisionPolicy, + Credential, + CredentialKind, + Mount, + MountState, + MountType, + PatternType, + Rule, + RuleAction, +) +from netmover.mounting.manager import CredentialData, MountConfig, MountManager +from netmover.validation import ( + MountOption, + ValidationError, + parse_extra_options, + safe_abspath, + safe_relpath, + slugify, + validate_id, + validate_mode, + validate_port, + validate_secret, + validate_server, + validate_share, + validate_username, +) + +Errors = dict[str, str] + +MIN_INTERVAL = 10 +_OWNER_RE = re.compile(r"^\d{1,10}(:\d{1,10})?$") + + +class AdminError(Exception): + """Aktion nicht möglich – Meldung für die Oberfläche.""" + + +def _bool(form: Mapping[str, str], key: str) -> bool: + return form.get(key, "") in ("on", "true", "1", "yes") + + +def _int( + form: Mapping[str, str], + key: str, + errors: Errors, + *, + default: int | None = None, + minimum: int | None = None, + label: str = "", +) -> int | None: + raw = (form.get(key) or "").strip() + if raw == "": + return default + try: + value = int(raw) + except ValueError: + errors[key] = "Bitte eine ganze Zahl angeben." + return default + if minimum is not None and value < minimum: + errors[key] = f"{label or 'Wert'} muss mindestens {minimum} sein." + return value + + +# -- Anbindungen ------------------------------------------------------------------------ + + +@dataclass +class MountInput: + name: str = "" + type: MountType = MountType.CIFS + server: str = "" + share: str = "" + port: int | None = None + username: str = "" + password: str = "" # leer = unverändert + domain: str = "" + auth: CredentialKind = CredentialKind.PASSWORD # nur sftp: Schlüssel oder Passwort + uid: int | None = None + gid: int | None = None + file_mode: str | None = None + dir_mode: str | None = None + extra_options: str = "" + auto_mount: bool = True + enabled: bool = True + + def config(self) -> MountConfig: + return MountConfig( + name=self.name or "neu", + type=self.type, + server=self.server, + share=self.share, + port=self.port, + uid=self.uid, + gid=self.gid, + file_mode=self.file_mode, + dir_mode=self.dir_mode, + extra_options=self.extra_options, + ) + + +def mount_input_from_model(m: Mount) -> MountInput: + c = m.credential + return MountInput( + name=m.name, + type=m.type, + server=m.server, + share=m.share, + port=m.port, + username=c.username if c else "", + domain=(c.domain or "") if c else "", + auth=c.kind if c else CredentialKind.PASSWORD, + uid=m.uid, + gid=m.gid, + file_mode=m.file_mode, + dir_mode=m.dir_mode, + extra_options=m.extra_options, + auto_mount=m.auto_mount, + enabled=m.enabled, + ) + + +def parse_mount_form(form: Mapping[str, str]) -> tuple[MountInput, Errors]: + errors: Errors = {} + data = MountInput( + name=(form.get("name") or "").strip(), + server=(form.get("server") or "").strip(), + share=(form.get("share") or "").strip(), + username=(form.get("username") or "").strip(), + password=form.get("password") or "", + domain=(form.get("domain") or "").strip(), + extra_options=(form.get("extra_options") or "").strip(), + auto_mount=_bool(form, "auto_mount"), + enabled=_bool(form, "enabled"), + ) + try: + data.type = MountType(form.get("type", "cifs")) + except ValueError: + errors["type"] = "Unbekannter Typ." + return data, errors + try: + data.auth = CredentialKind(form.get("auth") or "password") + except ValueError: + errors["auth"] = "Unbekannte Anmeldeart." + + if not data.name: + errors["name"] = "Name fehlt." + elif len(data.name) > 100: + errors["name"] = "Höchstens 100 Zeichen." + else: + try: + slugify(data.name) + except ValidationError as exc: + errors["name"] = str(exc) + + def check[T](key: str, fn: Callable[[], T], fallback: T) -> T: + try: + return fn() + except ValidationError as exc: + errors[key] = str(exc) + return fallback + + data.server = check("server", lambda: validate_server(data.server), data.server) + data.share = check("share", lambda: validate_share(data.type, data.share), data.share) + no_options: list[MountOption] = [] + check("extra_options", lambda: parse_extra_options(data.type, data.extra_options), no_options) + check("password", lambda: validate_secret(data.password), "") + if data.type is MountType.CIFS: + data.uid = check("uid", lambda: validate_id("uid", form.get("uid")), None) + data.gid = check("gid", lambda: validate_id("gid", form.get("gid")), None) + data.file_mode = check( + "file_mode", lambda: validate_mode("file_mode", form.get("file_mode")), None + ) + data.dir_mode = check( + "dir_mode", lambda: validate_mode("dir_mode", form.get("dir_mode")), None + ) + if data.type is MountType.SFTP: + data.port = check("port", lambda: validate_port(form.get("port")), None) + if data.username: + data.username = check("username", lambda: validate_username(data.username), data.username) + if data.domain: + check("domain", lambda: validate_username(data.domain), "") + if data.type is MountType.SFTP and not data.username: + errors["username"] = "SFTP braucht einen Benutzernamen." + if data.type is not MountType.SFTP: + data.auth = CredentialKind.PASSWORD + return data, errors + + +def credential_for_test( + data: MountInput, existing: Mount | None, secrets: SecretBox +) -> CredentialData | None: + """Zugangsdaten für „Testen“ aus dem Formular – leeres Passwort = gespeichertes.""" + if data.type in (MountType.NFS3, MountType.NFS4) or not data.username: + return None + secret = data.password + stored = existing.credential if existing is not None else None + if not secret and stored is not None and stored.kind is data.auth: + secret = secrets.decrypt(stored.secret_enc) + if not secret: + return None + return CredentialData(data.auth, data.username, secret, data.domain or None) + + +_CONNECTION_FIELDS = ( + "type", + "server", + "share", + "port", + "uid", + "gid", + "file_mode", + "dir_mode", + "extra_options", +) + + +def save_mount( + s: Session, + secrets: SecretBox, + mounts: MountManager, + data: MountInput, + mount_id: int | None = None, +) -> tuple[Mount | None, Errors]: + errors: Errors = {} + m = s.get(Mount, mount_id) if mount_id is not None else None + if mount_id is not None and m is None: + raise AdminError("Anbindung existiert nicht.") + dup = s.scalar(select(Mount.id).where(Mount.name == data.name)) + if dup is not None and dup != mount_id: + errors["name"] = "Name ist bereits vergeben." + return None, errors + + if m is None: + base = slugify(data.name) + slug, n = base, 1 + while s.scalar(select(Mount.id).where(Mount.slug == slug)) is not None: + n += 1 + slug = f"{base[:44]}-{n}" + m = Mount(slug=slug, state=MountState.UNMOUNTED, extra_options="") + if data.type.is_kernel_mount: + m.mountpoint = str(mounts.mountpoint_for(slug)) + s.add(m) + else: + changed = [f for f in _CONNECTION_FIELDS if getattr(m, f) != getattr(data, f)] + if ( + changed + and m.type.is_kernel_mount + and m.state in (MountState.MOUNTED, MountState.DEGRADED) + ): + errors["_"] = ( + "Verbindungsdaten lassen sich nur ändern, wenn die Anbindung ausgehängt ist." + ) + return None, errors + if m.type is MountType.SFTP and (m.server != data.server or m.port != data.port): + m.host_key = None # anderer Server → Hostschlüssel neu bestätigen + if data.type.is_kernel_mount and not m.mountpoint: + m.mountpoint = str(mounts.mountpoint_for(m.slug)) + if not data.type.is_kernel_mount: + m.mountpoint = None + + for f in ( + "name", + "type", + "server", + "share", + "port", + "uid", + "gid", + "file_mode", + "dir_mode", + "extra_options", + "auto_mount", + "enabled", + ): + setattr(m, f, getattr(data, f)) + if not data.type.is_kernel_mount: + m.auto_mount = False + + # Zugangsdaten: jede Anbindung pflegt ihre eigenen. + cred = m.credential + needs_cred = data.type is MountType.SFTP or ( + data.type is MountType.CIFS and bool(data.username) + ) + if not needs_cred: + if cred is not None: + m.credential = None + s.flush() + if not cred.mounts: + s.delete(cred) + else: + kind_changed = cred is not None and cred.kind is not data.auth + if cred is None or kind_changed: + if data.auth is CredentialKind.SSH_KEY: + private, public = generate_ssh_keypair(f"netmover@{m.slug}") + secret, pub = private, public + else: + if not data.password: + errors["password"] = "Passwort fehlt." # noqa: S105 + s.rollback() + return None, errors + secret, pub = data.password, None + if cred is None: + cred = Credential(name=f"mount:{m.slug}") + s.add(cred) + m.credential = cred + cred.kind = data.auth + cred.secret_enc = secrets.encrypt(secret) + cred.public_key = pub + elif data.password and data.auth is CredentialKind.PASSWORD: + cred.secret_enc = secrets.encrypt(data.password) + cred.username = data.username + cred.domain = data.domain or None + try: + s.flush() + except IntegrityError: + s.rollback() + errors["_"] = "Speichern fehlgeschlagen (Name oder Mountpoint doppelt)." + return None, errors + return m, errors + + +def delete_mount(s: Session, mounts: MountManager, mount_id: int) -> None: + m = s.get(Mount, mount_id) + if m is None: + return + used = s.scalars( + select(Rule.name).where((Rule.src_mount_id == mount_id) | (Rule.dst_mount_id == mount_id)) + ).all() + if used: + raise AdminError(f"Wird noch von Regeln benutzt: {', '.join(used)}") + if m.type.is_kernel_mount and m.state is not MountState.UNMOUNTED: + mounts.unmount(mount_id) + cred = m.credential + s.delete(m) + s.flush() + if cred is not None and not cred.mounts: + s.delete(cred) + + +# -- Regeln ---------------------------------------------------------------------------------- + + +@dataclass +class RuleInput: + name: str = "" + enabled: bool = True + src_mount_id: int | None = None + src_subpath: str = "" + dst_kind: str = "mount" # "mount" | "local" + dst_mount_id: int | None = None + dst_path: str = "" + pattern: str = "*" + pattern_type: PatternType = PatternType.GLOB + recursive: bool = False + interval_seconds: int = 30 + stability_seconds: int = 5 + min_age_seconds: int = 0 + max_size_mb: int | None = None + op_timeout_seconds: int = 120 + action: RuleAction = RuleAction.MOVE + on_collision: CollisionPolicy = CollisionPolicy.RENAME + set_owner: str = "" + set_mode: str = "" + dry_run: bool = False + quarantine_path: str = "" + quarantine_after: int = 3 + extra: dict[str, str] = field(default_factory=dict) + + +def rule_input_from_model(r: Rule) -> RuleInput: + return RuleInput( + name=r.name, + enabled=r.enabled, + src_mount_id=r.src_mount_id, + src_subpath=r.src_subpath, + dst_kind="mount" if r.dst_mount_id else "local", + dst_mount_id=r.dst_mount_id, + dst_path=r.dst_path, + pattern=r.pattern, + pattern_type=r.pattern_type, + recursive=r.recursive, + interval_seconds=r.interval_seconds, + stability_seconds=r.stability_seconds, + min_age_seconds=r.min_age_seconds, + max_size_mb=r.max_size_mb, + op_timeout_seconds=r.op_timeout_seconds, + action=r.action, + on_collision=r.on_collision, + set_owner=r.set_owner or "", + set_mode=r.set_mode or "", + dry_run=r.dry_run, + quarantine_path=r.quarantine_path or "", + quarantine_after=r.quarantine_after, + ) + + +def parse_rule_form(s: Session, form: Mapping[str, str]) -> tuple[RuleInput, Errors]: + errors: Errors = {} + d = RuleInput( + name=(form.get("name") or "").strip(), + enabled=_bool(form, "enabled"), + src_subpath=form.get("src_subpath") or "", + dst_kind=form.get("dst_kind") or "mount", + dst_path=(form.get("dst_path") or "").strip(), + pattern=(form.get("pattern") or "").strip() or "*", + recursive=_bool(form, "recursive"), + set_owner=(form.get("set_owner") or "").strip(), + set_mode=(form.get("set_mode") or "").strip(), + dry_run=_bool(form, "dry_run"), + quarantine_path=form.get("quarantine_path") or "", + ) + if not d.name: + errors["name"] = "Name fehlt." + for key, enum_cls in ( + ("pattern_type", PatternType), + ("action", RuleAction), + ("on_collision", CollisionPolicy), + ): + try: + setattr(d, key, enum_cls(form.get(key) or getattr(d, key).value)) + except ValueError: + errors[key] = "Ungültiger Wert." + d.src_mount_id = _int(form, "src_mount_id", errors) + d.dst_mount_id = _int(form, "dst_mount_id", errors) if d.dst_kind == "mount" else None + d.interval_seconds = ( + _int(form, "interval_seconds", errors, default=30, minimum=MIN_INTERVAL, label="Intervall") + or 30 + ) + d.stability_seconds = _int(form, "stability_seconds", errors, default=5, minimum=0) or 0 + d.min_age_seconds = _int(form, "min_age_seconds", errors, default=0, minimum=0) or 0 + d.max_size_mb = _int(form, "max_size_mb", errors, minimum=1) + d.op_timeout_seconds = ( + _int(form, "op_timeout_seconds", errors, default=120, minimum=10, label="Zeitlimit") or 120 + ) + d.quarantine_after = _int(form, "quarantine_after", errors, default=3, minimum=1) or 3 + + src = s.get(Mount, d.src_mount_id) if d.src_mount_id else None + if src is None: + errors["src_mount_id"] = "Quelle wählen." + elif not src.type.is_kernel_mount: + errors["src_mount_id"] = "SFTP-Anbindungen sind nur als Ziel möglich." + try: + d.src_subpath = safe_relpath("src_subpath", d.src_subpath) + except ValidationError as exc: + errors["src_subpath"] = str(exc) + if d.dst_kind == "mount": + if d.dst_mount_id is None or s.get(Mount, d.dst_mount_id) is None: + errors["dst_mount_id"] = "Ziel wählen." + try: + d.dst_path = safe_relpath("dst_path", d.dst_path) + except ValidationError as exc: + errors["dst_path"] = str(exc) + if ( + d.dst_mount_id == d.src_mount_id + and "src_subpath" not in errors + and d.dst_path == d.src_subpath + ): + errors["dst_path"] = "Quelle und Ziel sind derselbe Ordner." + elif d.dst_kind == "local": + try: + d.dst_path = safe_abspath("dst_path", d.dst_path) + except ValidationError as exc: + errors["dst_path"] = str(exc) + else: + errors["dst_kind"] = "Ungültige Zielart." + + if d.pattern_type is PatternType.REGEX: + try: + re.compile(d.pattern) + except re.error as exc: + errors["pattern"] = f"Ungültiger regulärer Ausdruck: {exc}" + if d.stability_seconds >= d.op_timeout_seconds: + errors["stability_seconds"] = "Muss kleiner als das Zeitlimit sein." + if d.set_owner and not _OWNER_RE.fullmatch(d.set_owner): + errors["set_owner"] = "Als „uid:gid“ angeben, z. B. „1000:1000“." + try: + d.set_mode = validate_mode("set_mode", d.set_mode) or "" + except ValidationError as exc: + errors["set_mode"] = str(exc) + if d.quarantine_path: + try: + d.quarantine_path = safe_relpath("quarantine_path", d.quarantine_path) + except ValidationError as exc: + errors["quarantine_path"] = str(exc) + return d, errors + + +def save_rule(s: Session, d: RuleInput, rule_id: int | None = None) -> tuple[Rule | None, Errors]: + dup = s.scalar(select(Rule.id).where(Rule.name == d.name)) + if dup is not None and dup != rule_id: + return None, {"name": "Name ist bereits vergeben."} + r = s.get(Rule, rule_id) if rule_id is not None else None + if rule_id is not None and r is None: + raise AdminError("Regel existiert nicht.") + if r is None: + r = Rule() + s.add(r) + for f in ( + "name", + "enabled", + "src_mount_id", + "src_subpath", + "dst_mount_id", + "dst_path", + "pattern", + "pattern_type", + "recursive", + "interval_seconds", + "stability_seconds", + "min_age_seconds", + "max_size_mb", + "op_timeout_seconds", + "action", + "on_collision", + "dry_run", + "quarantine_after", + ): + setattr(r, f, getattr(d, f)) + r.set_owner = d.set_owner or None + r.set_mode = d.set_mode or None + r.quarantine_path = d.quarantine_path or None + s.flush() + return r, {} diff --git a/src/netmover/transfer.py b/src/netmover/transfer.py new file mode 100644 index 0000000..bfdb2e4 --- /dev/null +++ b/src/netmover/transfer.py @@ -0,0 +1,196 @@ +"""Export/Import der Konfiguration als JSON – ohne Zugangsdaten. + +Ausgenommen sind Passwörter, private Schlüssel, Benutzernamen und die +Webhook-URL (enthält meist ein Token). Nach einem Import müssen Zugangsdaten +neu eingegeben werden. Anbindungen und Regeln werden über ihren Namen +zugeordnet; vorhandene Namen werden übersprungen, nie überschrieben. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from netmover import __version__ +from netmover.models import ( + CollisionPolicy, + Mount, + MountState, + MountType, + PatternType, + Rule, + RuleAction, +) +from netmover.mounting.manager import MountManager +from netmover.settings_store import AppSettings, NotifyMode +from netmover.validation import ( + ValidationError, + parse_extra_options, + safe_relpath, + slugify, + validate_server, + validate_share, +) + +FORMAT_VERSION = 1 + +MOUNT_FIELDS = ( + "name", + "type", + "server", + "share", + "port", + "host_key", + "uid", + "gid", + "file_mode", + "dir_mode", + "extra_options", + "auto_mount", + "enabled", +) +RULE_FIELDS = ( + "name", + "enabled", + "src_subpath", + "dst_path", + "pattern", + "pattern_type", + "recursive", + "interval_seconds", + "stability_seconds", + "min_age_seconds", + "max_size_mb", + "op_timeout_seconds", + "action", + "on_collision", + "set_owner", + "set_mode", + "dry_run", + "quarantine_path", + "quarantine_after", +) +SETTING_FIELDS = ( + "retention_days", + "notify_mode", + "notify_after", + "notify_recovery", + "webhook_template", + "webhook_content_type", +) + + +def _plain(value: Any) -> Any: + return value.value if hasattr(value, "value") else value + + +def export_config(s: Session, settings: AppSettings) -> dict[str, Any]: + mounts = s.scalars(select(Mount).order_by(Mount.name)).all() + names = {m.id: m.name for m in mounts} + return { + "format": FORMAT_VERSION, + "netmover": __version__, + "mounts": [{f: _plain(getattr(m, f)) for f in MOUNT_FIELDS} for m in mounts], + "rules": [ + { + **{f: _plain(getattr(r, f)) for f in RULE_FIELDS}, + "src_mount": names[r.src_mount_id], + "dst_mount": names.get(r.dst_mount_id) if r.dst_mount_id else None, + } + for r in s.scalars(select(Rule).order_by(Rule.name)) + ], + "settings": {f: _plain(getattr(settings, f)) for f in SETTING_FIELDS}, + } + + +@dataclass +class ImportReport: + created: list[str] = field(default_factory=list) + skipped: list[str] = field(default_factory=list) + errors: list[str] = field(default_factory=list) + settings: dict[str, Any] = field(default_factory=dict) + + +def import_config(s: Session, mounts: MountManager, data: dict[str, Any]) -> ImportReport: + report = ImportReport() + if not isinstance(data, dict) or data.get("format") != FORMAT_VERSION: + report.errors.append("Unbekanntes Format – erwartet wird ein NetMover-Export (format 1).") + return report + + for raw in data.get("mounts") or []: + name = str(raw.get("name", "")) + if s.scalar(select(Mount.id).where(Mount.name == name)) is not None: + report.skipped.append(f"Anbindung „{name}“ (existiert)") + continue + try: + mtype = MountType(raw["type"]) + slug = slugify(name) + if s.scalar(select(Mount.id).where(Mount.slug == slug)) is not None: + raise ValidationError("name", "Slug bereits vergeben") + m = Mount( + name=name, + slug=slug, + type=mtype, + server=validate_server(str(raw["server"])), + share=validate_share(mtype, str(raw["share"])), + port=raw.get("port"), + host_key=raw.get("host_key"), + uid=raw.get("uid"), + gid=raw.get("gid"), + file_mode=raw.get("file_mode"), + dir_mode=raw.get("dir_mode"), + extra_options=str(raw.get("extra_options") or ""), + # ohne Zugangsdaten nicht automatisch einhängen + auto_mount=bool(raw.get("auto_mount")) and mtype is not MountType.CIFS, + enabled=bool(raw.get("enabled", True)), + state=MountState.UNMOUNTED, + mountpoint=str(mounts.mountpoint_for(slug)) if mtype.is_kernel_mount else None, + ) + parse_extra_options(mtype, m.extra_options) + s.add(m) + s.flush() + report.created.append(f"Anbindung „{name}“") + except (KeyError, ValueError, ValidationError) as exc: + report.errors.append(f"Anbindung „{name}“: {exc}") + + ids = {name: mid for mid, name in s.execute(select(Mount.id, Mount.name)).all()} + for raw in data.get("rules") or []: + name = str(raw.get("name", "")) + if s.scalar(select(Rule.id).where(Rule.name == name)) is not None: + report.skipped.append(f"Regel „{name}“ (existiert)") + continue + try: + src = ids[raw["src_mount"]] + dst = ids[raw["dst_mount"]] if raw.get("dst_mount") else None + r = Rule(src_mount_id=src, dst_mount_id=dst) + for f in RULE_FIELDS: + if f in raw: + setattr(r, f, raw[f]) + r.pattern_type = PatternType(r.pattern_type) + r.action = RuleAction(r.action) + r.on_collision = CollisionPolicy(r.on_collision) + r.src_subpath = safe_relpath("src_subpath", r.src_subpath or "") + if int(r.interval_seconds) < 10: + raise ValueError("Intervall < 10 s") + r.enabled = False # erst nach Eingabe der Zugangsdaten aktivieren + s.add(r) + s.flush() + report.created.append(f"Regel „{name}“ (deaktiviert)") + except KeyError as exc: + report.errors.append(f"Regel „{name}“: Anbindung {exc} fehlt") + except (ValueError, ValidationError) as exc: + report.errors.append(f"Regel „{name}“: {exc}") + + if isinstance(data.get("settings"), dict): + report.settings = {f: data["settings"][f] for f in SETTING_FIELDS if f in data["settings"]} + return report + + +def apply_imported_settings(current: AppSettings, values: dict[str, Any]) -> AppSettings: + """Nach dem Commit des Imports aufrufen (eigene Transaktion).""" + for key, value in values.items(): + setattr(current, key, NotifyMode(value) if key == "notify_mode" else value) + return current