Florian WildeandClaude Opus 5.5
f5ea5840d4
Seite „Zugangsdaten“ (vorher „Schlüssel“) mit Passwörtern und SSH-Schlüsseln; Anbindungen wählen gespeicherte Passwörter aus, ein neues Passwort gilt sofort für alle. /keys leitet auf /credentials um. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
624 lines
23 KiB
Python
624 lines
23 KiB
Python
"""Weboberfläche: Einrichtung, Anmeldung, CSRF, Seiten und Aktionen (FakeMountBackend)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import select
|
||
|
||
from netmover.app_state import NetMover
|
||
from netmover.config import Settings
|
||
from netmover.db import session_scope
|
||
from netmover.models import Credential, Mount, MountState, Rule, Run, RunStatus
|
||
from netmover.mounting.fake import FakeMountBackend
|
||
from netmover.web.app import create_app
|
||
from tests.webutil import PASSWORD, csrf_of, do_setup, post
|
||
|
||
# -- Einrichtung und Anmeldung --------------------------------------------------------------
|
||
|
||
|
||
def test_first_start_forces_setup(raw_client: TestClient) -> None:
|
||
for url in ("/", "/mounts", "/rules", "/log", "/settings", "/login"):
|
||
r = raw_client.get(url)
|
||
assert r.status_code == 303 and r.headers["location"] == "/setup", url
|
||
|
||
|
||
def test_setup_requires_code_and_strong_password(raw_client: TestClient) -> None:
|
||
token = csrf_of(raw_client.get("/setup").text)
|
||
code = raw_client.app.state.setup_code.code # type: ignore[attr-defined]
|
||
bad = [
|
||
{"code": "falsch", "password": PASSWORD, "confirm": PASSWORD},
|
||
{"code": code, "password": "kurz", "confirm": "kurz"},
|
||
{"code": code, "password": PASSWORD, "confirm": PASSWORD + "x"},
|
||
]
|
||
for data in bad:
|
||
r = raw_client.post("/setup", data={"csrf_token": token, **data})
|
||
assert r.status_code == 400
|
||
assert raw_client.get("/").status_code == 303
|
||
|
||
|
||
def test_setup_only_once(client: TestClient) -> None:
|
||
assert client.get("/setup").headers["location"] == "/login"
|
||
|
||
|
||
def test_login_logout_and_cookie_flags(raw_client: TestClient) -> None:
|
||
do_setup(raw_client)
|
||
token = csrf_of(raw_client.get("/").text)
|
||
raw_client.post("/logout", data={"csrf_token": token})
|
||
assert raw_client.get("/").status_code == 303
|
||
page = raw_client.get("/login")
|
||
r = raw_client.post("/login", data={"csrf_token": csrf_of(page.text), "password": "falsch"})
|
||
assert r.status_code == 401
|
||
r = raw_client.post("/login", data={"csrf_token": csrf_of(page.text), "password": PASSWORD})
|
||
assert r.status_code == 303
|
||
cookie = r.headers["set-cookie"].lower()
|
||
assert "httponly" in cookie and "samesite=lax" in cookie
|
||
assert raw_client.get("/").status_code == 200
|
||
|
||
|
||
def test_login_lockout(raw_client: TestClient) -> None:
|
||
do_setup(raw_client)
|
||
raw_client.cookies.clear()
|
||
token = csrf_of(raw_client.get("/login").text)
|
||
codes = [
|
||
raw_client.post("/login", data={"csrf_token": token, "password": "x"}).status_code
|
||
for _ in range(6)
|
||
]
|
||
assert codes[:5] == [401] * 5 and codes[5] == 429
|
||
|
||
|
||
def test_csrf_required_for_post(client: TestClient) -> None:
|
||
assert client.post("/rules/1/toggle").status_code == 403
|
||
assert client.post("/rules/1/toggle", data={"csrf_token": "falsch"}).status_code == 403
|
||
assert post(client, "/rules/1/toggle").status_code == 303
|
||
r = client.post("/rules/1/toggle", headers={"X-CSRF-Token": client.headers["X-Test-CSRF"]})
|
||
assert r.status_code == 303
|
||
|
||
|
||
def test_htmx_request_without_login_gets_hx_redirect(raw_client: TestClient) -> None:
|
||
do_setup(raw_client)
|
||
raw_client.cookies.clear()
|
||
r = raw_client.get("/partials/status", headers={"HX-Request": "true"})
|
||
assert r.status_code == 204 and r.headers["HX-Redirect"].startswith("/login")
|
||
|
||
|
||
def test_security_headers(client: TestClient) -> None:
|
||
r = client.get("/")
|
||
assert "script-src 'self'" in r.headers["content-security-policy"]
|
||
assert r.headers["x-frame-options"] == "DENY"
|
||
assert r.headers["cache-control"] == "no-store"
|
||
|
||
|
||
def test_no_inline_scripts_or_styles(client: TestClient) -> None:
|
||
for url in (
|
||
"/",
|
||
"/mounts",
|
||
"/mounts/new",
|
||
"/keys",
|
||
"/rules",
|
||
"/rules/new",
|
||
"/log",
|
||
"/settings",
|
||
):
|
||
html = client.get(url).text
|
||
assert "style=" not in html, url
|
||
assert not re.search(r"<script(?![^>]*\bsrc=)", html), url
|
||
|
||
|
||
def test_healthz(client: TestClient, state: NetMover) -> None:
|
||
r = client.get("/healthz")
|
||
assert r.status_code == 200 and r.json()["status"] == "ok"
|
||
with session_scope(state.session_factory) as s:
|
||
s.add(
|
||
Mount(
|
||
name="kaputt",
|
||
slug="kaputt",
|
||
type="cifs",
|
||
server="nas",
|
||
share="x",
|
||
mountpoint="/mnt/netmover/kaputt",
|
||
state=MountState.ERROR,
|
||
last_error="weg",
|
||
)
|
||
)
|
||
r = client.get("/healthz")
|
||
assert r.status_code == 503 and "kaputt" in r.json()["problems"][0]
|
||
|
||
|
||
# -- Anbindungen ------------------------------------------------------------------------------
|
||
|
||
|
||
MOUNT_FORM = {
|
||
"name": "Synology Scans",
|
||
"type": "cifs",
|
||
"server": "nas",
|
||
"share": "scans",
|
||
"username": "scanner",
|
||
"password": "Pa$$w0rd,mit=Zeichen",
|
||
"uid": "1000",
|
||
"gid": "1000",
|
||
"file_mode": "0644",
|
||
"dir_mode": "0755",
|
||
"extra_options": "vers=3.0",
|
||
"auto_mount": "on",
|
||
"enabled": "on",
|
||
}
|
||
|
||
|
||
def _mount_id(state: NetMover, name: str) -> int:
|
||
with state.session_factory() as s:
|
||
mid = s.scalar(select(Mount.id).where(Mount.name == name))
|
||
assert mid is not None
|
||
return mid
|
||
|
||
|
||
def test_create_mount_test_mount_unmount_delete(
|
||
client: TestClient, state: NetMover, backend: FakeMountBackend
|
||
) -> None:
|
||
backend.export("nas", "scans")
|
||
r = post(client, "/mounts/new", MOUNT_FORM)
|
||
assert r.status_code == 303, r.text
|
||
mid = _mount_id(state, "Synology Scans")
|
||
with state.session_factory() as s:
|
||
m = s.get(Mount, mid)
|
||
assert m is not None and m.slug == "synology-scans" and m.credential is not None
|
||
assert "Pa$$" not in m.credential.secret_enc
|
||
assert state.secrets.decrypt(m.credential.secret_enc) == MOUNT_FORM["password"]
|
||
|
||
# Passwort taucht nirgends auf
|
||
for url in ("/mounts", f"/mounts/{mid}/edit"):
|
||
assert MOUNT_FORM["password"] not in client.get(url).text
|
||
|
||
r = post(client, f"/mounts/{mid}/test")
|
||
assert r.status_code == 200 and "Quelle mit Löschen" in r.text and "✓" in r.text
|
||
|
||
assert post(client, f"/mounts/{mid}/mount").status_code == 303
|
||
assert backend.is_mounted(state.mounts.mountpoint_for("synology-scans"))
|
||
spec = [c for c in backend.calls if c.action == "mount"][-1].spec
|
||
assert spec is not None and spec.credentials is not None
|
||
assert spec.credentials.password == MOUNT_FORM["password"]
|
||
assert "Fehler" not in client.get("/").text
|
||
|
||
assert post(client, f"/mounts/{mid}/unmount").status_code == 303
|
||
assert not backend.mounted
|
||
assert post(client, f"/mounts/{mid}/delete").status_code == 303
|
||
with state.session_factory() as s:
|
||
assert s.scalars(select(Mount)).all() == []
|
||
assert s.scalars(select(Credential)).all() == []
|
||
|
||
|
||
def test_empty_password_keeps_secret(client: TestClient, state: NetMover) -> None:
|
||
post(client, "/mounts/new", MOUNT_FORM)
|
||
mid = _mount_id(state, "Synology Scans")
|
||
r = post(client, f"/mounts/{mid}/edit", {**MOUNT_FORM, "password": "", "username": "neu"})
|
||
assert r.status_code == 303
|
||
with state.session_factory() as s:
|
||
m = s.get(Mount, mid)
|
||
assert m is not None and m.credential is not None
|
||
assert m.credential.username == "neu"
|
||
assert state.secrets.decrypt(m.credential.secret_enc) == MOUNT_FORM["password"]
|
||
|
||
|
||
def test_mount_form_validation_shows_errors(client: TestClient, state: NetMover) -> None:
|
||
bad = {**MOUNT_FORM, "server": "nas;reboot", "extra_options": "password=x"}
|
||
r = post(client, "/mounts/new", bad)
|
||
assert r.status_code == 400
|
||
assert "Ungültiger Servername" in r.text and "gesperrt" in r.text
|
||
with state.session_factory() as s:
|
||
assert s.scalars(select(Mount)).all() == []
|
||
|
||
|
||
def test_connection_change_requires_unmount(
|
||
client: TestClient, state: NetMover, backend: FakeMountBackend
|
||
) -> None:
|
||
backend.export("nas", "scans")
|
||
post(client, "/mounts/new", MOUNT_FORM)
|
||
mid = _mount_id(state, "Synology Scans")
|
||
post(client, f"/mounts/{mid}/mount")
|
||
r = post(client, f"/mounts/{mid}/edit", {**MOUNT_FORM, "share": "anders", "password": ""})
|
||
assert r.status_code == 400 and "ausgehängt" in r.text
|
||
# Nicht-Verbindungsfelder gehen trotzdem
|
||
r = post(client, f"/mounts/{mid}/edit", {**MOUNT_FORM, "password": "", "auto_mount": ""})
|
||
assert r.status_code == 303
|
||
|
||
|
||
def test_form_test_with_unsaved_data(client: TestClient, backend: FakeMountBackend) -> None:
|
||
backend.export("nas", "scans")
|
||
r = post(client, "/mounts/test", MOUNT_FORM)
|
||
assert r.status_code == 200 and "Einhängen" in r.text and "✓" in r.text
|
||
r = post(client, "/mounts/test", {**MOUNT_FORM, "share": "fehlt"})
|
||
assert ("✗" in r.text and "existiert auf dem Server nicht" not in r.text) or "✗" in r.text
|
||
|
||
|
||
def test_sftp_mount_generates_key(client: TestClient, state: NetMover) -> None:
|
||
r = post(
|
||
client,
|
||
"/mounts/new",
|
||
{
|
||
"name": "paperless",
|
||
"type": "sftp",
|
||
"server": "10.0.0.5",
|
||
"share": "/opt/consume",
|
||
"port": "22",
|
||
"auth": "ssh_key",
|
||
"username": "paperless",
|
||
"enabled": "on",
|
||
},
|
||
)
|
||
assert r.status_code == 303, r.text
|
||
page = client.get("/mounts").text
|
||
assert "ssh-ed25519 " in page and "BEGIN OPENSSH PRIVATE KEY" not in page
|
||
with state.session_factory() as s:
|
||
m = s.scalars(select(Mount)).one()
|
||
assert m.mountpoint is None and m.auto_mount is False
|
||
|
||
|
||
def test_delete_mount_used_by_rule_is_refused(client: TestClient, state: NetMover) -> None:
|
||
post(client, "/mounts/new", MOUNT_FORM)
|
||
mid = _mount_id(state, "Synology Scans")
|
||
with session_scope(state.session_factory) as s:
|
||
s.add(Rule(name="r", src_mount_id=mid, dst_path="/tmp"))
|
||
post(client, f"/mounts/{mid}/delete")
|
||
assert "Wird noch von Regeln benutzt" in client.get("/mounts").text
|
||
assert _mount_id(state, "Synology Scans") == mid
|
||
|
||
|
||
# -- Regeln -----------------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def mounted(
|
||
client: TestClient, state: NetMover, backend: FakeMountBackend, tmp_path: Path
|
||
) -> tuple[int, Path, Path]:
|
||
remote = backend.export("nas", "scans")
|
||
post(client, "/mounts/new", MOUNT_FORM)
|
||
mid = _mount_id(state, "Synology Scans")
|
||
post(client, f"/mounts/{mid}/mount")
|
||
consume = tmp_path / "consume"
|
||
consume.mkdir()
|
||
return mid, remote, consume
|
||
|
||
|
||
def rule_form(mid: int, consume: Path, **kw: str) -> dict[str, str]:
|
||
data = {
|
||
"name": "Scans → Paperless",
|
||
"enabled": "on",
|
||
"src_mount_id": str(mid),
|
||
"src_subpath": "",
|
||
"dst_kind": "local",
|
||
"dst_path": str(consume),
|
||
"pattern": "*.pdf",
|
||
"pattern_type": "glob",
|
||
"interval_seconds": "30",
|
||
"stability_seconds": "0",
|
||
"min_age_seconds": "0",
|
||
"op_timeout_seconds": "120",
|
||
"action": "move",
|
||
"on_collision": "rename",
|
||
"quarantine_after": "3",
|
||
}
|
||
data.update(kw)
|
||
return data
|
||
|
||
|
||
def test_rule_validation(client: TestClient, mounted: tuple[int, Path, Path]) -> None:
|
||
mid, _, consume = mounted
|
||
r = post(
|
||
client,
|
||
"/rules/new",
|
||
rule_form(
|
||
mid,
|
||
consume,
|
||
interval_seconds="5",
|
||
pattern_type="regex",
|
||
pattern="([",
|
||
set_owner="root",
|
||
dst_path="relativ",
|
||
),
|
||
)
|
||
assert r.status_code == 400
|
||
for msg in ("mindestens 10", "regulärer Ausdruck", "uid:gid", "absoluten Pfad"):
|
||
assert msg in r.text
|
||
|
||
|
||
def test_rule_create_dryrun_run(
|
||
client: TestClient, state: NetMover, mounted: tuple[int, Path, Path]
|
||
) -> None:
|
||
mid, remote, consume = mounted
|
||
(remote / "scan001.pdf").write_text("pdf")
|
||
(remote / "notiz.txt").write_text("txt")
|
||
r = post(client, "/rules/new", rule_form(mid, consume))
|
||
assert r.status_code == 303, r.text
|
||
with state.session_factory() as s:
|
||
rid = s.scalars(select(Rule.id)).one()
|
||
assert state.scheduler.next_run(rid) is not None
|
||
|
||
r = post(client, f"/rules/{rid}/dryrun")
|
||
assert r.status_code == 200
|
||
assert "scan001.pdf" in r.text and str(consume / "scan001.pdf") in r.text
|
||
assert "notiz.txt" not in r.text and "nichts verändert" in r.text
|
||
assert (remote / "scan001.pdf").exists()
|
||
|
||
result = state.runs.execute(rid) # „Jetzt ausführen“ synchron für den Test
|
||
assert result is not None
|
||
assert (consume / "scan001.pdf").exists() and not (remote / "scan001.pdf").exists()
|
||
|
||
log = client.get("/log").text
|
||
assert "Scans → Paperless" in log
|
||
with state.session_factory() as s:
|
||
run = s.scalars(select(Run).where(Run.dry_run.is_(False))).one()
|
||
assert run.status is RunStatus.OK
|
||
items = client.get(f"/log/{run.id}/items").text
|
||
assert "scan001.pdf" in items and "verschoben" in items
|
||
|
||
|
||
def test_rule_run_now_button(
|
||
client: TestClient, state: NetMover, mounted: tuple[int, Path, Path]
|
||
) -> None:
|
||
mid, _, consume = mounted
|
||
post(client, "/rules/new", rule_form(mid, consume))
|
||
with state.session_factory() as s:
|
||
rid = s.scalars(select(Rule.id)).one()
|
||
r = post(client, f"/rules/{rid}/run")
|
||
assert r.status_code == 303
|
||
assert "Lauf gestartet" in client.get("/").text
|
||
|
||
|
||
def test_rule_toggle_and_delete_sync_scheduler(
|
||
client: TestClient, state: NetMover, mounted: tuple[int, Path, Path]
|
||
) -> None:
|
||
mid, _, consume = mounted
|
||
post(client, "/rules/new", rule_form(mid, consume))
|
||
with state.session_factory() as s:
|
||
rid = s.scalars(select(Rule.id)).one()
|
||
post(client, f"/rules/{rid}/toggle")
|
||
assert state.scheduler.next_run(rid) is None
|
||
post(client, f"/rules/{rid}/toggle")
|
||
assert state.scheduler.next_run(rid) is not None
|
||
post(client, f"/rules/{rid}/delete")
|
||
assert state.scheduler.next_run(rid) is None
|
||
|
||
|
||
def test_status_page_shows_stale_warning(
|
||
client: TestClient, state: NetMover, mounted: tuple[int, Path, Path]
|
||
) -> None:
|
||
from datetime import timedelta
|
||
|
||
from netmover.models import utcnow
|
||
|
||
mid, _, consume = mounted
|
||
post(client, "/rules/new", rule_form(mid, consume))
|
||
with session_scope(state.session_factory) as s:
|
||
r = s.scalars(select(Rule)).one()
|
||
r.created_at = utcnow() - timedelta(hours=1)
|
||
r.last_success_at = utcnow() - timedelta(minutes=10)
|
||
state.started_at = utcnow() - timedelta(hours=1)
|
||
page = client.get("/").text
|
||
assert "länger als 3 × Intervall" in page and 'class="warn"' in page
|
||
assert client.get("/healthz").status_code == 503
|
||
|
||
|
||
# -- Einstellungen -------------------------------------------------------------------------------
|
||
|
||
|
||
def test_settings_webhook_url_is_masked(client: TestClient, state: NetMover) -> None:
|
||
from netmover.settings_store import load_settings
|
||
|
||
url = "https://ntfy.example/geheimes-topic-token"
|
||
r = post(
|
||
client,
|
||
"/settings",
|
||
{
|
||
"retention_days": "14",
|
||
"notify_mode": "every",
|
||
"notify_after": "3",
|
||
"webhook_url": url,
|
||
"webhook_content_type": "text/plain",
|
||
"webhook_template": "$message",
|
||
},
|
||
)
|
||
assert r.status_code == 303
|
||
page = client.get("/settings").text
|
||
assert "geheimes-topic-token" not in page and "ntfy.example" in page
|
||
# Maske unverändert zurückschicken → URL bleibt
|
||
masked = re.search(r'name="webhook_url" value="([^"]*)"', page)
|
||
assert masked
|
||
post(
|
||
client,
|
||
"/settings",
|
||
{
|
||
"retention_days": "14",
|
||
"notify_mode": "every",
|
||
"notify_after": "3",
|
||
"webhook_url": masked.group(1),
|
||
"webhook_content_type": "text/plain",
|
||
"webhook_template": "$message",
|
||
},
|
||
)
|
||
cfg = load_settings(state.session_factory)
|
||
assert cfg.webhook_url == url and cfg.retention_days == 14
|
||
|
||
|
||
def test_change_password(client: TestClient) -> None:
|
||
r = post(
|
||
client,
|
||
"/settings/password",
|
||
{"current": "falsch", "password": "neues-passwort-1", "confirm": "neues-passwort-1"},
|
||
)
|
||
assert "falsch" in client.get("/settings").text
|
||
r = post(
|
||
client,
|
||
"/settings/password",
|
||
{"current": PASSWORD, "password": "neues-passwort-1", "confirm": "neues-passwort-1"},
|
||
)
|
||
assert r.status_code == 303
|
||
|
||
|
||
def test_export_excludes_secrets_and_import_roundtrip(
|
||
client: TestClient,
|
||
state: NetMover,
|
||
mounted: tuple[int, Path, Path],
|
||
tmp_path: Path,
|
||
) -> None:
|
||
mid, _, consume = mounted
|
||
post(client, "/rules/new", rule_form(mid, consume))
|
||
r = client.get("/settings/export")
|
||
assert r.status_code == 200 and "attachment" in r.headers["content-disposition"]
|
||
text = r.text
|
||
assert MOUNT_FORM["password"] not in text and "scanner" not in text and "secret_enc" not in text
|
||
data = json.loads(text)
|
||
assert data["mounts"][0]["name"] == "Synology Scans"
|
||
assert data["rules"][0]["src_mount"] == "Synology Scans"
|
||
|
||
# in eine frische Instanz importieren
|
||
settings2 = Settings(
|
||
data_dir=tmp_path / "lib2",
|
||
run_dir=tmp_path / "run2",
|
||
mount_base=tmp_path / "mnt2",
|
||
mount_backend="fake",
|
||
)
|
||
state2 = NetMover(settings2, backend=FakeMountBackend(tmp_path / "fake2"))
|
||
with TestClient(create_app(settings2, state2), follow_redirects=False) as c2:
|
||
c2.headers["X-Test-CSRF"] = do_setup(c2)
|
||
r = c2.post(
|
||
"/settings/import",
|
||
data={"csrf_token": c2.headers["X-Test-CSRF"]},
|
||
files={"file": ("export.json", text, "application/json")},
|
||
)
|
||
assert r.status_code == 303 and r.headers["location"] == "/settings", r.headers
|
||
page = c2.get("/settings").text
|
||
assert "Importiert" in page and "neu eingeben" in page
|
||
with state2.session_factory() as s:
|
||
m = s.scalars(select(Mount)).one()
|
||
assert m.credential is None and m.auto_mount is False
|
||
rule = s.scalars(select(Rule)).one()
|
||
assert rule.enabled is False and rule.src_mount_id == m.id
|
||
# zweiter Import überspringt vorhandene Namen
|
||
c2.post(
|
||
"/settings/import",
|
||
data={"csrf_token": c2.headers["X-Test-CSRF"]},
|
||
files={"file": ("export.json", text, "application/json")},
|
||
)
|
||
assert "Übersprungen" in c2.get("/settings").text
|
||
|
||
|
||
# -- Schlüsselverwaltung -------------------------------------------------------------------
|
||
|
||
|
||
def test_keys_page_generate_import_select_delete(client: TestClient, state: NetMover) -> None:
|
||
from netmover.crypto import generate_ssh_keypair
|
||
|
||
r = post(client, "/credentials/keys/generate", {"name": "paperless-lxc"})
|
||
assert r.status_code == 303
|
||
private, public = generate_ssh_keypair()
|
||
r = post(
|
||
client,
|
||
"/credentials/keys/import",
|
||
{"name": "alt", "private_key": private, "passphrase": ""},
|
||
)
|
||
assert r.status_code == 303
|
||
page = client.get("/credentials").text
|
||
assert "paperless-lxc" in page and "alt" in page
|
||
assert public.split()[1] in page # öffentlicher Teil sichtbar
|
||
for line in private.splitlines()[1:-1]: # keine Zeile des privaten Schlüssels
|
||
assert line not in page
|
||
assert "restrict,command="/usr/lib/openssh/sftp-server"" in page
|
||
|
||
with state.session_factory() as s:
|
||
key_id = s.scalar(select(Credential.id).where(Credential.name == "alt"))
|
||
form = client.get("/mounts/new").text
|
||
assert f'value="{key_id}"' in form and "neuer Schlüssel nur für diese Anbindung" in form
|
||
|
||
r = post(
|
||
client,
|
||
"/mounts/new",
|
||
{
|
||
"name": "paperless",
|
||
"type": "sftp",
|
||
"server": "10.0.0.5",
|
||
"share": "/consume",
|
||
"port": "22",
|
||
"auth": "ssh_key",
|
||
"key_id": str(key_id),
|
||
"username": "root",
|
||
"enabled": "on",
|
||
},
|
||
)
|
||
assert r.status_code == 303, r.text
|
||
mounts = client.get("/mounts").text
|
||
assert "gespeicherten Zugangsdaten" in mounts and "„alt“" in mounts
|
||
edit = client.get(f"/mounts/{_mount_id(state, 'paperless')}/edit").text
|
||
assert f'<option value="{key_id}" selected>' in edit and 'value="root"' in edit
|
||
|
||
post(client, f"/credentials/{key_id}/delete")
|
||
assert "noch benutzt" in client.get("/credentials").text
|
||
post(client, f"/mounts/{_mount_id(state, 'paperless')}/delete")
|
||
post(client, f"/credentials/{key_id}/delete")
|
||
page = client.get("/credentials").text
|
||
assert "gelöscht" in page and ">alt<" not in page
|
||
|
||
|
||
def test_key_import_error_is_shown(client: TestClient) -> None:
|
||
post(
|
||
client,
|
||
"/credentials/keys/import",
|
||
{"name": "x", "private_key": "kein schlüssel", "passphrase": ""},
|
||
)
|
||
assert "Kein lesbarer privater Schlüssel" in client.get("/credentials").text
|
||
|
||
|
||
def test_no_escaped_attributes_in_pages(client: TestClient, state: NetMover) -> None:
|
||
"""Attribute in {{ … }} werden von Jinja maskiert – dann landet " im HTML."""
|
||
|
||
post(client, "/credentials/keys/generate", {"name": "k"})
|
||
for url in (
|
||
"/",
|
||
"/mounts",
|
||
"/mounts/new",
|
||
"/keys",
|
||
"/rules",
|
||
"/rules/new",
|
||
"/log",
|
||
"/settings",
|
||
"/help",
|
||
):
|
||
html = client.get(url).text
|
||
broken = re.findall(r"<[a-z][^<>]*="[^<>]*>", html) # nur in Tags, nicht im Text
|
||
assert not broken, (url, broken[:2])
|
||
|
||
|
||
def test_credentials_page_passwords(
|
||
client: TestClient, state: NetMover, backend: FakeMountBackend
|
||
) -> None:
|
||
r = post(
|
||
client,
|
||
"/credentials/passwords",
|
||
{"name": "NAS-Scanner", "username": "scanner", "password": "Geheim,123", "domain": ""},
|
||
)
|
||
assert r.status_code == 303
|
||
page = client.get("/credentials").text
|
||
assert "NAS-Scanner" in page and "scanner" in page and "Geheim,123" not in page
|
||
with state.session_factory() as s:
|
||
cid = s.scalar(select(Credential.id).where(Credential.name == "NAS-Scanner"))
|
||
form = client.get("/mounts/new").text
|
||
assert f'<option value="{cid}"' in form and "eigene Angaben für diese Anbindung" in form
|
||
backend.export("nas", "scans")
|
||
r = post(
|
||
client, "/mounts/new", {**MOUNT_FORM, "username": "", "password": "", "cred_id": str(cid)}
|
||
)
|
||
assert r.status_code == 303, r.text
|
||
assert "gespeicherten Zugangsdaten" in client.get("/mounts").text
|
||
mid = _mount_id(state, "Synology Scans")
|
||
assert "Quelle mit Löschen" in post(client, f"/mounts/{mid}/test").text
|
||
post(
|
||
client,
|
||
f"/credentials/{cid}/password",
|
||
{"username": "scanner2", "domain": "", "password": ""},
|
||
)
|
||
assert "gilt ab sofort für alle" in client.get("/credentials").text
|
||
assert client.get("/keys").headers["location"] == "/credentials"
|