feat: Verwaltung von Anbindungen/Regeln und Export/Import
Formularprüfung und Speichern unabhängig vom Web-Framework; Passwörter leer = unverändert, SFTP-Schlüsselpaar wird erzeugt. Export/Import als JSON ohne Zugangsdaten und Webhook-URL. 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
f008c216be
commit
e2587551b0
@@ -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, {}
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user