alientim/tesm
0
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases 41 Wiki Activity

TESM 2.0.0 -- Neubau

Browse Source 
Vollstaendiger Neubau der Anwendung. Der vorherige Stand bleibt unveraendert
im Zweig SONNET5 erhalten.

Aufbau: apps/tesm (Anwendung), packages/tesm-core (gemeinsamer Kern),
packages/tesm-licensing (Lizenzprotokoll), deploy (Installation, systemd,
privilegierter Helfer), docs, tests. Der Lizenzserver liegt in seinem eigenen
Repository; beide Repositorien bringen die gemeinsamen Pakete mit, damit sich
jedes allein installieren laesst.

Die wichtigsten Unterschiede zum Vorgaenger, jeweils an der Stelle im Code
kommentiert, an der der Fehler entstanden ist:

* Der Webprozess laeuft unprivilegiert. Alles, was Root braucht, geht ueber
  einen einzigen Helfer mit Positivlisten fuer jedes Argument.
* CSRF-Schutz ueberhaupt -- der Vorgaenger hatte keinen.
* Rechte werden serverseitig geprueft, nicht nur im Template ausgeblendet.
* Die nginx-Site wird bei jedem Lauf inhaltlich verglichen und erneuert.
* Jede erzeugte Konfiguration wird vor dem Uebernehmen geprueft (nginx, Kea).
* Kein Hostname im Lizenz-Fingerabdruck.
* Zwei Installationen auf einem Host stoeren sich nicht (eigener SITE_KEY).
* Verschachtelte Datenbankverbindungen sind ein Fehler, kein Deadlock.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
alientim
2026-09-02 18:00:06 +02:00
co-authored by Claude Opus 5
parent a193b9d6e9
commit 7354ff352b
227 changed files with 37605 additions and 21187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/conftest.py
+211
View File
@@ -0,0 +1,211 @@
"""Gemeinsame Testfixtures.
Jeder Test bekommt eine frische Instanz in einem temporaeren Verzeichnis --
eigene Datenbank, eigene Schluessel, eigenes Protokollverzeichnis. Nichts
beruehrt eine echte Installation.
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Any, Iterator
import pytest
@pytest.fixture()
def instance_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
base = tmp_path / "instance"
monkeypatch.setenv("TESM_BASE_DIR", str(base))
monkeypatch.setenv("TESM_LICENSE_BASE_DIR", str(tmp_path / "instance-license"))
# Tests laufen ueber http:// -- ein Secure-Cookie wuerde nie gesetzt.
monkeypatch.setenv("TESM_COOKIE_SECURE", "0")
monkeypatch.setenv("TESM_LICENSE_COOKIE_SECURE", "0")
monkeypatch.setenv("TESM_DEBUG", "0")
monkeypatch.setenv("TESM_LICENSE_DEBUG", "0")
return base
@pytest.fixture()
def core_app(instance_env: Path) -> Iterator[Any]:
"""Minimale App auf Basis von ``tesm_core`` -- ohne Fachlogik der beiden Apps."""
from flask import render_template_string
from tesm_core import load_core_config
from tesm_core.app_factory import create_base_app
from tesm_core.auth.models import login_required
from tesm_core.nav import NavItem, NavTree
from tesm_core.rbac.common import administration_area, observability_area
from tesm_core.rbac.model import Action, Area, PermissionTree, Resource, RolePreset
demo = Area(
key="demo",
label="Demo",
icon="grid",
resources=(
Resource(
key="widgets",
label="Widgets",
actions=(Action.VIEW, Action.CREATE, Action.EDIT, Action.DELETE, Action.SECRETS),
),
),
)
tree = PermissionTree([demo, administration_area(), observability_area()])
nav = NavTree(
[
NavItem(key="index", label="Start", icon="grid", endpoint="index"),
NavItem(
key="widgets", label="Widgets", icon="cpu", endpoint="widgets",
permissions=("widgets.view",),
),
NavItem(
key="administration",
label="Verwaltung",
icon="sliders",
children=(
NavItem(
key="users", label="Benutzer", icon="users", endpoint="admin.users",
permissions=("users.view",),
),
NavItem(
key="groups", label="Gruppen", icon="shield-users", endpoint="admin.groups",
permissions=("groups.view",),
),
NavItem(
key="trash", label="Papierkorb", icon="trash", endpoint="trash.overview",
permissions=("trash.view",),
),
NavItem(
key="diagnostics", label="Diagnose", icon="info",
endpoint="diagnostics.overview", permissions=("diagnostics.view",),
),
),
),
]
)
presets = (
RolePreset(
key="viewer",
name="Benutzer",
description="Nur lesen",
grants=(("widgets", (Action.VIEW,)),),
is_default=True,
),
)
config = load_core_config(
app_key="tesm", app_name="TESM Test", env_prefix="TESM_", default_base=instance_env
)
app = create_base_app(
config=config,
permissions=tree,
nav=nav,
role_presets=presets,
app_short="TT",
app_edition="Test",
enable_license=False,
)
@app.get("/")
def index(): # type: ignore[no-untyped-def]
return render_template_string(
'{% extends "tesm_core/base.html" %}{% block page_title %}Start{% endblock %}'
"{% block content %}<p>Start</p>{% endblock %}"
)
@app.get("/widgets")
@login_required
def widgets(): # type: ignore[no-untyped-def]
return render_template_string(
'{% extends "tesm_core/base.html" %}{% block page_title %}Widgets{% endblock %}'
"{% block content %}<p>Widgets</p>{% endblock %}"
)
app.config.update(TESTING=True)
yield app
@pytest.fixture()
def client(core_app: Any) -> Any:
return core_app.test_client()
@pytest.fixture()
def make_user(core_app: Any):
"""Erzeugt ein Konto direkt in der Datenbank."""
def factory(
username: str = "tester",
password: str = "Sicher-Passwort-2026",
*,
is_admin: bool = False,
permissions: tuple[str, ...] = (),
is_locked: bool = False,
) -> dict[str, Any]:
from tesm_core.db import Database
from tesm_core.extension import core
from tesm_core.security import passwords
extension = core(core_app)
with extension.database.session() as conn, extension.database.transaction(conn):
cursor = conn.execute(
"INSERT INTO users (username, password_hash, is_admin, is_locked, auth_source, "
"created_at, updated_at) VALUES (?,?,?,?,'local',datetime('now'),datetime('now'))",
(username, passwords.hash_password(password), int(is_admin), int(is_locked)),
)
user_id = int(cursor.lastrowid or 0)
if permissions:
group = conn.execute(
"INSERT INTO groups (name, description, created_at, updated_at) "
"VALUES (?,'Testgruppe',datetime('now'),datetime('now'))",
(f"grp-{username}",),
)
group_id = int(group.lastrowid or 0)
conn.executemany(
"INSERT INTO group_permissions (group_id, permission) VALUES (?,?)",
[(group_id, p) for p in extension.permissions.sanitize(permissions)],
)
conn.execute(
"INSERT INTO user_groups (user_id, group_id) VALUES (?,?)", (user_id, group_id)
)
row = Database.one(conn, "SELECT * FROM users WHERE id=?", (user_id,))
assert row is not None
return {**row, "password": password}
return factory
@pytest.fixture()
def login(client: Any):
def do_login(username: str, password: str) -> Any:
return client.post(
"/login",
data={"username": username, "password": password, "csrf_token": _token(client)},
follow_redirects=True,
)
return do_login
def _token(client: Any) -> str:
"""Holt ein CSRF-Token aus der Anmeldeseite."""
page = client.get("/login")
body = page.get_data(as_text=True)
marker = 'name="csrf_token" value="'
start = body.index(marker) + len(marker)
return body[start : body.index('"', start)]
@pytest.fixture()
def csrf_token(client: Any):
def get(path: str = "/login") -> str:
page = client.get(path)
body = page.get_data(as_text=True)
marker = 'name="csrf_token" value="'
if marker not in body:
marker = 'data-csrf="'
start = body.index(marker) + len(marker)
return body[start : body.index('"', start)]
return get
tests/e2e_smoke.py
+499
View File
@@ -0,0 +1,499 @@
"""End-to-End-Rauchtest gegen eine laufende Installation.
Laeuft auf dem Zielhost, spricht ueber HTTP mit nginx und der echten Anwendung
-- kein Testclient, keine Attrappen. Prueft genau das, was ein Mensch beim
ersten Durchklicken pruefen wuerde, plus die Dinge, die man beim Klicken leicht
uebersieht (Sicherheitsheader, CSRF, statische Dateien, Rechteschranken).
Aufruf:
python3 e2e_smoke.py http://127.0.0.1:8080 admin 'Passwort'
"""
from __future__ import annotations
import http.cookiejar
import json
import re
import secrets
import ssl
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
CSRF_RE = re.compile(r'name="csrf_token" value="([^"]+)"')
BODY_CSRF_RE = re.compile(r'data-csrf="([^"]+)"')
class Client:
def __init__(self, base: str) -> None:
self.base = base.rstrip("/")
self.https = self.base.startswith("https://")
self.jar = http.cookiejar.CookieJar()
# Der Test laeuft auf dem Zielhost selbst gegen ein Zertifikat, das
# dieselbe Maschine ausgestellt hat. Es zu pruefen hiesse, das eigene
# Zertifikat gegen sich selbst zu pruefen -- geprueft wird hier, ob TLS
# steht und die Anwendung darueber antwortet.
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
self.tls = urllib.request.HTTPSHandler(context=context)
self.opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.jar),
urllib.request.HTTPRedirectHandler(),
self.tls,
)
self.opener.addheaders = [("User-Agent", "tesm-e2e/2.0")]
class _NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *_args, **_kwargs): # type: ignore[override]
return None
# Zweiter Opener ohne Weiterleitung: nur so laesst sich pruefen, ob eine
# Seite tatsaechlich mit 302 auf die Anmeldung verweist.
self.plain = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(self.jar), _NoRedirect(), self.tls
)
self.plain.addheaders = [("User-Agent", "tesm-e2e/2.0")]
def get(self, path: str) -> tuple[int, str, dict[str, str]]:
request = urllib.request.Request(self.base + path)
try:
with self.opener.open(request, timeout=20) as response:
return response.status, response.read().decode("utf-8", "replace"), dict(
response.headers
)
except urllib.error.HTTPError as exc:
return exc.code, exc.read().decode("utf-8", "replace"), dict(exc.headers)
def get_raw(self, path: str) -> int:
"""Statuscode ohne Weiterleitung zu folgen."""
request = urllib.request.Request(self.base + path)
try:
with self.plain.open(request, timeout=20) as response:
return response.status
except urllib.error.HTTPError as exc:
return exc.code
def post(self, path: str, data: dict[str, object]) -> tuple[int, str]:
encoded = urllib.parse.urlencode(
[
(key, str(item))
for key, value in data.items()
for item in (value if isinstance(value, (list, tuple)) else [value])
]
).encode("utf-8")
request = urllib.request.Request(
self.base + path,
data=encoded,
headers={
"Content-Type": "application/x-www-form-urlencoded",
"Origin": self.base,
"Referer": self.base + path,
},
)
try:
with self.opener.open(request, timeout=30) as response:
return response.status, response.read().decode("utf-8", "replace")
except urllib.error.HTTPError as exc:
return exc.code, exc.read().decode("utf-8", "replace")
def token(self, path: str) -> str:
_, body, _ = self.get(path)
match = CSRF_RE.search(body) or BODY_CSRF_RE.search(body)
if not match:
raise AssertionError(f"Kein CSRF-Token auf {path}")
return match.group(1)
PASSED: list[str] = []
FAILED: list[str] = []
#: Jeder Lauf legt eigene Datensaetze an. Sonst laesst sich der Test kein
#: zweites Mal gegen dieselbe Installation ausfuehren -- beim ersten Lauf
#: landet ein Geraet im Papierkorb, und die Anwendung weigert sich danach
#: voellig zu Recht, dieselbe MAC noch einmal zu vergeben.
RUN = secrets.token_hex(3)
MAC = ":".join(("AA", "BB", RUN[0:2], RUN[2:4], RUN[4:6], "01")).upper()
SUBNET = f"10.99.{int(RUN[0:2], 16)}"
FLASH_RE = re.compile(r'id="flashed-messages">(.*?)</script>', re.S)
ROW_SPLIT = re.compile(r"<tr[ >]|<li[ >]")
def flash(body: str) -> str:
"""Die Meldungen der Seite -- damit ein Fehlschlag erklaert, was schiefging."""
match = FLASH_RE.search(body)
if not match:
return "keine Meldung"
try:
return " | ".join(text for _level, text in json.loads(match.group(1)))
except (ValueError, TypeError):
return match.group(1)[:120]
def id_for(body: str, needle: str, pattern: str) -> str:
"""Sucht die Kennung in der Zeile, in der ``needle`` steht.
Die Listenansicht nennt die Kennung nur in Formular- und Link-Zielen. Der
Test darf sie nicht raten -- eine feste ``1`` funktioniert genau einmal und
trifft beim naechsten Lauf einen fremden Datensatz.
"""
for chunk in ROW_SPLIT.split(body):
if needle in chunk:
match = re.search(pattern, chunk)
if match:
return match.group(1)
return ""
def check(name: str, condition: bool, detail: str = "") -> None:
if condition:
PASSED.append(name)
print(f" [ok] {name}")
else:
FAILED.append(f"{name}: {detail}")
print(f" [FEHL] {name} -- {detail}")
def local_host_fingerprints() -> set[str]:
"""Die Fingerprints, die ``ssh-keyscan`` fuer diesen Host meldet.
Der Vergleich ist der Kern der Pruefung: die Anwendung darf nicht
irgendeinen plausiblen Wert anzeigen, sondern genau den des Gegenuebers.
"""
try:
scan = subprocess.run(
["ssh-keyscan", "-T", "5", "-p", "22", "127.0.0.1"],
capture_output=True,
text=True,
timeout=20,
)
fingerprints: set[str] = set()
for line in scan.stdout.splitlines():
if not line.strip() or line.startswith("#"):
continue
result = subprocess.run(
["ssh-keygen", "-lf", "-"], input=line, capture_output=True, text=True, timeout=10
)
match = re.search(r"(SHA256:[A-Za-z0-9+/]{43})", result.stdout)
if match:
fingerprints.add(match.group(1))
return fingerprints
except (OSError, subprocess.SubprocessError):
return set()
def main(base: str, username: str, password: str) -> int:
client = Client(base)
print(f"\n== Rauchtest gegen {base} (Lauf {RUN}) ==\n")
# -- oeffentlich ---------------------------------------------------------
print("Oeffentlich erreichbar:")
status, body, headers = client.get("/gesundheit")
check("Gesundheitsendpunkt", status == 200 and json.loads(body)["status"] == "ok", body[:80])
status, body, headers = client.get("/login")
check("Anmeldeseite", status == 200 and "Anmelden" in body, f"HTTP {status}")
if client.https:
cookie = headers.get("Set-Cookie", "")
check("Sitzungscookie mit Secure", "Secure" in cookie, cookie[:80] or "kein Cookie")
check("Sitzungscookie HttpOnly", "HttpOnly" in cookie, cookie[:80] or "kein Cookie")
check("Sitzungscookie SameSite", "SameSite" in cookie, cookie[:80] or "kein Cookie")
csp = headers.get("Content-Security-Policy", "")
check("CSP mit Nonce", "nonce-" in csp and "unsafe-inline" not in csp.split("style-src")[0],
csp[:90])
check("X-Frame-Options", headers.get("X-Frame-Options") == "DENY", str(headers.get("X-Frame-Options")))
check("Vorgangsnummer im Header", bool(headers.get("X-Request-Id")), "fehlt")
status, body, _ = client.get("/")
check("Oeffentliche Statusuebersicht", status == 200 and "Statusuebersicht" in body,
f"HTTP {status}")
check("Ohne Anmeldung keine Details", "oeffentliche Ansicht" in body, "Hinweis fehlt")
for asset in ("/core-assets/css/base.css", "/core-assets/js/app.js"):
status, body, _ = client.get(asset)
check(f"Statische Datei {asset}", status == 200 and len(body) > 1000, f"HTTP {status}")
print("\nSchranken ohne Anmeldung:")
for path in ("/clients/", "/verwaltung/benutzer", "/diagnose/"):
status = client.get_raw(path)
check(f"{path} verlangt Anmeldung", status in (302, 401), f"HTTP {status}")
print("\nCSRF:")
status, body = client.post("/login", {"username": "x", "password": "y"})
check("POST ohne Token wird abgewiesen", status == 403, f"HTTP {status}")
# -- Anmeldung -----------------------------------------------------------
print("\nAnmeldung:")
token = client.token("/login")
status, body = client.post(
"/login", {"username": username, "password": password, "csrf_token": token}
)
check("Anmeldung erfolgreich", "Willkommen" in body or status == 200, f"HTTP {status}")
status, body, _ = client.get("/konto/")
check("Kontoseite erreichbar", status == 200 and username in body, f"HTTP {status}")
# -- Seiten --------------------------------------------------------------
print("\nSeiten (angemeldet):")
pages = [
("/", "Statusuebersicht"),
("/clients/", "Clients"),
("/switche/", "Switche"),
("/zugangsdaten/", "Zugangsdaten"),
("/protokolle/laufend", "Protokoll"),
("/protokolle/aenderungen", "Aenderungsprotokoll"),
("/protokolle/neustarts", "Neustarts"),
("/einstellungen/system", "Systemeinstellungen"),
("/einstellungen/netzwerk", "Netzwerk"),
("/einstellungen/webserver", "Webserver"),
("/einstellungen/verzeichnisdienst", "Verzeichnisdienst"),
("/lizenz/", "Lizenz"),
("/sicherung/", "Sicherung"),
("/papierkorb/", "Papierkorb"),
("/diagnose/", "Diagnose"),
("/verwaltung/benutzer", "Benutzer"),
("/verwaltung/gruppen", "Gruppen"),
("/verwaltung/sitzungen", "Sitzungen"),
("/konto/sicherheit", "Sicherheit"),
]
for path, needle in pages:
status, body, _ = client.get(path)
ok = status == 200 and needle in body and "Interner Fehler" not in body
check(f"{path}", ok, f"HTTP {status}")
# Die Modulrouten sind an die Lizenz gebunden. Welcher Zustand gerade gilt,
# darf der Test nicht vorschreiben -- er darf nur verlangen, dass alle drei
# dasselbe sagen. Ein Modul, das ohne Lizenz durchkommt, waere ein Loch;
# ein Modul, das mit Lizenz 404 liefert, waere ein Ausfall.
print("\nLizenzpflichtige Module:")
module_status = {
path: client.get(path)[0] for path in ("/dhcp/", "/wartung/", "/dateifreigaben/")
}
licensed = set(module_status.values()) == {200}
unlicensed = set(module_status.values()) == {404}
check(
"Alle Modulrouten im selben Zustand",
licensed or unlicensed,
", ".join(f"{path} {code}" for path, code in module_status.items()),
)
print(f" [info] Lizenz ist {'aktiv' if licensed else 'nicht aktiv'}.")
# -- Daten anlegen -------------------------------------------------------
print("\nDaten anlegen:")
credential_name = f"E2E-Switchlogin-{RUN}"
switch_name = f"e2e-switch-{RUN}"
device_name = f"E2E-Kamera-{RUN}"
secret = f"Geheim-fuer-den-Test-{RUN}"
token = client.token("/zugangsdaten/")
status, body = client.post(
"/zugangsdaten/neu",
{
"csrf_token": token,
"name": credential_name,
"username": "operator",
"secret": secret,
"category": "switch",
},
)
check("Zugangsdaten angelegt", "angelegt" in body, flash(body))
credential_id = id_for(body, credential_name, r"/zugangsdaten/(\d+)/")
check("Kennung der Zugangsdaten gefunden", bool(credential_id), "nicht gefunden")
token = client.token("/switche/")
status, body = client.post(
"/switche/neu",
{
"csrf_token": token,
"hostname": switch_name,
"ip": f"{SUBNET}.2",
"ssh_port": "22",
"credential_id": credential_id,
},
)
check("Switch angelegt", "angelegt" in body, flash(body))
switch_id = id_for(body, switch_name, r"/switche/(\d+)")
check("Kennung des Switches gefunden", bool(switch_id), "nicht gefunden")
token = client.token("/clients/")
status, body = client.post(
"/clients/neu",
{
"csrf_token": token,
"name": device_name,
"mac": MAC,
"ip": f"{SUBNET}.50",
"switch_id": switch_id,
"port": "7",
"ssh_port": "22",
"is_active": "1",
"auto_restart": "1",
},
)
check("Geraet angelegt", "angelegt" in body, flash(body))
device_id = id_for(body, device_name, r"/clients/(\d+)")
check("Kennung des Geraets gefunden", bool(device_id), "nicht gefunden")
status, body, _ = client.get("/clients/")
check(
"Geraet in der Liste",
device_name in body and MAC.lower() in body,
"nicht gefunden",
)
status, body, _ = client.get("/")
check("Geraet auf der Uebersicht", device_name in body, "nicht gefunden")
print("\nValidierung:")
token = client.token("/clients/")
status, body = client.post(
"/clients/neu",
{"csrf_token": token, "name": "Kaputt", "mac": "unsinn", "ip": f"{SUBNET}.51"},
)
check("Ungueltige MAC abgelehnt", "hexadezimale" in body, flash(body))
token = client.token("/clients/")
status, body = client.post(
"/clients/neu",
{"csrf_token": token, "name": "Kaputt", "mac": MAC[:-2] + "FE", "ip": "999.1.1.1"},
)
check("Ungueltige IP abgelehnt", "gueltige IP-Adresse" in body, flash(body))
token = client.token("/clients/")
status, body = client.post(
"/clients/neu",
{"csrf_token": token, "name": "Doppelt", "mac": MAC, "ip": f"{SUBNET}.52"},
)
check("Doppelte MAC abgelehnt", "angelegt" not in body, "zweimal dieselbe MAC")
# -- Host-Schluessel eines Clients ---------------------------------------
# Gegen einen echten SSH-Dienst: der Host selbst. Nur so ist geprueft, dass
# der gelesene Fingerprint tatsaechlich der des Gegenuebers ist -- eine
# Attrappe wuerde genau den Fehler durchlassen, auf den es hier ankommt.
print("\nSSH-Host-Schluessel eines Clients:")
token = client.token("/clients/")
ssh_name = f"E2E-SSH-{RUN}"
status, body = client.post(
"/clients/neu",
{
"csrf_token": token,
"name": ssh_name,
"mac": MAC[:-2] + "0F",
"ip": "127.0.0.1",
"ssh_port": "22",
"credential_id": credential_id,
"is_active": "0",
},
)
check("Client fuer den SSH-Test angelegt", "angelegt" in body, flash(body))
ssh_id = id_for(body, ssh_name, r"/clients/(\d+)")
check("Kennung gefunden", bool(ssh_id), "nicht gefunden")
if ssh_id:
status, body, _ = client.get(f"/clients/{ssh_id}")
check("Detailseite zeigt die Schluesselkarte", "SSH-Host-Schluessel" in body, "fehlt")
check("Zustand offen", "Noch kein Schluessel hinterlegt" in body, "unerwartet")
token = client.token(f"/clients/{ssh_id}")
status, body = client.post(
f"/clients/{ssh_id}/hostkey/pruefen", {"csrf_token": token}
)
check("Schluessel ausgelesen", "Host-Schluessel gelesen" in body, flash(body))
match = re.search(r"(SHA256:[A-Za-z0-9+/]{43})", body)
fingerprint = match.group(1) if match else ""
check("Fingerprint erhalten", bool(fingerprint), "nicht gefunden")
expected = local_host_fingerprints()
check(
"Fingerprint stimmt mit ssh-keyscan ueberein",
bool(expected) and fingerprint in expected,
f"gelesen {fingerprint}, erwartet eines von {sorted(expected)}",
)
key_match = re.search(r'name="key_type" value="([^"]+)"', body)
key_type = key_match.group(1) if key_match else ""
token = client.token(f"/clients/{ssh_id}")
status, body = client.post(
f"/clients/{ssh_id}/hostkey/freigeben",
{"csrf_token": token, "key_type": key_type, "fingerprint": fingerprint},
)
check("Schluessel freigegeben", "freigegeben" in body, flash(body))
check("Fingerprint steht auf der Seite", fingerprint in body, "nicht gefunden")
token = client.token(f"/clients/{ssh_id}")
status, body = client.post(
f"/clients/{ssh_id}/hostkey/freigeben",
{"csrf_token": token, "key_type": key_type, "fingerprint": "SHA256:zu-kurz"},
)
check("Unsinniger Fingerprint abgelehnt", "kein gueltiger Fingerprint" in body, flash(body))
status, body, _ = client.get("/protokolle/aenderungen")
check("Freigabe protokolliert", "device.host_key_trusted" in body, "kein Eintrag")
token = client.token(f"/clients/{ssh_id}")
status, body = client.post(
f"/clients/{ssh_id}/hostkey/verwerfen", {"csrf_token": token}
)
check("Schluessel verworfen", "entfernt" in body, flash(body))
print("\nNeustart-Protokoll:")
status, body, _ = client.get("/protokolle/neustarts")
check("Seite heisst Neustarts", status == 200 and "Neustarts" in body, f"HTTP {status}")
check("Alte PoE-Adresse ist weg", client.get_raw("/protokolle/poe") == 404, "noch da")
check("Filter nach Methode vorhanden", "methode=ssh" in body, "kein Filter")
print("\nGeheimnisse:")
token = client.token("/zugangsdaten/")
status, body = client.post(
f"/zugangsdaten/{credential_id}/anzeigen", {"csrf_token": token}
)
check("Klartextanzeige moeglich (frisch angemeldet)", secret in body, f"HTTP {status}")
status, body, _ = client.get("/protokolle/aenderungen")
check("Zugriff protokolliert", "credential.secret_revealed" in body, "kein Eintrag")
print("\nPapierkorb:")
token = client.token("/clients/")
client.post(f"/clients/{device_id}/loeschen", {"csrf_token": token})
status, body, _ = client.get("/papierkorb/")
check("Geloeschtes Geraet im Papierkorb", device_name in body, "nicht gefunden")
token = client.token("/papierkorb/")
status, body = client.post(
f"/papierkorb/devices/{device_id}/wiederherstellen", {"csrf_token": token}
)
if licensed:
check("Wiederherstellen mit Lizenz moeglich", "wiederhergestellt" in body, flash(body))
else:
check("Wiederherstellen ohne Lizenz gesperrt", "Lizenz" in body, flash(body))
print("\nDiagnose:")
status, body, _ = client.get("/diagnose/")
check("Audit-Kette unversehrt", "unversehrt" in body, "Kette gebrochen?")
# Nicht "liegt die Datei da", sondern "antwortet er". Genau der
# Unterschied hat einmal verdeckt, dass jede Systemaktion an
# NoNewPrivileges scheiterte, waehrend die Seite "erreichbar" meldete.
check("Systemhelfer antwortet", "Helfer antwortet" in body, "Helfer meldet sich nicht")
check("Schluesselrechte in Ordnung", "in Ordnung" in body, "Problem gemeldet")
print("\nAbmeldung:")
token = client.token("/konto/")
status, body = client.post("/logout", {"csrf_token": token})
status = client.get_raw("/clients/")
check("Nach Abmeldung gesperrt", status in (302, 401), f"HTTP {status}")
print(f"\n== Ergebnis: {len(PASSED)} bestanden, {len(FAILED)} fehlgeschlagen ==")
for entry in FAILED:
print(f" FEHLER: {entry}")
return 1 if FAILED else 0
if __name__ == "__main__":
if len(sys.argv) != 4:
print(__doc__)
raise SystemExit(2)
raise SystemExit(main(sys.argv[1], sys.argv[2], sys.argv[3]))
tests/test_core_security.py
+721
View File
@@ -0,0 +1,721 @@
"""Tests der Sicherheitsgrenzen des Kerns.
Jeder Test hier deckt eine Grenze ab, die im Vorgaengerprojekt entweder fehlte
oder nur im Template durchgesetzt wurde.
"""
from __future__ import annotations
import json
import sqlite3
from datetime import datetime, timedelta, timezone
import pytest
from tesm_core import audit
from tesm_core.db import Database, NestedConnectionError
from tesm_core.extension import core
from tesm_core.keystore import DecryptionError, Keystore
from tesm_core.migrations import split_statements
from tesm_core.rbac import service as rbac
from tesm_core.rbac.model import Action, Area, PermissionTree, Resource, RolePreset
from tesm_core.security import passwords, totp
# --------------------------------------------------------------------------- #
# Rechtebaum
# --------------------------------------------------------------------------- #
def _tree() -> PermissionTree:
return PermissionTree(
[
Area(
key="ops",
label="Betrieb",
icon="cpu",
resources=(
Resource(
"devices",
"Geraete",
(Action.VIEW, Action.CREATE, Action.EDIT, Action.DELETE, Action.EXECUTE),
),
Resource("switches", "Switche", (Action.VIEW, Action.EDIT)),
),
),
Area(
key="admin",
label="Verwaltung",
icon="sliders",
resources=(Resource("users", "Benutzer", (Action.VIEW, Action.CREATE)),),
),
]
)
def test_area_columns_only_contain_used_actions():
tree = _tree()
ops = tree.area("ops")
admin = tree.area("admin")
assert Action.EXECUTE in ops.columns()
assert Action.EXECUTE not in admin.columns()
assert Action.DELETE not in admin.columns()
def test_sanitize_adds_view_and_area_switch():
tree = _tree()
result = tree.sanitize(["devices.delete"])
assert result == ["devices.delete", "devices.view", "ops.view"]
def test_sanitize_drops_unknown_permissions():
tree = _tree()
assert tree.sanitize(["gibtsnicht.view", "devices.view"]) == ["devices.view", "ops.view"]
def test_kill_switch_disables_all_children():
tree = _tree()
granted = {"devices.view", "devices.edit", "switches.view"}
assert tree.effective(granted) == set() # ohne ops.view greift nichts
assert tree.effective(granted | {"ops.view"}) == granted | {"ops.view"}
def test_duplicate_resource_keys_rejected():
with pytest.raises(ValueError):
PermissionTree(
[
Area("a", "A", "grid", (Resource("x", "X", (Action.VIEW,)),)),
Area("b", "B", "grid", (Resource("x", "X", (Action.VIEW,)),)),
]
)
def test_resource_without_view_rejected():
with pytest.raises(ValueError):
PermissionTree([Area("a", "A", "grid", (Resource("x", "X", (Action.EDIT,)),))])
def test_preset_expansion():
tree = _tree()
preset = RolePreset(
key="ops-read",
name="Betrieb lesen",
description="",
grants=(("devices", (Action.VIEW,)), ("switches", "*")),
)
assert tree.expand_preset(preset) == [
"devices.view",
"ops.view",
"switches.edit",
"switches.view",
]
# --------------------------------------------------------------------------- #
# Eskalationsschutz
# --------------------------------------------------------------------------- #
class _Actor:
def __init__(self, is_admin: bool, permissions: set[str]):
self.is_admin = is_admin
self.id = 42
self._permissions = permissions
def has(self, *permissions: str) -> bool:
return self.is_admin or all(p in self._permissions for p in permissions)
def test_no_amplification_blocks_unknown_permission():
tree = _tree()
actor = _Actor(False, {"ops.view", "devices.view"})
with pytest.raises(rbac.PermissionDenied) as exc:
rbac.guard_no_amplification(actor, tree, ["devices.delete"])
assert "nicht besitzen" in str(exc.value)
def test_no_amplification_allows_own_permissions():
tree = _tree()
actor = _Actor(False, {"ops.view", "devices.view"})
rbac.guard_no_amplification(actor, tree, ["devices.view", "ops.view"])
def test_admin_may_grant_anything():
tree = _tree()
rbac.guard_no_amplification(_Actor(True, set()), tree, ["devices.delete"])
def test_group_editor_cannot_escalate_via_new_group(core_app, make_user):
"""Der zentrale Fall: ein Gruppenverwalter darf sich nicht selbst hoeherstufen."""
from tesm_core.auth.service import load_user
extension = core(core_app)
row = make_user("gruppenchef", "Ein-gutes-Passwort-1", permissions=("groups.create", "groups.view"))
with core_app.app_context(), extension.database.session() as conn:
actor = load_user(conn, int(row["id"]))
assert actor is not None
with extension.database.transaction(conn):
with pytest.raises(rbac.PermissionDenied):
rbac.create_group(
conn,
name="Hintertuer",
description="",
permissions=["users.create", "administration.view"],
tree=extension.permissions,
actor=actor,
)
def test_last_admin_cannot_be_demoted(core_app, make_user):
from tesm_core.auth.service import load_user
extension = core(core_app)
admin_row = make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
with core_app.app_context(), extension.database.session() as conn:
actor = load_user(conn, int(admin_row["id"]))
second = conn.execute(
"INSERT INTO users (username, password_hash, is_admin, auth_source, created_at, updated_at) "
"VALUES ('zweiter','x',1,'local',datetime('now'),datetime('now'))"
)
second_id = int(second.lastrowid or 0)
with extension.database.transaction(conn):
# Solange zwei Admins existieren, ist das Herabstufen erlaubt.
rbac.update_user(
conn, second_id, actor=actor, tree=extension.permissions, is_admin=False
)
with extension.database.transaction(conn):
with pytest.raises(rbac.PermissionDenied):
rbac.guard_last_admin(conn, int(admin_row["id"]), action="Herabstufen")
def test_cannot_change_own_admin_flag(core_app, make_user):
from tesm_core.auth.service import load_user
extension = core(core_app)
row = make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
with core_app.app_context(), extension.database.session() as conn:
actor = load_user(conn, int(row["id"]))
with extension.database.transaction(conn):
with pytest.raises(rbac.PermissionDenied):
rbac.update_user(
conn, int(row["id"]), actor=actor, tree=extension.permissions, is_admin=False
)
def test_non_admin_cannot_touch_admin_account(core_app, make_user):
from tesm_core.auth.service import load_user
extension = core(core_app)
admin_row = make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
editor_row = make_user("helfer", "Ein-gutes-Passwort-1", permissions=("users.edit", "users.view"))
with core_app.app_context(), extension.database.session() as conn:
actor = load_user(conn, int(editor_row["id"]))
with extension.database.transaction(conn):
with pytest.raises(rbac.PermissionDenied):
rbac.update_user(
conn,
int(admin_row["id"]),
actor=actor,
tree=extension.permissions,
is_locked=True,
)
def test_system_group_permissions_locked(core_app, make_user):
from tesm_core.auth.service import load_user
extension = core(core_app)
row = make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
with core_app.app_context(), extension.database.session() as conn:
actor = load_user(conn, int(row["id"]))
default = next(g for g in rbac.list_groups(conn) if g.is_default)
with extension.database.transaction(conn):
with pytest.raises(rbac.PermissionDenied):
rbac.update_group(
conn,
default.id,
permissions=[],
tree=extension.permissions,
actor=actor,
)
# Mit ausdruecklicher Freischaltung geht es.
rbac.update_group(
conn,
default.id,
permissions=[],
tree=extension.permissions,
actor=actor,
unlock_system=True,
)
def test_presets_are_not_reset_on_restart(core_app):
"""Anders als im Vorgaenger bleiben Anpassungen an Systemgruppen erhalten."""
extension = core(core_app)
with extension.database.session() as conn:
default = next(g for g in rbac.list_groups(conn) if g.is_default)
with extension.database.transaction(conn):
conn.execute("DELETE FROM group_permissions WHERE group_id=?", (default.id,))
created = rbac.ensure_presets(conn, extension.permissions, extension.role_presets)
assert created == []
assert rbac.group_permissions(conn, default.id) == []
# --------------------------------------------------------------------------- #
# Datenbank
# --------------------------------------------------------------------------- #
def test_second_connection_during_write_transaction_raises(tmp_path):
database = Database(tmp_path / "test.db")
with database.session() as conn:
conn.execute("CREATE TABLE t (a INTEGER)")
with database.transaction(conn):
conn.execute("INSERT INTO t VALUES (1)")
with pytest.raises(NestedConnectionError) as exc:
database.connect()
assert "Schreibtransaktion" in str(exc.value)
def test_nested_transactions_commit_once(tmp_path):
database = Database(tmp_path / "test.db")
with database.session() as conn:
conn.execute("CREATE TABLE t (a INTEGER)")
with database.transaction(conn):
conn.execute("INSERT INTO t VALUES (1)")
with database.transaction(conn):
conn.execute("INSERT INTO t VALUES (2)")
assert Database.value(conn, "SELECT COUNT(*) FROM t") == 2
def test_inner_rollback_keeps_outer(tmp_path):
database = Database(tmp_path / "test.db")
with database.session() as conn:
conn.execute("CREATE TABLE t (a INTEGER)")
with database.transaction(conn):
conn.execute("INSERT INTO t VALUES (1)")
try:
with database.transaction(conn):
conn.execute("INSERT INTO t VALUES (2)")
raise RuntimeError("abbruch")
except RuntimeError:
pass
assert Database.value(conn, "SELECT COUNT(*) FROM t") == 1
def test_foreign_keys_enforced(tmp_path):
database = Database(tmp_path / "test.db")
with database.session() as conn:
conn.execute("CREATE TABLE parent (id INTEGER PRIMARY KEY)")
conn.execute("CREATE TABLE child (p INTEGER REFERENCES parent(id))")
with pytest.raises(sqlite3.IntegrityError):
conn.execute("INSERT INTO child VALUES (99)")
def test_wal_mode_active(tmp_path):
database = Database(tmp_path / "test.db")
with database.session() as conn:
assert Database.value(conn, "PRAGMA journal_mode").lower() == "wal"
def test_split_statements_handles_comments_and_strings():
script = """
-- ein Kommentar mit ; Semikolon
CREATE TABLE t (a TEXT DEFAULT 'x;y');
INSERT INTO t VALUES ('a;b'); /* Block; Kommentar */
"""
statements = split_statements(script)
assert len(statements) == 2
assert "x;y" in statements[0]
assert "a;b" in statements[1]
# --------------------------------------------------------------------------- #
# Schluessel und Verschluesselung
# --------------------------------------------------------------------------- #
def test_keystore_roundtrip_and_aad_binding(tmp_path):
keystore = Keystore(tmp_path / "data.keys")
token = keystore.encrypt("geheim", aad="credentials.password:1")
assert keystore.decrypt(token, aad="credentials.password:1") == "geheim"
# Dasselbe Chiffrat an anderer Stelle eingespielt -> Fehler.
with pytest.raises(DecryptionError):
keystore.decrypt(token, aad="credentials.password:2")
def test_keystore_rotation_keeps_old_values_readable(tmp_path):
keystore = Keystore(tmp_path / "data.keys")
old_token = keystore.encrypt("alt", aad="x")
new_id = keystore.rotate()
assert new_id == "2"
assert keystore.decrypt(old_token, aad="x") == "alt"
assert keystore.needs_reencrypt(old_token) is True
new_token = keystore.encrypt("neu", aad="x")
assert keystore.needs_reencrypt(new_token) is False
def test_keystore_file_is_private(tmp_path):
import os
import stat
path = tmp_path / "data.keys"
Keystore(path)
if os.name == "posix":
mode = path.stat().st_mode
assert not mode & (stat.S_IRGRP | stat.S_IROTH)
def test_tampered_ciphertext_rejected(tmp_path):
keystore = Keystore(tmp_path / "data.keys")
token = keystore.encrypt("geheim", aad="x")
prefix, key_id, nonce, ciphertext = token.split(".", 3)
broken = f"{prefix}.{key_id}.{nonce}.{ciphertext[:-4]}AAAA"
with pytest.raises(DecryptionError):
keystore.decrypt(broken, aad="x")
# --------------------------------------------------------------------------- #
# Passwoerter und TOTP
# --------------------------------------------------------------------------- #
def test_argon2_hash_and_verify():
stored = passwords.hash_password("Ein-gutes-Passwort-1")
assert stored.startswith("$argon2id$")
assert passwords.verify_password(stored, "Ein-gutes-Passwort-1") is True
assert passwords.verify_password(stored, "falsch") is False
assert passwords.needs_rehash(stored) is False
def test_bcrypt_legacy_hash_accepted_and_flagged_for_rehash():
bcrypt = pytest.importorskip("bcrypt")
legacy = bcrypt.hashpw(b"Altes-Passwort-1", bcrypt.gensalt(rounds=4)).decode()
assert passwords.is_legacy_hash(legacy) is True
assert passwords.verify_password(legacy, "Altes-Passwort-1") is True
assert passwords.needs_rehash(legacy) is True
def test_password_normalisation_across_platforms():
import unicodedata
nfd = unicodedata.normalize("NFD", "Grüße-aus-München-1")
nfc = unicodedata.normalize("NFC", "Grüße-aus-München-1")
assert nfd != nfc
stored = passwords.hash_password(nfd)
assert passwords.verify_password(stored, nfc) is True
def test_password_policy_rules():
policy = passwords.PasswordPolicy(min_length=12)
assert policy.check("Ein-gutes-Passwort-1") == []
assert any("Zeichen erforderlich" in p for p in policy.check("kurz"))
assert any("Zeichenarten" in p for p in policy.check("aaaaaaaaaaaaaaa"))
assert any("Benutzernamen" in p for p in policy.check("Tester-Passwort-1", username="tester"))
assert any("haeufig" in p for p in policy.check("passwort1234"))
def test_totp_generates_and_verifies():
secret = totp.generate_secret()
code = totp.current_code(secret)
assert totp.verify_code(secret, code) is True
assert totp.verify_code(secret, "000000") is False
assert totp.verify_code(secret, code, now=0) is False
def test_totp_drift_tolerance():
secret = totp.generate_secret()
now = 1_700_000_000
previous = totp.code_at(secret, int(now // totp.PERIOD) - 1)
assert totp.verify_code(secret, previous, now=now) is True
far = totp.code_at(secret, int(now // totp.PERIOD) - 5)
assert totp.verify_code(secret, far, now=now) is False
def test_totp_uri_contains_issuer_and_secret():
secret = totp.generate_secret()
uri = totp.provisioning_uri(secret, account="anna", issuer="TESM Test")
assert uri.startswith("otpauth://totp/")
assert secret in uri
assert "issuer=TESM%20Test" in uri
# --------------------------------------------------------------------------- #
# Aenderungsprotokoll
# --------------------------------------------------------------------------- #
def test_audit_chain_detects_modification(core_app):
extension = core(core_app)
with extension.database.session() as conn:
with extension.database.transaction(conn):
for index in range(5):
audit.log(conn, actor="tester", action="test.event", target=f"ziel-{index}")
assert audit.verify_chain(conn)["ok"] is True
with extension.database.transaction(conn):
conn.execute("UPDATE audit_log SET detail='manipuliert' WHERE id=3")
result = audit.verify_chain(conn)
assert result["ok"] is False
assert result["broken_at"] == 3
assert "Hash" in result["reason"]
def test_audit_chain_detects_deletion(core_app):
extension = core(core_app)
with extension.database.session() as conn:
with extension.database.transaction(conn):
for index in range(5):
audit.log(conn, actor="tester", action="test.event", target=str(index))
with extension.database.transaction(conn):
conn.execute("DELETE FROM audit_log WHERE id=3")
result = audit.verify_chain(conn)
assert result["ok"] is False
def test_audit_archiving_moves_full_days(core_app, tmp_path):
extension = core(core_app)
old_day = (datetime.now(timezone.utc) - timedelta(days=3)).replace(microsecond=0)
with extension.database.session() as conn:
with extension.database.transaction(conn):
for index in range(40):
audit.log(
conn,
actor="tester",
action="test.old",
target=str(index),
ts=old_day + timedelta(seconds=index),
)
for index in range(5):
audit.log(conn, actor="tester", action="test.new", target=str(index))
with extension.database.transaction(conn):
written = audit.archive_old_rows(conn, tmp_path / "archiv", threshold=10, target=5)
assert written
# Uebrig bleiben ausschliesslich Zeilen des laufenden Tages (inkl. der
# Startzeile, die die App beim Migrieren selbst geschrieben hat).
today = datetime.now(timezone.utc).date().isoformat()
remaining = Database.all(conn, "SELECT DISTINCT substr(ts,1,10) AS day FROM audit_log")
assert [row["day"] for row in remaining] == [today]
lines = (tmp_path / "archiv" / written[0].name).read_text(encoding="utf-8").splitlines()
assert len(lines) == 40
assert json.loads(lines[0])["action"] == "test.old"
# --------------------------------------------------------------------------- #
# Sitzungen
# --------------------------------------------------------------------------- #
def test_session_id_rotates_on_login(client, make_user, login):
make_user("anna", "Ein-gutes-Passwort-1")
client.get("/login")
before = client.get_cookie("tesm_session")
login("anna", "Ein-gutes-Passwort-1")
after = client.get_cookie("tesm_session")
assert after is not None
if before is not None:
assert before.value != after.value
def test_session_revocation_takes_effect_immediately(client, core_app, make_user, login):
from tesm_core.security import sessions as session_store
row = make_user("anna", "Ein-gutes-Passwort-1")
login("anna", "Ein-gutes-Passwort-1")
assert client.get("/konto/").status_code == 200
extension = core(core_app)
with extension.database.session() as conn, extension.database.transaction(conn):
session_store.revoke_all_for_user(conn, int(row["id"]), reason="test")
response = client.get("/konto/")
assert response.status_code == 302
assert "/login" in response.headers["Location"]
def test_password_change_revokes_other_sessions(core_app, make_user):
from tesm_core.security import sessions as session_store
extension = core(core_app)
row = make_user("anna", "Ein-gutes-Passwort-1")
first = core_app.test_client()
second = core_app.test_client()
for client_instance in (first, second):
page = client_instance.get("/login")
body = page.get_data(as_text=True)
marker = 'name="csrf_token" value="'
start = body.index(marker) + len(marker)
token = body[start : body.index('"', start)]
client_instance.post(
"/login",
data={"username": "anna", "password": "Ein-gutes-Passwort-1", "csrf_token": token},
follow_redirects=True,
)
with extension.database.session() as conn:
assert len(session_store.active_for_user(conn, int(row["id"]))) == 2
page = first.get("/konto/sicherheit")
body = page.get_data(as_text=True)
marker = 'name="csrf_token" value="'
start = body.index(marker) + len(marker)
token = body[start : body.index('"', start)]
first.post(
"/konto/passwort",
data={
"current_password": "Ein-gutes-Passwort-1",
"new_password": "Neues-Passwort-2026",
"repeat_password": "Neues-Passwort-2026",
"csrf_token": token,
},
follow_redirects=True,
)
assert second.get("/konto/").status_code == 302
assert first.get("/konto/").status_code == 200
def test_reauth_required_for_sensitive_action(client, make_user, login, csrf_token):
make_user("anna", "Ein-gutes-Passwort-1")
login("anna", "Ein-gutes-Passwort-1")
# Frisch angemeldet gilt die Bestaetigung -- danach kuenstlich altern lassen.
from flask import session as flask_session
with client.session_transaction() as sess:
sess["_reauth_at"] = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat(
timespec="seconds"
)
token = csrf_token("/konto/sicherheit")
response = client.post("/konto/zwei-faktor/starten", data={"csrf_token": token})
assert response.status_code == 302
assert "/bestaetigen" in response.headers["Location"]
assert flask_session is not None
# --------------------------------------------------------------------------- #
# Der Rechteweg muss tatsaechlich begehbar sein
# --------------------------------------------------------------------------- #
def test_units_do_not_disable_the_privilege_path():
"""``NoNewPrivileges`` schaltet setuid ab -- und damit sudo, und damit den Helfer.
Auf beiden Testhosts stand "Helfer erreichbar", waehrend in Wahrheit jede
Systemaktion an ``sudo: the "no new privileges" flag is set`` scheiterte:
nginx uebernehmen, Netzwerk aendern, Zertifikat anfordern, Kea schreiben.
Die Haertung hatte genau den Mechanismus abgeschaltet, den sie schuetzen
sollte.
"""
from pathlib import Path
root = Path(__file__).resolve().parent.parent / "deploy" / "systemd"
units = list(root.glob("*.service"))
assert units
for unit in units:
text = unit.read_text(encoding="utf-8")
assert "NoNewPrivileges=yes" not in text, (
f"{unit.name}: verhindert sudo und damit jede Systemaktion."
)
# Gemessen: auch ein eingeschraenktes Bounding Set bricht sudo mit
# "unable to change to root gid" -- inklusive CapabilityBoundingSet=CAP_NET_RAW.
for line in text.splitlines():
if line.startswith("CapabilityBoundingSet="):
raise AssertionError(
f"{unit.name}: {line!r} nimmt sudo CAP_SETUID/CAP_SETGID."
)
# Die eigentliche Grenze bleibt bestehen.
# "strict" sperrt auch den Helfer aus /etc aus -- gemessen.
assert "ProtectSystem=full" in text, unit.name
assert "ProtectSystem=strict" not in text, (
f"{unit.name}: strict sperrt den Helfer aus /etc aus."
)
assert "PrivateTmp=yes" in text, unit.name
assert "User=root" not in text, unit.name
def test_the_helper_is_probed_not_just_looked_for():
""""Datei liegt da" ist keine Auskunft darueber, ob der Aufruf gelingt."""
import inspect
from tesm_core.sysops import SysOps
assert hasattr(SysOps, "selftest")
source = inspect.getsource(SysOps.selftest)
assert 'self.run("service-status", "nginx")' in source
assert 'self.run("nginx-test")' not in source, (
"Ein fehlgeschlagenes nginx -t heisst 'Konfiguration kaputt', nicht 'Helfer kaputt'."
)
# Die beiden Fehlerbilder, die im Betrieb wirklich vorkommen, werden benannt.
assert "no new privileges" in source
assert "a password is required" in source
def test_selftest_explains_a_blocked_sudo(monkeypatch):
from tesm_core.sysops import Result, SysOps
class Blocked(SysOps):
@property
def available(self) -> bool:
return True
def run(self, verb, *args, stdin="", timeout=None):
return Result(
False, "", 'sudo: The "no new privileges" flag is set, which prevents sudo', 1
)
result = Blocked(app_key="tesm").selftest()
assert not result.ok
assert "NoNewPrivileges" in result.message
assert "systemctl" in result.message
def test_selftest_reports_a_missing_sudoers_rule():
from tesm_core.sysops import Result, SysOps
class NeedsPassword(SysOps):
@property
def available(self) -> bool:
return True
def run(self, verb, *args, stdin="", timeout=None):
return Result(False, "", "sudo: a password is required", 1)
result = NeedsPassword(app_key="tesm").selftest()
assert not result.ok
assert "sudoers" in result.message
def test_selftest_says_so_when_the_helper_is_absent(tmp_path):
from tesm_core.sysops import SysOps
result = SysOps(app_key="tesm", helper=tmp_path / "gibt-es-nicht").selftest()
assert not result.ok
assert "fehlt" in result.message
def test_authenticate_keeps_directory_groups(core_app, monkeypatch):
"""Die Verzeichnisgruppen duerfen auf dem Weg nach oben nicht verlorengehen.
``_verify_credentials`` hat sie korrekt geliefert, ``authenticate`` hat aus
dem Ergebnis aber ein neues Objekt gebaut und das Feld dabei fallen lassen.
Sichtbar war davon nichts: die Anmeldung gelang, nur band sich keine
Freigabe mehr ein. Genau solche stillen Verluste faengt dieser Test.
"""
from tesm_core.auth import directory, service
from tesm_core.extension import core
profil = directory.DirectoryProfile(
username="mitarbeiter",
dn="CN=mitarbeiter,OU=Benutzer,DC=firma,DC=local",
group_dns=("CN=GG_Technik,OU=Gruppen,DC=firma,DC=local",),
)
monkeypatch.setattr(directory, "is_enabled", lambda conn: True)
monkeypatch.setattr(
directory, "authenticate", lambda conn, name, pw: directory.DirectoryResult("ok", profile=profil)
)
with core_app.test_request_context("/login"):
with core(core_app).database.session() as conn:
outcome = service.authenticate(conn, "mitarbeiter", "egal", ip="127.0.0.1")
assert outcome.ok, outcome.message
assert outcome.group_dns == profil.group_dns, "die Gruppen sind unterwegs verlorengegangen"
tests/test_core_smoke.py
+246
View File
@@ -0,0 +1,246 @@
"""Rauchtest des Grundgeruests: Start, Anmeldung, Rechte, CSRF, Header."""
from __future__ import annotations
import re
def test_app_starts_and_health_ok(client):
response = client.get("/gesundheit")
assert response.status_code == 200
assert response.get_json()["status"] == "ok"
def test_login_page_renders_and_sets_csrf(client):
response = client.get("/login")
assert response.status_code == 200
body = response.get_data(as_text=True)
assert "Anmelden" in body
assert 'name="csrf_token"' in body
def test_security_headers_present(client):
response = client.get("/login")
csp = response.headers["Content-Security-Policy"]
assert "default-src 'self'" in csp
assert "'unsafe-inline'" not in csp.split("script-src")[1].split(";")[0]
assert re.search(r"nonce-[A-Za-z0-9_-]{10,}", csp)
assert response.headers["X-Frame-Options"] == "DENY"
assert response.headers["X-Content-Type-Options"] == "nosniff"
assert "camera=()" in response.headers["Permissions-Policy"]
def test_login_and_logout(client, make_user, login):
make_user("anna", "Ein-gutes-Passwort-1")
response = login("anna", "Ein-gutes-Passwort-1")
assert response.status_code == 200
assert "Willkommen" in response.get_data(as_text=True)
page = client.get("/konto/")
assert page.status_code == 200
assert "anna" in page.get_data(as_text=True)
def test_wrong_password_is_generic(client, make_user, login):
make_user("anna", "Ein-gutes-Passwort-1")
response = login("anna", "falsch")
body = response.get_data(as_text=True)
assert "Benutzername oder Passwort ist falsch" in body
# Kein Hinweis darauf, dass das Konto existiert.
assert "gesperrt" not in body
def test_unknown_user_same_message(client, login):
body = login("gibtsnicht", "irgendwas").get_data(as_text=True)
assert "Benutzername oder Passwort ist falsch" in body
def test_locked_account_rejected(client, make_user, login):
make_user("gesperrt", "Ein-gutes-Passwort-1", is_locked=True)
body = login("gesperrt", "Ein-gutes-Passwort-1").get_data(as_text=True)
assert "gesperrt" in body
def test_rate_limit_blocks_after_many_attempts(client, make_user, login):
make_user("anna", "Ein-gutes-Passwort-1")
messages = [login("anna", "falsch").get_data(as_text=True) for _ in range(9)]
assert any("voruebergehend gesperrt" in body for body in messages)
def test_protected_page_redirects_to_login(client):
response = client.get("/widgets")
assert response.status_code == 302
assert "/login" in response.headers["Location"]
def test_permission_required_blocks_without_grant(client, make_user, login):
make_user("anna", "Ein-gutes-Passwort-1")
login("anna", "Ein-gutes-Passwort-1")
assert client.get("/verwaltung/benutzer").status_code == 403
def test_permission_granted_allows_access(client, make_user, login):
make_user("bob", "Ein-gutes-Passwort-1", permissions=("users.view",))
login("bob", "Ein-gutes-Passwort-1")
response = client.get("/verwaltung/benutzer")
assert response.status_code == 200
assert "Benutzer" in response.get_data(as_text=True)
def test_admin_sees_everything(client, make_user, login):
make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
login("root", "Ein-gutes-Passwort-1")
for path in ("/verwaltung/benutzer", "/verwaltung/gruppen", "/papierkorb/", "/diagnose/"):
assert client.get(path).status_code == 200, path
def test_csrf_missing_token_is_rejected(client, make_user, login):
make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
login("root", "Ein-gutes-Passwort-1")
response = client.post("/verwaltung/benutzer/neu", data={"username": "boese"})
assert response.status_code == 403
assert "Sicherheitstoken" in response.get_data(as_text=True)
def test_csrf_wrong_origin_is_rejected(client, make_user, login, csrf_token):
make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
login("root", "Ein-gutes-Passwort-1")
token = csrf_token("/verwaltung/benutzer")
response = client.post(
"/verwaltung/benutzer/neu",
data={"username": "boese", "csrf_token": token},
headers={"Origin": "https://angreifer.example"},
)
assert response.status_code == 403
def test_nav_hides_pages_without_permission(client, make_user, login):
make_user("anna", "Ein-gutes-Passwort-1")
login("anna", "Ein-gutes-Passwort-1")
body = client.get("/").get_data(as_text=True)
assert "Diagnose" not in body
assert "Widgets" not in body
def test_nav_shows_pages_with_permission(client, make_user, login):
make_user("bob", "Ein-gutes-Passwort-1", permissions=("widgets.view",))
login("bob", "Ein-gutes-Passwort-1")
body = client.get("/").get_data(as_text=True)
assert "Widgets" in body
def test_default_group_created_from_preset(core_app):
from tesm_core.extension import core
from tesm_core.rbac import service as rbac
extension = core(core_app)
with extension.database.session() as conn:
groups = rbac.list_groups(conn)
names = {group.name for group in groups}
assert "Benutzer" in names
default = next(group for group in groups if group.name == "Benutzer")
assert default.is_default is True
assert "widgets.view" in default.permissions
assert "demo.view" in default.permissions # Bereichsschalter wird mitgesetzt
def test_error_page_shows_request_id(client):
response = client.get("/gibt-es-nicht")
assert response.status_code == 404
body = response.get_data(as_text=True)
assert "404" in body
assert "Vorgangsnummer" in body
# --------------------------------------------------------------------------- #
# Verzeichnisgruppen: suchen statt abtippen
# --------------------------------------------------------------------------- #
def test_group_search_is_open_to_fileshare_editors():
"""Wer Freigaben zuordnet, muss die Gruppen finden koennen.
Sonst braeuchte er zusaetzlich das Recht, den Verzeichnisdienst zu
konfigurieren -- eine Rechteausweitung fuer eine reine Lesesuche.
"""
import inspect
from tesm_core.views import hostsettings
source = inspect.getsource(hostsettings.directory_group_search)
assert "fileshare.edit" in source
assert 'mode="any"' in source or "mode='any'" in source
def test_the_picker_never_forces_typing_a_dn():
"""Ein abgetippter DN ist ein Tippfehler, der erst bei der Anmeldung auffaellt."""
from pathlib import Path
root = Path(__file__).resolve().parent.parent
picker = (
root / "packages" / "tesm-core" / "src" / "tesm_core" / "templates"
/ "tesm_core" / "_group_picker.html"
).read_text(encoding="utf-8")
assert "directory_group_search" in picker
assert "data-group-query" in picker
# Tippen bleibt trotzdem moeglich: ohne Dienstkonto gibt es keine Suche.
assert 'name="{{ dn_field }}"' in picker
for template in (
root / "packages" / "tesm-core" / "src" / "tesm_core" / "templates"
/ "tesm_core" / "settings_directory.html",
root / "apps" / "tesm" / "src" / "tesm" / "templates" / "tesm" / "fileshare_mappings.html",
):
text = template.read_text(encoding="utf-8")
assert "group_picker(" in text, template.name
def test_transport_choice_carries_its_port():
"""Umgestellter Transport mit altem Port ergibt einen Fehler ohne Aussage."""
from pathlib import Path
root = Path(__file__).resolve().parent.parent
page = (
root / "packages" / "tesm-core" / "src" / "tesm_core" / "templates"
/ "tesm_core" / "settings_directory.html"
).read_text(encoding="utf-8")
assert 'data-behavior="transport-port"' in page
assert 'data-port-target="#ldap-port"' in page
script = (
root / "packages" / "tesm-core" / "src" / "tesm_core" / "static"
/ "tesm_core" / "js" / "app.js"
).read_text(encoding="utf-8")
assert "transport-port" in script
assert 'ldaps: "636"' in script
assert 'starttls: "389"' in script
# Ein absichtlich abweichender Port darf nicht ueberschrieben werden.
assert "known.includes(port.value)" in script
def test_permission_matrix_has_no_nameless_checkbox(client, make_user, login):
"""In der Rechtematrix darf kein Kaestchen ohne ``name`` stehen.
Fuer nicht vorgesehene Aktionen stand dort ein deaktiviertes Kaestchen.
Das JS hat beim Einschalten des Bereichs *alle* Kaestchen der Zeile
freigegeben -- auch dieses. Es liess sich anhaken, uebertrug wegen des
fehlenden ``name`` aber nichts: es sah erteilt aus und war nach dem
Speichern verschwunden. Ohne Eingabefeld kann das nicht wieder passieren.
"""
import re
make_user("root", "Ein-gutes-Passwort-1", is_admin=True)
login("root", "Ein-gutes-Passwort-1")
body = client.get("/verwaltung/gruppen").get_data(as_text=True)
if "permission-matrix" not in body:
gruppe = re.search(r'href="(/verwaltung/gruppen/\d+)"', body)
assert gruppe, "keine Gruppe zum Pruefen gefunden"
body = client.get(gruppe.group(1)).get_data(as_text=True)
assert "permission-matrix" in body, "Rechtematrix nicht gefunden"
kaestchen = re.findall(r"<input[^>]*type=\"checkbox\"[^>]*>", body)
assert kaestchen, "keine Kaestchen in der Matrix"
ohne_namen = [k for k in kaestchen if "name=" not in k]
assert not ohne_namen, f"Kaestchen ohne name: {ohne_namen[:3]}"
# Der Platzhalter ist jetzt reine Darstellung.
assert "matrix__na" in body
tests/test_legacy_transfer.py
+279
View File
@@ -0,0 +1,279 @@
"""Der Weg vom alten TESM in dieses: Uebersetzer und Import zusammen.
Die Testdatei wird **genau so** gebaut, wie der Vorgaenger sie schreibt
(PBKDF2-HMAC-SHA256, 390000 Runden, Fernet, derselbe Umschlag). Damit prueft
der Test das echte Format und nicht meine Vorstellung davon.
"""
from __future__ import annotations
import base64
import json
from typing import Any
import pytest
pytest.importorskip("cryptography")
LEGACY_PASS = "alte-passphrase-2026"
NEW_PASS = "neue-passphrase-2026"
@pytest.fixture()
def tesm_app(instance_env: Any):
from tesm import create_app
return create_app(TESTING=True)
def _seal_legacy(payload: dict[str, Any], passphrase: str) -> str:
"""Baut einen Umschlag wie ``export_data()`` im alten TESM."""
import secrets
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
salt = secrets.token_bytes(16)
kdf = PBKDF2HMAC(algorithm=hashes.SHA256(), length=32, salt=salt, iterations=390000)
key = base64.urlsafe_b64encode(kdf.derive(passphrase.encode("utf-8")))
token = Fernet(key).encrypt(json.dumps(payload).encode("utf-8"))
return json.dumps(
{
"app": "tesm",
"version": 2,
"exported_at": "2026-09-02 12:00:00",
"salt": base64.urlsafe_b64encode(salt).decode("utf-8"),
"payload": token.decode("utf-8"),
}
)
#: Ein Ausschnitt, der die Eigenheiten des alten Formats abdeckt: Switch- und
#: Zugangsdatenbezug per Name, geschachtelte DHCP-Optionswerte, "name" statt
#: "hostname" an der Reservierung, sowie Abschnitte, die nicht uebersetzt
#: werden koennen.
LEGACY_PAYLOAD: dict[str, Any] = {
"credentials": [
{"name": "Switch-Admin", "username": "manager", "password": "geheim-1", "category": "switch"},
{"name": "Linux-Root", "username": "root", "password": "geheim-2", "category": "linux"},
{"name": "Windows-Dienst", "username": "svc", "password": "geheim-3", "category": "windows"},
],
"switches": [
{"hostname": "sw-keller", "ip": "192.168.80.2", "ssh_port": 22, "credential_name": "Switch-Admin"},
],
"devices": [
{
"mac": "4c:52:62:25:b9:e4",
"ip": "192.168.80.137",
"port": "12",
"name": "Autodarts",
"switch_hostname": "sw-keller",
"is_active": 1,
"ssh_port": 22,
"credential_name": "Linux-Root",
},
{
"mac": "aa:bb:cc:00:00:02",
"ip": "192.168.80.50",
"port": "3",
"name": "Kasse",
"switch_hostname": "sw-keller",
"is_active": 1,
"ssh_port": 22,
"credential_name": "Windows-Dienst",
},
],
"dhcp": {
"settings": {"dhcp_domain": "firma.local", "dhcp_lease_default": "1200"},
"subnets": [
{
"interface": "eth0",
"range_start": "192.168.80.100",
"range_end": "192.168.80.200",
"gateway": "192.168.80.1",
"dns": "192.168.80.1",
"sort_order": 0,
"enabled": 1,
}
],
"reservations": [{"mac": "aa:bb:cc:00:00:09", "ip": "192.168.80.90", "name": "drucker"}],
"options": [
{
"code": 225,
"name": "url",
"type": "string",
"description": "Terminal-URL",
"is_standard": 0,
"values": [
{"device_mac": "4c:52:62:25:b9:e4", "value": "https://play.autodarts.com"},
{"device_mac": None, "value": "https://intranet.firma.local"},
],
},
# Ohne Wert gibt es nichts zu senden -- wird gemeldet, nicht uebernommen.
{"code": 226, "name": "leer", "type": "string", "description": "", "is_standard": 0, "values": []},
],
},
# Diese Abschnitte kann der Uebersetzer nicht abbilden.
"users": [{"username": "anna", "password_hash": "$2b$12$abcdefghijklmnopqrstuv"}],
"known_hosts": "192.168.80.2 ssh-ed25519 AAAA...\n",
}
def test_legacy_envelope_needs_the_right_passphrase():
from tesm.services.legacy_transfer import LegacyTransferError, open_legacy
datei = _seal_legacy({"credentials": []}, LEGACY_PASS)
assert open_legacy(datei, LEGACY_PASS) == {"credentials": []}
with pytest.raises(LegacyTransferError, match="Passphrase"):
open_legacy(datei, "falsch")
def test_legacy_envelope_of_another_app_is_refused():
from tesm.services.legacy_transfer import LegacyTransferError, open_legacy
fremd = json.dumps({"app": "etwas-anderes", "salt": "AAAA", "payload": "x"})
with pytest.raises(LegacyTransferError, match="nicht zu TESM"):
open_legacy(fremd, LEGACY_PASS)
def test_report_names_what_it_cannot_translate():
"""Nicht uebersetzbare Abschnitte muessen benannt werden, nicht verschwinden."""
from tesm.services.legacy_transfer import convert, open_legacy
_, bericht = convert(open_legacy(_seal_legacy(LEGACY_PAYLOAD, LEGACY_PASS), LEGACY_PASS))
text = bericht.as_text()
assert "bcrypt" in text, "der Hinweis zu lokalen Konten fehlt"
assert "Host-Schluessel" in text
assert "ohne Wert" in text, "die wertlose Option wurde nicht gemeldet"
assert bericht.counts["credentials"] == 3
assert bericht.counts["devices"] == 2
assert bericht.counts["dhcp-Optionswerte"] == 2
def test_missing_credentials_are_reported(tesm_app):
"""Ein zugeordneter, aber nicht exportierter Zugang bleibt beim Import leer."""
from tesm.services.legacy_transfer import convert, open_legacy
ohne = {key: value for key, value in LEGACY_PAYLOAD.items() if key != "credentials"}
_, bericht = convert(open_legacy(_seal_legacy(ohne, LEGACY_PASS), LEGACY_PASS))
hinweise = " ".join(bericht.notes)
assert "Switch-Admin" in hinweise and "Linux-Root" in hinweise
def test_full_round_trip_into_the_database(tesm_app):
"""Alte Datei -> Uebersetzer -> neues Format -> vorhandener Import."""
from tesm.services import transfer
from tesm.services.legacy_transfer import convert, open_legacy
from tesm_core.extension import core
ext = core(tesm_app)
nutzlast, _ = convert(open_legacy(_seal_legacy(LEGACY_PAYLOAD, LEGACY_PASS), LEGACY_PASS))
# Ueber Siegel und Entsiegeln, damit auch der neue Umschlag geprueft ist.
datei = transfer.seal(nutzlast, NEW_PASS)
entsiegelt, _kopf = transfer.unseal(datei, NEW_PASS)
with ext.database.session() as conn, ext.database.transaction(conn):
transfer.apply_import(
conn,
ext.keystore,
entsiegelt,
categories=("credentials", "switches", "devices", "dhcp", "settings"),
set_setting=lambda key, value: ext.settings.set(conn, key, value, actor="test"),
)
with ext.database.session() as conn:
# -- Zugangsdaten: Passwort muss wieder lesbar sein ------------------
from tesm.services import inventory
zugang = {c.name: c for c in inventory.list_credentials(conn)}
assert set(zugang) == {"Switch-Admin", "Linux-Root", "Windows-Dienst"}
assert zugang["Linux-Root"].category == "linux"
benutzer, passwort = inventory.credential_secret(
conn, ext.keystore, zugang["Linux-Root"].id
)
assert (benutzer, passwort) == ("root", "geheim-2")
# -- Switch und Verknuepfung ----------------------------------------
switch = conn.execute("SELECT * FROM switches WHERE hostname='sw-keller'").fetchone()
assert switch is not None
assert switch["ip"] == "192.168.80.2"
assert switch["credential_id"] == zugang["Switch-Admin"].id
# -- Clients: Switch, Port, Zugang, abgeleitete Kategorie ------------
geraet = conn.execute(
"SELECT * FROM devices WHERE mac='4c:52:62:25:b9:e4'"
).fetchone()
assert geraet["name"] == "Autodarts"
assert geraet["switch_id"] == switch["id"]
assert geraet["port"] == "12"
assert geraet["credential_id"] == zugang["Linux-Root"].id
assert geraet["category"] == "linux", "Kategorie aus den Zugangsdaten abgeleitet"
windows = conn.execute("SELECT * FROM devices WHERE mac='aa:bb:cc:00:00:02'").fetchone()
assert windows["category"] == "windows"
# -- DHCP: Subnetz, Reservierung, Optionswerte ----------------------
subnetz = conn.execute("SELECT * FROM dhcp_subnets").fetchone()
assert (subnetz["range_start"], subnetz["range_end"]) == (
"192.168.80.100",
"192.168.80.200",
)
reservierung = conn.execute("SELECT * FROM dhcp_reservations").fetchone()
assert reservierung["hostname"] == "drucker", "alt 'name' -> neu 'hostname'"
option = conn.execute("SELECT * FROM dhcp_options WHERE code=225").fetchone()
assert option is not None and option["name"] == "url"
werte = {
row["device_id"]: row["value"]
for row in conn.execute(
"SELECT device_id, value FROM dhcp_option_values WHERE option_id=?",
(option["id"],),
)
}
# Ein globaler Wert und einer, der an genau diesem Geraet haengt.
assert werte[None] == "https://intranet.firma.local"
assert werte[geraet["id"]] == "https://play.autodarts.com"
# -- Einstellungen: nur die DHCP-Schluessel -------------------------
assert ext.settings.get(conn, "dhcp_domain") == "firma.local"
def test_option_value_of_an_unknown_device_is_not_made_global(tesm_app):
"""Ein Wert fuer ein unbekanntes Geraet darf nicht an alle gehen.
Sonst wuerde aus einer geraetespezifischen Terminal-URL beim Import eine,
die jeder Client bekommt -- die stillste Art, eine Konfiguration zu
verfaelschen.
"""
from tesm.services import transfer
from tesm_core.extension import core
ext = core(tesm_app)
nutzlast = {
"dhcp": {
"options": [
{
"code": 225,
"name": "url",
"type": "string",
"description": "",
"is_standard": 0,
"value": "https://nur-fuer-eines",
"device_mac": "ff:ff:ff:ff:ff:ff",
}
]
}
}
with ext.database.session() as conn, ext.database.transaction(conn):
transfer.apply_import(
conn,
ext.keystore,
nutzlast,
categories=("dhcp",),
set_setting=lambda key, value: None,
)
with ext.database.session() as conn:
assert conn.execute("SELECT COUNT(*) FROM dhcp_option_values").fetchone()[0] == 0
tests/test_licensing.py
+408
View File
@@ -0,0 +1,408 @@
"""Phase-0-Tests: das Lizenzprotokoll isoliert, ohne Flask, ohne Datenbank.
Deckt bewusst auch die Angriffsfaelle ab, die im Vorgaengerprojekt nicht
abgesichert waren: Signatur-Wiederverwendung ueber Domaenengrenzen, Replay
alter Anfragen/Antworten, Downgrade eines nachgelieferten Lizenz-Updates und
Ausstellung durch einen nicht gepinnten Aussteller.
"""
from __future__ import annotations
import copy
from datetime import timedelta
import pytest
import tesm_licensing as lic
VENDOR = {
"name": "WiS GmbH",
"phone": "+49 000 000",
"email": "lizenzen@example.invalid",
"address": "Musterweg 1, 12345 Musterstadt",
"logo_base64": "",
}
CUSTOMER = {"name": "Beispiel AG", "contact_email": "it@example.invalid", "customer_ref": "K-1"}
@pytest.fixture()
def master_keys() -> tuple[str, str]:
return lic.generate_keypair()
def issue(master_keys, **overrides):
priv, pub = master_keys
params = dict(
customer=CUSTOMER,
license_type="standard",
modules=[],
valid_days=365,
master_private_key_b64=priv,
master_public_key_b64=pub,
master_endpoint="https://lizenz.example.invalid/",
vendor=VENDOR,
)
params.update(overrides)
return lic.issue_license(**params)
# --------------------------------------------------------------------------- #
# Lizenzdatei
# --------------------------------------------------------------------------- #
def test_issued_license_verifies_and_splits(master_keys):
_, pub = master_keys
bundle, client_pub = issue(master_keys, modules=["dhcp", "maintenance"])
document, client_priv = lic.split_bundle(bundle)
lic.verify_license(document, trusted_master_keys=[pub])
assert document["license_pubkey"] == client_pub
assert lic.public_from_private(client_priv) == client_pub
# Der private Schluessel darf nicht im Lizenzdokument stehen.
assert "client_key" not in document
assert "license_privkey" not in document
status = lic.license_status(document, issuer_pinned=True)
assert status.modules == {"dhcp", "maintenance"}
assert status.modules_active is True
assert status.expired is False
assert status.days_left == 365
assert status.warnings == ()
def test_tampered_field_breaks_signature(master_keys):
bundle, _ = issue(master_keys)
document, _ = lic.split_bundle(bundle)
document["modules"] = ["dhcp", "fileshare", "maintenance"]
with pytest.raises(lic.SignatureInvalid):
lic.verify_license(document)
def test_swapped_master_key_is_detected(master_keys):
"""Ein Angreifer, der Payload *und* master_pubkey ersetzt, faellt am Pinning auf."""
bundle, _ = issue(master_keys)
document, _ = lic.split_bundle(bundle)
evil_priv, evil_pub = lic.generate_keypair()
forged = lic.resign_license(
document, master_private_key_b64=evil_priv, changes={"modules": list(lic.ALL_MODULES)}
)
forged["master_pubkey"] = evil_pub
forged["master_key_id"] = lic.key_id(evil_pub)
forged["signature"] = lic.sign(
{k: v for k, v in forged.items() if k != "signature"},
evil_priv,
lic.SIG_CONTEXT_LICENSE,
)
# In sich konsistent -- genau das war die Schwaeche von V1.
lic.verify_license(forged)
# Mit gepinntem Aussteller fliegt es auf.
_, real_pub = master_keys
with pytest.raises(lic.UntrustedIssuer):
lic.verify_license(forged, trusted_master_keys=[real_pub])
def test_key_id_mismatch_rejected(master_keys):
bundle, _ = issue(master_keys)
document, _ = lic.split_bundle(bundle)
document["master_key_id"] = "0" * 16
with pytest.raises(lic.InvalidLicenseFile):
lic.verify_license(document)
def test_license_signature_not_reusable_as_response(master_keys):
"""Domaenentrennung: Lizenz-Signatur darf nicht als Antwort-Signatur gelten."""
_, pub = master_keys
bundle, _ = issue(master_keys)
document, _ = lic.split_bundle(bundle)
payload = {k: v for k, v in document.items() if k != "signature"}
assert lic.verify(payload, document["signature"], pub, lic.SIG_CONTEXT_LICENSE) is True
assert lic.verify(payload, document["signature"], pub, lic.SIG_CONTEXT_RESPONSE) is False
assert lic.verify(payload, document["signature"], pub, lic.SIG_CONTEXT_REQUEST) is False
def test_lifetime_license(master_keys):
bundle, _ = issue(master_keys, license_type="enterprise", lifetime=True, valid_days=0)
document, _ = lic.split_bundle(bundle)
lic.verify_license(document)
status = lic.license_status(document, issuer_pinned=True)
assert status.lifetime is True
assert status.expires_at is not None
assert status.days_left is None
assert status.expired is False
assert status.modules_active is True
def test_trial_lifetime_forbidden(master_keys):
with pytest.raises(ValueError):
issue(master_keys, license_type="trial", lifetime=True, valid_days=0)
def test_license_server_role_rejects_modules(master_keys):
with pytest.raises(ValueError):
issue(master_keys, role="license_server", modules=["dhcp"])
def test_grace_period_and_expiry(master_keys):
bundle, _ = issue(master_keys, valid_days=10, modules=["dhcp"])
document, _ = lic.split_bundle(bundle)
issued = lic.parse_iso(document["issued_at"])
warn = lic.license_status(document, now=issued + timedelta(days=9), issuer_pinned=True)
assert warn.expiring_soon is True and warn.expired is False and warn.modules_active is True
grace = lic.license_status(document, now=issued + timedelta(days=25), issuer_pinned=True)
assert grace.expired is True
assert grace.grace_active is True
assert grace.modules_active is True # Kulanz laeuft noch
assert grace.days_since_expiry == 15
dead = lic.license_status(document, now=issued + timedelta(days=45), issuer_pinned=True)
assert dead.grace_active is False
assert dead.modules_active is False
def test_not_before_blocks_future_license(master_keys):
bundle, _ = issue(master_keys)
document, _ = lic.split_bundle(bundle)
issued = lic.parse_iso(document["issued_at"])
status = lic.license_status(document, now=issued - timedelta(hours=1), issuer_pinned=True)
assert status.not_yet_valid is True
assert status.modules_active is False
def test_status_without_pinning_warns(master_keys):
bundle, _ = issue(master_keys)
document, _ = lic.split_bundle(bundle)
status = lic.license_status(document)
assert any("gepinnt" in w for w in status.warnings)
def test_role_defaults_to_customer_when_missing(master_keys):
bundle, _ = issue(master_keys)
document, _ = lic.split_bundle(bundle)
document.pop("role")
assert lic.license_status(document).role == "customer"
def test_v1_bundle_compatibility(master_keys):
"""Eine alte Datei mit eingebettetem license_privkey laesst sich noch aufteilen."""
bundle, _ = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
legacy = {**document, "license_privkey": client_priv}
doc2, priv2 = lic.split_bundle(legacy)
assert priv2 == client_priv
assert "license_privkey" not in doc2
# --------------------------------------------------------------------------- #
# Aktivierungsprotokoll
# --------------------------------------------------------------------------- #
def test_activation_round_trip(master_keys):
master_priv, master_pub = master_keys
bundle, license_pub = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
request = lic.build_client_request("activate", document, client_priv, hostname="tesm-kunde")
lic.verify_client_request(request, license_pub, expected_action="activate")
response = lic.build_master_response(
"activate",
license_id=request["license_id"],
fingerprint=request["fingerprint"],
request_nonce=request["nonce"],
master_private_key=master_priv,
)
lic.verify_master_response(
response,
master_pub,
expected_license_id=document["license_id"],
expected_fingerprint=request["fingerprint"],
expected_action="activate",
expected_nonce=request["nonce"],
current_seq=document["seq"],
)
def test_offline_code_round_trip_is_identical(master_keys):
master_priv, master_pub = master_keys
bundle, license_pub = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
request = lic.build_client_request("activate", document, client_priv)
code = lic.encode_code(request)
# Ein per Mail transportierter Code darf Umbrueche/Whitespace enthalten.
wrapped = "\n".join(code[i : i + 48] for i in range(0, len(code), 48))
assert lic.decode_code(wrapped) == request
lic.verify_client_request(lic.decode_code(wrapped), license_pub)
response = lic.build_master_response(
"activate",
license_id=request["license_id"],
fingerprint=request["fingerprint"],
request_nonce=request["nonce"],
master_private_key=master_priv,
)
lic.verify_master_response(
lic.decode_code(lic.encode_code(response)),
master_pub,
expected_license_id=document["license_id"],
expected_fingerprint=request["fingerprint"],
expected_action="activate",
expected_nonce=request["nonce"],
)
def test_request_signature_requires_correct_license_key(master_keys):
bundle, _ = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
other_priv, other_pub = lic.generate_keypair()
request = lic.build_client_request("heartbeat", document, client_priv)
with pytest.raises(lic.SignatureInvalid):
lic.verify_client_request(request, other_pub)
assert other_priv # nur zur Vollstaendigkeit
def test_request_replay_is_rejected(master_keys):
bundle, license_pub = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
request = lic.build_client_request("activate", document, client_priv)
used: set[str] = set()
def seen(nonce: str) -> bool:
if nonce in used:
return True
used.add(nonce)
return False
lic.verify_client_request(request, license_pub, seen_nonce=seen)
with pytest.raises(lic.ReplayDetected):
lic.verify_client_request(request, license_pub, seen_nonce=seen)
def test_stale_request_is_rejected(master_keys):
bundle, license_pub = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
old = lic.utcnow() - timedelta(seconds=lic.REQUEST_MAX_AGE_SECONDS + 60)
request = lic.build_client_request("heartbeat", document, client_priv, now=old)
with pytest.raises(lic.ReplayDetected):
lic.verify_client_request(request, license_pub)
def test_unsigned_request_does_not_burn_nonce(master_keys):
"""Nonce-Verbrauch erst nach gueltiger Signatur -- sonst waere die Nonce-Tabelle
ein kostenloser DoS-Vektor."""
bundle, license_pub = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
request = lic.build_client_request("activate", document, client_priv)
request["signature"] = lic.sign(
{"boese": True}, lic.generate_keypair()[0], lic.SIG_CONTEXT_REQUEST
)
calls: list[str] = []
with pytest.raises(lic.SignatureInvalid):
lic.verify_client_request(request, license_pub, seen_nonce=lambda n: calls.append(n) or False)
assert calls == []
def test_response_bound_to_request_nonce(master_keys):
master_priv, master_pub = master_keys
bundle, _ = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
first = lic.build_client_request("activate", document, client_priv)
response = lic.build_master_response(
"activate",
license_id=first["license_id"],
fingerprint=first["fingerprint"],
request_nonce=first["nonce"],
master_private_key=master_priv,
)
second = lic.build_client_request("activate", document, client_priv)
with pytest.raises(lic.ReplayDetected):
lic.verify_master_response(
response,
master_pub,
expected_license_id=document["license_id"],
expected_fingerprint=second["fingerprint"],
expected_action="activate",
expected_nonce=second["nonce"],
)
def test_license_update_downgrade_rejected(master_keys):
master_priv, master_pub = master_keys
bundle, _ = issue(master_keys, modules=["dhcp"])
document, client_priv = lic.split_bundle(bundle)
updated = lic.resign_license(
document, master_private_key_b64=master_priv, changes={"modules": ["dhcp", "fileshare"]}
)
assert updated["seq"] == document["seq"] + 1
lic.verify_license(updated, trusted_master_keys=[master_pub])
request = lic.build_client_request("heartbeat", updated, client_priv)
stale = lic.build_master_response(
"heartbeat",
license_id=request["license_id"],
fingerprint=request["fingerprint"],
request_nonce=request["nonce"],
master_private_key=master_priv,
license_update=document, # alte Revision
)
with pytest.raises(lic.ReplayDetected):
lic.verify_master_response(
stale,
master_pub,
expected_license_id=updated["license_id"],
expected_fingerprint=request["fingerprint"],
expected_action="heartbeat",
expected_nonce=request["nonce"],
current_seq=updated["seq"],
)
def test_report_size_limit(master_keys):
bundle, license_pub = issue(master_keys, role="license_server")
document, client_priv = lic.split_bundle(bundle)
report = {"customers": [{"name": f"K{i}"} for i in range(lic.MAX_REPORT_ROWS + 1)]}
request = lic.build_client_request("heartbeat", document, client_priv, report=report)
with pytest.raises(lic.ProtocolViolation):
lic.verify_client_request(request, license_pub)
def test_malformed_fingerprint_and_nonce_rejected(master_keys):
bundle, license_pub = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
base = lic.build_client_request("activate", document, client_priv)
bad_fp = copy.deepcopy(base)
bad_fp["fingerprint"] = "nicht-hex"
with pytest.raises(lic.ProtocolViolation):
lic.verify_client_request(bad_fp, license_pub)
bad_nonce = copy.deepcopy(base)
bad_nonce["nonce"] = "kurz"
with pytest.raises(lic.ProtocolViolation):
lic.verify_client_request(bad_nonce, license_pub)
def test_wrong_action_for_endpoint(master_keys):
bundle, license_pub = issue(master_keys)
document, client_priv = lic.split_bundle(bundle)
request = lic.build_client_request("heartbeat", document, client_priv)
with pytest.raises(lic.ProtocolViolation):
lic.verify_client_request(request, license_pub, expected_action="activate")
def test_fingerprint_is_stable_and_hex():
first = lic.system_fingerprint()
assert first == lic.system_fingerprint()
assert len(first) == 64
int(first, 16)
tests/test_repo_hygiene.py
+326
View File
@@ -0,0 +1,326 @@
"""Prueft das Repository selbst -- die Fehlerklassen, die erst beim Ausrollen auffallen.
Der Vorgaenger hat sich hier zweimal die Finger verbrannt: CRLF-Zeilenenden aus
einer Windows-Buildmaschine machten Shell-Skripte auf dem Ziel unbrauchbar
("bad interpreter"), und Schluesseldateien landeten versehentlich im Repo.
Beides faellt hier auf, bevor ein Paket gebaut wird.
"""
from __future__ import annotations
import os
import re
import shutil
import subprocess
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).resolve().parent.parent
SKIP_DIRS = {
".git",
".venv",
"__pycache__",
"instance",
".pytest_cache",
".ruff_cache",
"node_modules",
}
TEXT_SUFFIXES = {
".py", ".sh", ".html", ".css", ".js", ".json", ".toml", ".md",
".yaml", ".yml", ".cfg", ".ini", ".txt", ".service", ".conf",
}
#: Dateien ohne Endung, die trotzdem Text sind.
EXTENSIONLESS_TEXT = {"tesm-helper", "tesm", "tesm-license"}
SECRET_NAMES = {
"data.keys",
"secret.key",
"license.json",
"license_key.json",
"master_signing_key.json",
"fernet.key",
"known_hosts",
}
SECRET_SUFFIXES = {".db", ".db-wal", ".db-shm", ".pem", ".p12", ".pfx"}
def _repo_files() -> list[Path]:
files: list[Path] = []
for path in REPO_ROOT.rglob("*"):
if not path.is_file():
continue
if any(part in SKIP_DIRS for part in path.relative_to(REPO_ROOT).parts):
continue
files.append(path)
return files
def _text_files() -> list[Path]:
return [
path
for path in _repo_files()
if path.suffix in TEXT_SUFFIXES or path.name in EXTENSIONLESS_TEXT
]
def test_no_crlf_line_endings():
"""CRLF im Paket macht Shell-Skripte auf dem Linux-Ziel unbrauchbar."""
offenders = [
str(path.relative_to(REPO_ROOT))
for path in _text_files()
if b"\r\n" in path.read_bytes()
]
assert not offenders, (
"Diese Dateien haben CRLF-Zeilenenden und wuerden auf dem Zielsystem brechen:\n "
+ "\n ".join(offenders)
)
def test_no_secrets_committed():
"""Schluessel, Datenbanken und Lizenzdateien gehoeren nie ins Repo."""
offenders = [
str(path.relative_to(REPO_ROOT))
for path in _repo_files()
if path.name in SECRET_NAMES or path.suffix in SECRET_SUFFIXES
]
assert not offenders, "Geheimnisse im Repository:\n " + "\n ".join(offenders)
def test_shell_scripts_have_shebang_and_strict_mode():
scripts = [path for path in _repo_files() if path.suffix == ".sh" or path.name == "tesm-helper"]
assert scripts, "Es wurden keine Shell-Skripte gefunden -- Test prueft ins Leere."
for script in scripts:
text = script.read_text(encoding="utf-8")
assert text.startswith("#!"), f"{script.name}: Shebang fehlt"
assert "set -euo pipefail" in text, f"{script.name}: 'set -euo pipefail' fehlt"
def test_helper_validates_every_verb():
"""Jedes Verb des privilegierten Helfers muss im case-Block auftauchen.
Der Helfer ist die einzige Stelle mit Root-Rechten. Ein Verb, das die
Anwendung kennt, der Helfer aber nicht, waere entweder tot -- oder,
schlimmer, ein Hinweis auf eine ungeprueft durchgereichte Aktion.
"""
from tesm_core.sysops import HELPER_VERBS
helper = (REPO_ROOT / "deploy" / "tesm-helper").read_text(encoding="utf-8")
# case-Muster koennen mehrere Verben mit | buendeln.
patterns: set[str] = set()
for line in helper.splitlines():
match = re.match(r"^\s{2,4}([a-z|-]+)\)\s*$", line)
if match:
patterns.update(match.group(1).split("|"))
missing = [verb for verb in HELPER_VERBS if verb not in patterns]
assert not missing, f"Verben fehlen im Helfer: {missing}"
def test_helper_has_no_unquoted_argument_passthrough():
"""Der Helfer darf Argumente nie ungeprueft an eine Shell weiterreichen."""
helper = (REPO_ROOT / "deploy" / "tesm-helper").read_text(encoding="utf-8")
for forbidden in ("eval ", "bash -c", "sh -c"):
assert forbidden not in helper, f"Der Helfer enthaelt {forbidden!r}"
def test_helper_blocks_path_traversal_everywhere():
"""Jede Pfadpruefung im Helfer muss ".." abweisen.
Ohne diese Sperre laesst sich eine Positivliste ueber den Basisnamen
unterlaufen: ``/etc/nginx/sites-available/../../../root/tesm`` hat den
zulaessigen Basisnamen ``tesm``, liest aber ``/root/tesm``.
"""
helper = (REPO_ROOT / "deploy" / "tesm-helper").read_text(encoding="utf-8")
checks = re.findall(
r"^check_\w*(?:path|target|webroot)\w*\(\) \{(.*?)^\}", helper, re.S | re.M
)
assert checks, "Es wurden keine Pfadpruefungen gefunden -- Test prueft ins Leere."
for body in checks:
assert '*".."*' in body, "Pfadpruefung ohne ..-Sperre: " + body
@pytest.mark.skipif(shutil.which("bash") is None, reason="bash nicht verfuegbar")
def test_helper_accepts_instances_but_no_invented_keys():
"""Der Namensraum einer Instanz ist eng begrenzt.
Er landet ungefiltert in Pfaden wie ``/etc/nginx/sites-available/<key>``.
"""
script = (REPO_ROOT / "deploy" / "tesm-helper").read_text(encoding="utf-8")
start = script.index("check_app_key() {")
end = script.index(chr(10) + "}", start) + 2
prelude = 'APP_KEYS_ALLOWED=("tesm" "tesm-license")' + chr(10) + "fail() { exit 1; }" + chr(10)
snippet = prelude + script[start:end]
def accepted(key: str) -> bool:
command = snippet + chr(10) + 'check_app_key "' + key + '"'
return subprocess.run(["bash", "-c", command], capture_output=True).returncode == 0
for good in ("tesm", "tesm-license", "tesm-opus", "tesm-license-opus", "tesm-test2"):
assert accepted(good), good
for bad in ("", "foo", "tesm-", "tesm-../etc", "../tesm", "tesm-GROSS", "tesm-" + "x" * 20):
assert not accepted(bad), bad
def test_systemd_units_are_hardened():
units = list((REPO_ROOT / "deploy" / "systemd").glob("*.service"))
assert units
for unit in units:
text = unit.read_text(encoding="utf-8")
assert "User=root" not in text, f"{unit.name} laeuft als root"
# ProtectSystem=full, nicht strict: strict sperrt auch den privilegierten
# Helfer aus /etc aus -- gemessen, siehe test_core_security.py.
for directive in ("ProtectSystem=full", "PrivateTmp=yes"):
assert directive in text, f"{unit.name}: {directive} fehlt"
# NoNewPrivileges waere hier kontraproduktiv: es schaltet setuid ab und
# damit sudo -- der Dienst koennte den privilegierten Helfer nicht mehr
# aufrufen, und jede Systemaktion aus der Oberflaeche waere tot. Genau
# so war es, und die Diagnoseseite meldete trotzdem "Helfer erreichbar".
assert "NoNewPrivileges=yes" not in text, (
f"{unit.name}: NoNewPrivileges=yes verhindert sudo und damit den Helfer."
)
def test_sudoers_grants_only_the_helper():
for name in ("tesm", "tesm-license"):
raw = (REPO_ROOT / "deploy" / "sudoers" / name).read_text(encoding="utf-8")
# Kommentare erklaeren, was bewusst *nicht* getan wird -- sie duerfen den
# Test nicht ausloesen.
rules = [line for line in raw.splitlines() if line.strip() and not line.startswith("#")]
body = "\n".join(rules)
assert "NOPASSWD: ALL" not in body, f"{name}: zu weite sudo-Regel"
assert "/usr/local/lib/tesm/tesm-helper" in body
for line in rules:
if "NOPASSWD" in line:
assert line.strip().endswith("/usr/local/lib/tesm/tesm-helper"), (
f"{name}: NOPASSWD-Regel erlaubt mehr als den Helfer: {line}"
)
@pytest.mark.parametrize("app", ["tesm", "tesm-license"])
def test_nginx_template_serves_static_from_app_root(app: str):
"""Der alias-Pfad ist genau der Fehler, der im Vorgaenger jahrelang ueberlebte."""
text = (REPO_ROOT / "deploy" / "nginx" / f"{app}.conf").read_text(encoding="utf-8")
assert f"alias /srv/{app}/static/;" in text
assert f"alias /srv/{app}/static-core/;" in text
def test_documentation_referenced_by_the_units_exists():
"""``Documentation=`` in der Unit muss auf eine Datei zeigen, die es gibt.
Sonst laeuft ``systemctl help tesm`` ins Leere -- und schlimmer: der
Verweis suggeriert eine Anleitung, die niemand geschrieben hat.
"""
for unit in (REPO_ROOT / "deploy" / "systemd").glob("*.service"):
for line in unit.read_text(encoding="utf-8").splitlines():
if not line.startswith("Documentation=file:"):
continue
target = line.split("file:", 1)[1].strip()
# /srv/<app>/docs/X -> docs/X im Repository
relative = target.split("/docs/", 1)[1]
assert (REPO_ROOT / "docs" / relative).is_file(), (
f"{unit.name} verweist auf {target}, docs/{relative} fehlt"
)
def test_readme_links_point_to_existing_files():
readme = (REPO_ROOT / "README.md").read_text(encoding="utf-8")
for target in re.findall(r"\]\((docs/[^)#]+)\)", readme):
assert (REPO_ROOT / target).is_file(), f"README verweist auf {target}"
def _templates() -> list[Path]:
return [
path
for path in _repo_files()
if path.suffix == ".html" and "templates" in path.parts
]
def test_no_html_entities_inside_escaped_macro_calls():
"""Makros escapen ihren Text -- Entities darin erscheinen woertlich.
Auf der Wartungsseite stand deshalb ``&bdquo;Linux-Client&ldquo;`` als
sichtbarer Text. Der Fehler ist unauffaellig, weil dieselbe Schreibweise
im uebrigen Vorlagentext voellig richtig ist.
"""
# Genau bis zum schliessenden "}}" des Aufrufs -- ein greedy Ausdruck
# verschluckte sonst den nachfolgenden Seiteninhalt und meldete dessen
# Tabellenmarkup als Treffer.
pattern = re.compile(
r"\{\{\s*(?:alert|empty_state|pill|stat)\((?:[^{}]|\{[^{}]*\})*?\)\s*"
r"(?:\|\s*safe\s*)?\}\}",
re.S,
)
offenders: list[str] = []
for path in _templates():
for match in pattern.finditer(path.read_text(encoding="utf-8")):
call = match.group(0)
# Frueher stand hier eine Ausnahme fuer "| safe". Sie war falsch:
# das Filter wirkt auf das *Ergebnis* des Makros, escaped wurde aber
# schon beim Einsetzen des Arguments. Auf der Geraeteseite stand
# deshalb woertlich "<em>und</em>" im Text.
if re.search(r"&[a-zA-Z]{2,10};|</?[a-z]{1,10}>", call):
offenders.append(
f"{path.relative_to(REPO_ROOT)}: {' '.join(call.split())[:100]}"
)
assert not offenders, (
"HTML im escapten Makrotext -- das erscheint woertlich auf der Seite:\n "
+ "\n ".join(offenders)
)
def test_refresh_regions_never_point_at_a_whole_page():
"""``data-refresh-url`` schreibt die Antwort in ``innerHTML``.
Zeigt die URL auf eine vollstaendige Seite, landet die ganze Seite in
einem Ausschnitt -- bei jedem Durchlauf eine Ebene tiefer. Genau das ist
auf der Wartungsseite passiert: die Karte "Auftraege" holte sich
``maintenance.overview``, also sich selbst.
Erlaubt sind nur Endpunkte, deren Name sie als Ausschnitt ausweist.
"""
fragment_markers = ("fragment", "tiles", "output", "status", "partial", "ausschnitt")
pattern = re.compile(r"data-refresh-url=\"\{\{\s*url_for\('([^']+)'")
found = 0
for path in _templates():
for match in pattern.finditer(path.read_text(encoding="utf-8")):
endpoint = match.group(1)
found += 1
assert any(marker in endpoint.lower() for marker in fragment_markers), (
f"{path.relative_to(REPO_ROOT)}: data-refresh-url zeigt auf {endpoint!r} -- "
"das sieht nach einer ganzen Seite aus, nicht nach einem Ausschnitt."
)
assert found, "Keine Aktualisierungsbereiche gefunden -- Test prueft ins Leere."
def test_refresh_loop_treats_zero_as_off():
"""``Number(x || 15000)`` macht aus einer 0 klammheimlich 15000."""
script = (
REPO_ROOT
/ "packages"
/ "tesm-core"
/ "src"
/ "tesm_core"
/ "static"
/ "tesm_core"
/ "js"
/ "app.js"
).read_text(encoding="utf-8")
assert "refreshInterval || " not in script
assert "interval <= 0" in script, "Ein Intervall von 0 muss die Aktualisierung abschalten."
def test_licensing_package_is_shared_not_copied():
"""Es darf keine zweite Kopie des Lizenzprotokolls geben.
Genau diese Doppelung -- zwei byte-identisch zu haltende ``licensing.py`` --
war im Vorgaenger die gefaehrlichste Invariante.
"""
copies = [
str(path.relative_to(REPO_ROOT))
for path in _repo_files()
if path.name == "licensing.py"
]
assert not copies, f"Kopien des Lizenzmoduls gefunden: {copies}"
tests/test_tesm_restarts_ssh.py
+1376
View File
File diff suppressed because it is too large Load Diff
tests/test_tesm_smoke.py
+968
View File
@@ -0,0 +1,968 @@
"""Rauchtest der TESM-Anwendung: jede Seite rendert, jede Aktion greift.
Deckt bewusst auch die Faelle ab, die im Vorgaengerprojekt kaputtgingen:
Rechteschranken, Lizenzgates auf Modulrouten, MAC-Normalisierung, und die
Trennung zwischen oeffentlicher und angemeldeter Statusuebersicht.
"""
from __future__ import annotations
from typing import Any
import pytest
@pytest.fixture()
def tesm_app(instance_env: Any):
from tesm import create_app
app = create_app(TESTING=True)
yield app
@pytest.fixture()
def tclient(tesm_app: Any):
return tesm_app.test_client()
@pytest.fixture()
def admin(tesm_app: Any):
from tesm_core.extension import core
from tesm_core.security import passwords
extension = core(tesm_app)
with extension.database.session() as conn, extension.database.transaction(conn):
conn.execute(
"INSERT INTO users (username, password_hash, is_admin, auth_source, created_at, "
"updated_at) VALUES ('root',?,1,'local',datetime('now'),datetime('now'))",
(passwords.hash_password("Ein-gutes-Passwort-1"),),
)
return {"username": "root", "password": "Ein-gutes-Passwort-1"}
def _token(client: Any, path: str = "/login") -> str:
body = client.get(path).get_data(as_text=True)
for marker in ('name="csrf_token" value="', 'data-csrf="'):
if marker in body:
start = body.index(marker) + len(marker)
return body[start : body.index('"', start)]
raise AssertionError(f"Kein CSRF-Token in {path}")
@pytest.fixture()
def login_admin(tclient: Any, admin: dict[str, str]):
def do() -> Any:
return tclient.post(
"/login",
data={
"username": admin["username"],
"password": admin["password"],
"csrf_token": _token(tclient),
},
follow_redirects=True,
)
return do
# --------------------------------------------------------------------------- #
# Start und Grundzustand
# --------------------------------------------------------------------------- #
def test_app_boots_and_migrations_applied(tesm_app):
from tesm_core.extension import core
status = core(tesm_app).migration_status()
components = {row["component"] for row in status}
assert components == {"core", "tesm"}
names = {(row["component"], row["name"]) for row in status}
assert ("tesm", "baseline") in names
assert ("tesm", "seed-dhcp-options") in names
def test_dhcp_option_catalogue_seeded(tesm_app):
from tesm_core.db import Database
from tesm_core.extension import core
extension = core(tesm_app)
with extension.database.session() as conn:
count = Database.value(conn, "SELECT COUNT(*) FROM dhcp_options WHERE is_standard=1")
host_name = Database.value(conn, "SELECT COUNT(*) FROM dhcp_options WHERE code=12")
assert count == 56
# host-name (12) ist bewusst nicht im Katalog -- Kea beantwortet das ueber
# das hostname-Feld der Reservierung.
assert host_name == 0
def test_role_presets_created(tesm_app):
from tesm_core.extension import core
from tesm_core.rbac import service as rbac
extension = core(tesm_app)
with extension.database.session() as conn:
names = {group.name for group in rbac.list_groups(conn)}
assert {"Benutzer", "Betrieb", "Support", "Netzwerk", "Revision", "Administration"} <= names
def test_public_dashboard_hides_details(tclient):
response = tclient.get("/")
assert response.status_code == 200
body = response.get_data(as_text=True)
assert "Statusuebersicht" in body
assert "oeffentliche Ansicht" in body
def test_health_endpoint(tclient):
assert tclient.get("/gesundheit").get_json()["status"] == "ok"
# --------------------------------------------------------------------------- #
# Seiten
# --------------------------------------------------------------------------- #
ADMIN_PAGES = [
"/",
"/clients/",
"/switche/",
"/zugangsdaten/",
"/protokolle/laufend",
"/protokolle/verlauf",
"/protokolle/aenderungen",
"/protokolle/neustarts",
"/einstellungen/system",
"/einstellungen/netzwerk",
"/einstellungen/webserver",
"/einstellungen/verzeichnisdienst",
"/lizenz/",
"/sicherung/",
"/papierkorb/",
"/diagnose/",
"/verwaltung/benutzer",
"/verwaltung/gruppen",
"/verwaltung/sitzungen",
"/konto/",
"/konto/sicherheit",
]
@pytest.mark.parametrize("path", ADMIN_PAGES)
def test_admin_pages_render(tclient, login_admin, path):
login_admin()
response = tclient.get(path)
assert response.status_code == 200, f"{path} -> {response.status_code}"
assert "Interner Fehler" not in response.get_data(as_text=True)
def test_licensed_modules_are_hidden_without_license(tclient, login_admin):
"""Ohne Lizenz sind DHCP, Wartung und Dateifreigaben komplett unerreichbar."""
login_admin()
for path in ("/dhcp/", "/wartung/", "/dateifreigaben/", "/protokolle/kea"):
assert tclient.get(path).status_code == 404, path
body = tclient.get("/").get_data(as_text=True)
assert "DHCP-Server" not in body
assert "Wartung" not in body
# --------------------------------------------------------------------------- #
# Geraete
# --------------------------------------------------------------------------- #
def test_device_lifecycle(tclient, login_admin):
login_admin()
token = _token(tclient, "/clients/")
created = tclient.post(
"/clients/neu",
data={
"csrf_token": token,
"name": "Kamera Eingang",
"mac": "AA-BB-CC-DD-EE-FF",
"ip": "192.168.1.50",
"is_active": "1",
"auto_restart": "1",
"ssh_port": "22",
},
follow_redirects=True,
)
assert created.status_code == 200
body = created.get_data(as_text=True)
assert "angelegt" in body
# MAC wird normalisiert.
assert "aa:bb:cc:dd:ee:ff" in tclient.get("/clients/").get_data(as_text=True)
listing = tclient.get("/clients/").get_data(as_text=True)
assert "Kamera Eingang" in listing
# Loeschen legt in den Papierkorb, MAC bleibt reserviert.
tclient.post("/clients/1/loeschen", data={"csrf_token": token}, follow_redirects=True)
trash = tclient.get("/papierkorb/").get_data(as_text=True)
assert "Kamera Eingang" in trash
again = tclient.post(
"/clients/neu",
data={
"csrf_token": token,
"name": "Zweite Kamera",
"mac": "aa:bb:cc:dd:ee:ff",
"ip": "192.168.1.51",
"ssh_port": "22",
},
follow_redirects=True,
)
assert "Papierkorb" in again.get_data(as_text=True)
def test_device_requires_port_when_switch_assigned(tclient, login_admin):
login_admin()
token = _token(tclient, "/switche/")
tclient.post(
"/switche/neu",
data={"csrf_token": token, "hostname": "sw01", "ip": "192.168.1.2", "ssh_port": "22"},
follow_redirects=True,
)
response = tclient.post(
"/clients/neu",
data={
"csrf_token": token,
"name": "Ohne Port",
"mac": "11:22:33:44:55:66",
"ip": "192.168.1.60",
"switch_id": "1",
"ssh_port": "22",
},
follow_redirects=True,
)
assert "Portbezeichnung" in response.get_data(as_text=True)
def test_invalid_mac_rejected(tclient, login_admin):
login_admin()
token = _token(tclient, "/clients/")
response = tclient.post(
"/clients/neu",
data={"csrf_token": token, "name": "Kaputt", "mac": "xyz", "ip": "192.168.1.70"},
follow_redirects=True,
)
assert "hexadezimale" in response.get_data(as_text=True)
def test_invalid_ip_rejected(tclient, login_admin):
login_admin()
token = _token(tclient, "/clients/")
response = tclient.post(
"/clients/neu",
data={"csrf_token": token, "name": "Kaputt", "mac": "11:22:33:44:55:77", "ip": "999.1.1.1"},
follow_redirects=True,
)
assert "gueltige IP-Adresse" in response.get_data(as_text=True)
# --------------------------------------------------------------------------- #
# Zugangsdaten und Verschluesselung
# --------------------------------------------------------------------------- #
def test_credential_secret_is_encrypted_at_rest(tclient, tesm_app, login_admin):
from tesm_core.db import Database
from tesm_core.extension import core
login_admin()
token = _token(tclient, "/zugangsdaten/")
tclient.post(
"/zugangsdaten/neu",
data={
"csrf_token": token,
"name": "Switch-Admin",
"username": "admin",
"secret": "SuperGeheim-2026",
"category": "switch",
},
follow_redirects=True,
)
extension = core(tesm_app)
with extension.database.session() as conn:
stored = Database.value(conn, "SELECT secret_encrypted FROM credentials WHERE name=?", ("Switch-Admin",))
assert stored
assert "SuperGeheim" not in stored
assert stored.startswith("v1.")
def test_credential_reveal_requires_permission(tclient, tesm_app, login_admin):
from tesm_core.extension import core
from tesm_core.security import passwords
login_admin()
token = _token(tclient, "/zugangsdaten/")
tclient.post(
"/zugangsdaten/neu",
data={
"csrf_token": token,
"name": "Nur-Lesen",
"username": "admin",
"secret": "Geheim-Passwort-9",
"category": "switch",
},
follow_redirects=True,
)
tclient.get("/logout")
# Ein Konto mit credentials.view aber ohne credentials.secrets.
extension = core(tesm_app)
with extension.database.session() as conn, extension.database.transaction(conn):
cursor = conn.execute(
"INSERT INTO users (username, password_hash, auth_source, created_at, updated_at) "
"VALUES ('lisa',?,'local',datetime('now'),datetime('now'))",
(passwords.hash_password("Ein-gutes-Passwort-1"),),
)
user_id = int(cursor.lastrowid or 0)
group = conn.execute(
"INSERT INTO groups (name, created_at, updated_at) "
"VALUES ('nur-lesen',datetime('now'),datetime('now'))"
)
group_id = int(group.lastrowid or 0)
conn.executemany(
"INSERT INTO group_permissions (group_id, permission) VALUES (?,?)",
[(group_id, "monitoring.view"), (group_id, "credentials.view")],
)
conn.execute("INSERT INTO user_groups (user_id, group_id) VALUES (?,?)", (user_id, group_id))
client = tclient
client.post(
"/login",
data={"username": "lisa", "password": "Ein-gutes-Passwort-1", "csrf_token": _token(client)},
follow_redirects=True,
)
assert client.get("/zugangsdaten/").status_code == 200
response = client.post("/zugangsdaten/1/anzeigen", data={"csrf_token": _token(client, "/zugangsdaten/")})
assert response.status_code == 403
def test_credential_reveal_is_audited(tclient, tesm_app, login_admin):
from tesm_core import audit
from tesm_core.extension import core
login_admin()
token = _token(tclient, "/zugangsdaten/")
tclient.post(
"/zugangsdaten/neu",
data={
"csrf_token": token,
"name": "Geheim",
"username": "admin",
"secret": "Geheim-Passwort-9",
"category": "switch",
},
follow_redirects=True,
)
response = tclient.post(
"/zugangsdaten/1/anzeigen", data={"csrf_token": token}, follow_redirects=True
)
assert response.status_code == 200
assert "Geheim-Passwort-9" in response.get_data(as_text=True)
extension = core(tesm_app)
with extension.database.session() as conn:
rows = audit.search(conn, action="credential.secret_revealed")
assert rows and rows[0]["severity"] == "warning"
# --------------------------------------------------------------------------- #
# Switche und Host-Schluessel
# --------------------------------------------------------------------------- #
def test_switch_created_without_trusted_host_key(tclient, login_admin):
login_admin()
token = _token(tclient, "/switche/")
response = tclient.post(
"/switche/neu",
data={"csrf_token": token, "hostname": "sw-kern", "ip": "10.0.0.2", "ssh_port": "22"},
follow_redirects=True,
)
assert "freigegeben werden" in response.get_data(as_text=True)
detail = tclient.get("/switche/1").get_data(as_text=True)
assert "Noch kein Schluessel hinterlegt" in detail
def test_switch_delete_blocked_while_devices_attached(tclient, login_admin):
login_admin()
token = _token(tclient, "/switche/")
tclient.post(
"/switche/neu",
data={"csrf_token": token, "hostname": "sw01", "ip": "10.0.0.3", "ssh_port": "22"},
follow_redirects=True,
)
tclient.post(
"/clients/neu",
data={
"csrf_token": token,
"name": "Kamera",
"mac": "aa:11:22:33:44:55",
"ip": "10.0.0.50",
"switch_id": "1",
"port": "1",
"ssh_port": "22",
},
follow_redirects=True,
)
response = tclient.post("/switche/1/loeschen", data={"csrf_token": token}, follow_redirects=True)
assert "zugeordnet" in response.get_data(as_text=True)
def test_command_not_on_allowlist_rejected(tclient, login_admin):
login_admin()
token = _token(tclient, "/switche/")
tclient.post(
"/switche/neu",
data={"csrf_token": token, "hostname": "sw01", "ip": "10.0.0.4", "ssh_port": "22"},
follow_redirects=True,
)
response = tclient.post(
"/switche/1/kommando",
data={"csrf_token": token, "command": "erase startup-config"},
follow_redirects=True,
)
assert "Positivliste" in response.get_data(as_text=True)
# --------------------------------------------------------------------------- #
# Rechte
# --------------------------------------------------------------------------- #
def test_viewer_cannot_reach_admin_pages(tclient, tesm_app):
from tesm_core.extension import core
from tesm_core.rbac import service as rbac
from tesm_core.security import passwords
extension = core(tesm_app)
with extension.database.session() as conn, extension.database.transaction(conn):
cursor = conn.execute(
"INSERT INTO users (username, password_hash, auth_source, created_at, updated_at) "
"VALUES ('gast',?,'local',datetime('now'),datetime('now'))",
(passwords.hash_password("Ein-gutes-Passwort-1"),),
)
user_id = int(cursor.lastrowid or 0)
default = next(group for group in rbac.list_groups(conn) if group.is_default)
conn.execute(
"INSERT INTO user_groups (user_id, group_id) VALUES (?,?)", (user_id, default.id)
)
tclient.post(
"/login",
data={"username": "gast", "password": "Ein-gutes-Passwort-1", "csrf_token": _token(tclient)},
follow_redirects=True,
)
assert tclient.get("/clients/").status_code == 200
for path in ("/zugangsdaten/", "/verwaltung/benutzer", "/einstellungen/system", "/diagnose/"):
assert tclient.get(path).status_code == 403, path
def test_permission_matrix_columns_per_area(tesm_app):
from tesm.permissions import PERMISSIONS
from tesm_core.rbac.model import Action
monitoring = PERMISSIONS.area("monitoring")
assert Action.SECRETS in monitoring.columns() # Zugangsdaten
assert Action.DELETE in monitoring.columns()
network = PERMISSIONS.area("network")
assert Action.CREATE not in network.columns() # DHCP kennt kein "Anlegen"
assert Action.SECRETS not in network.columns()
def test_delete_is_a_separate_right(tesm_app):
"""Kernaenderung gegenueber dem Vorgaenger: Aendern deckt nicht mehr Loeschen ab."""
from tesm.permissions import PERMISSIONS
granted = {"monitoring.view", "devices.view", "devices.edit"}
effective = PERMISSIONS.effective(granted)
assert "devices.edit" in effective
assert "devices.delete" not in effective
# --------------------------------------------------------------------------- #
# Offline-Aktivierung: der Code muss sichtbar sein
# --------------------------------------------------------------------------- #
def test_offline_request_keeps_its_code():
"""Die Seite forderte zum Uebermitteln eines Codes auf -- und zeigte keinen.
``handshake`` erzeugte ihn, der Blueprint verwarf ihn, und nach der
Weiterleitung war er weg. Er gehoert deshalb zur gespeicherten Anfrage:
zwischen Anfrage und Antwort liegen in der Praxis eine E-Mail und ein
Arbeitstag.
"""
import inspect
from tesm_core import licensing_client
source = inspect.getsource(licensing_client.LicenseManager.handshake)
assert '"code": code,' in source
assert "code = lic.encode_code(request_payload)" in source
def test_the_reason_for_going_offline_is_named():
""""War nicht erreichbar" ist bei einem Zertifikatsfehler schlicht falsch."""
from tesm_core.licensing_client import describe_transport_error
certificate = describe_transport_error(
Exception("[SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: self-signed certificate"),
"https://10.0.0.1:8443",
)
assert "selbstsigniertes Zertifikat" in certificate
assert "https://10.0.0.1:8443" in certificate
assert "aufloesen" in describe_transport_error(
Exception("Name or service not known"), "https://nope.invalid"
)
assert "nimmt keine Verbindung an" in describe_transport_error(
Exception("Connection refused"), "https://10.0.0.1:8443"
)
assert "rechtzeitig" in describe_transport_error(
Exception("operation timed out"), "https://10.0.0.1:8443"
)
def test_license_page_shows_the_request_code():
from pathlib import Path
template = (
Path(__file__).resolve().parent.parent
/ "apps" / "tesm" / "src" / "tesm" / "templates" / "tesm" / "license.html"
).read_text(encoding="utf-8")
assert "pending.code" in template
assert "Anfragecode" in template
assert 'data-copy="#offline-request"' in template
# --------------------------------------------------------------------------- #
# Dateifreigaben
# --------------------------------------------------------------------------- #
def _license_all_modules(tesm_app):
"""Alle Module freischalten -- die Freigabeseiten haengen an der Lizenz."""
from tesm_core.extension import core
manager = core(tesm_app).license
if manager is not None:
manager.module_active = lambda _name: True # type: ignore[method-assign]
def test_share_setup_page_is_reachable_and_searches_groups(tesm_app, tclient, login_admin):
"""Die Einrichtung war nur ueber die Adresszeile erreichbar.
Der Menuepunkt haengte an vorhandenen Einbindungen -- die es ohne
Zuordnung nie gibt. Wer Zuordnungen pflegen darf, muss die Seite auch
finden, und der Gruppen-DN muss suchbar sein statt abtippbar.
"""
_license_all_modules(tesm_app)
login_admin()
page = tclient.get("/dateifreigaben/zuordnungen")
assert page.status_code == 200
body = page.get_data(as_text=True)
assert "data-group-query" in body, "Gruppensuche fehlt"
assert 'name="ad_group_dn"' in body
nav = tclient.get("/").get_data(as_text=True)
assert "/dateifreigaben/zuordnungen" in nav, "Zuordnungen fehlen in der Navigation"
def test_share_mapping_round_trip(tesm_app, tclient, login_admin):
_license_all_modules(tesm_app)
login_admin()
token = _token(tclient, "/dateifreigaben/zuordnungen")
response = tclient.post(
"/dateifreigaben/zuordnungen",
data={
"label": "Technik",
"unc": r"\\fileserver\technik",
"ad_group_name": "GG_Technik",
"ad_group_dn": "CN=GG_Technik,OU=Gruppen,DC=firma,DC=local",
"csrf_token": token,
},
follow_redirects=True,
)
assert response.status_code == 200
assert "Zuordnung gespeichert" in response.get_data(as_text=True)
from tesm.services import fileshare
from tesm_core.extension import core
with core(tesm_app).database.session() as conn:
assert len(fileshare.list_mappings(conn)) == 1
# Der Anspruch entsteht ueber die Verzeichnisgruppe -- und nur darueber.
assert fileshare.shares_for_groups(
conn, ["cn=gg_technik,ou=gruppen,dc=firma,dc=local"]
)
assert not fileshare.shares_for_groups(conn, [])
def test_login_triggers_share_mounting(tesm_app, tclient, login_admin, monkeypatch):
"""Der Ausloeser fehlte: Zuordnungen bestanden, eingebunden wurde nie.
Ohne diesen Haken blieb der gesamte Freigabe-Teil wirkungslos -- und weil
der Menuepunkt an Einbindungen haengt, war er auch unsichtbar.
"""
from tesm_core.extension import core
seen: list[dict] = []
def fake_mount(conn, **kwargs):
seen.append(kwargs)
return []
_license_all_modules(tesm_app)
monkeypatch.setattr("tesm.services.fileshare.mount_for_login", fake_mount)
assert core(tesm_app).login_hooks, "kein Anmeldehaken registriert"
login_admin()
assert seen, "die Anmeldung hat die Einbindung nicht ausgeloest"
# Das Passwort wird durchgereicht, aber nirgends abgelegt.
assert seen[0]["password"]
with tclient.session_transaction() as session:
assert not any("passw" in str(key).lower() for key in session)
def test_logout_releases_shares(tesm_app, tclient, login_admin, monkeypatch):
"""Was die Anmeldung eingebunden hat, muss die Abmeldung wieder loesen.
Sonst bliebe der Zugriff mit dem Konto des Benutzers offen, bis die
Zeitgrenze der Einbindung greift -- Stunden spaeter.
"""
from tesm_core.extension import core
_license_all_modules(tesm_app)
released: list[int] = []
monkeypatch.setattr(
"tesm.services.fileshare.mount_for_login", lambda conn, **kw: []
)
monkeypatch.setattr(
"tesm.services.fileshare.release_all_for_user",
lambda conn, user_id: released.append(user_id) or 1,
)
assert core(tesm_app).logout_hooks, "kein Abmeldehaken registriert"
login_admin()
token = _token(tclient, "/")
tclient.post("/logout", data={"csrf_token": token})
assert released, "die Abmeldung hat die Freigaben nicht geloest"
@pytest.mark.parametrize(
"eingabe",
[
r"\\dc2022\daten",
"//dc2022/daten",
r"\\dc2022/daten",
"//dc2022/daten/",
"\\\\dc2022\\daten\\", # Trenner am Ende
"//dc2022//daten",
" //dc2022/daten ",
],
)
def test_unc_accepts_both_separators(eingabe):
"""Backslash und Schraegstrich muessen beide gehen -- und zwar gleich.
Wer den Pfad aus dem Explorer kopiert, bekommt Backslashes; wer ihn aus
einer Weboberflaeche oder von einem Linux-Kollegen bekommt, Schraegstriche.
Ueberzaehlige Trenner und einer am Ende gehoeren zum Alltag und duerfen
keine Ablehnung ausloesen.
"""
from tesm.services.fileshare import validate_unc
assert validate_unc(eingabe) == "\\\\dc2022\\daten"
@pytest.mark.parametrize(
"eingabe",
[
"/dc2022/daten", # ein Trenner: absoluter Pfad auf diesem Rechner
"//dc2022", # ohne Freigabenamen
"dc2022/daten",
"//dc 2022/daten",
],
)
def test_unc_rejects_what_is_not_a_share(eingabe):
from tesm.services.fileshare import FileshareError, validate_unc
with pytest.raises(FileshareError):
validate_unc(eingabe)
# --------------------------------------------------------------------------- #
# DHCP: Reservierungen und Optionen je Geraet
# --------------------------------------------------------------------------- #
def _netz(cidr: str = "192.168.80.0/24", interface: str = "eth0"):
from tesm.services.kea import InterfaceNetwork
return InterfaceNetwork(interface=interface, cidr=cidr)
def _geraet(conn, name: str, mac: str, ip: str) -> int:
cur = conn.execute(
"INSERT INTO devices (name, mac, ip, is_active, created_at, updated_at) "
"VALUES (?,?,?,1,datetime('now'),datetime('now'))",
(name, mac, ip),
)
return int(cur.lastrowid)
def test_reservations_only_for_existing_ranges(tesm_app):
"""Eine Reservierung ausserhalb aller Bereiche liefert Kea nie aus.
Sie trotzdem zu zeigen, hat Reservierungen versprochen, die es nicht gab --
in der Liste standen Adressen, die in der erzeugten Konfiguration fehlten.
"""
from tesm.services import kea
from tesm_core.extension import core
ext = core(tesm_app)
with ext.database.session() as conn, ext.database.transaction(conn):
_geraet(conn, "drinnen", "aa:bb:cc:dd:ee:01", "192.168.80.50")
_geraet(conn, "draussen", "aa:bb:cc:dd:ee:02", "10.99.0.50")
conn.execute(
"INSERT INTO dhcp_subnets (interface, range_start, range_end, enabled, "
"created_at, updated_at) VALUES ('eth0','192.168.80.100','192.168.80.200',1,"
"datetime('now'),datetime('now'))"
)
with ext.database.session() as conn:
netze = kea.configured_networks(conn, [_netz()])
assert [str(n) for n in netze] == ["192.168.80.0/24"]
uebersprungen: list[dict[str, str]] = []
eintraege = kea.collect_reservations(
conn, only_networks=netze, skipped=uebersprungen
)
namen = {e["name"] for e in eintraege}
assert namen == {"drinnen"}, namen
assert [e["ip"] for e in uebersprungen] == ["10.99.0.50"]
# Ohne Eingrenzung bleibt das alte Verhalten -- der Erzeuger filtert selbst.
assert len(kea.collect_reservations(conn)) == 2
def test_device_option_lands_in_its_reservation(tesm_app):
"""Ein Wert je Geraet gehoert in dessen Reservierung, nicht in den globalen Teil."""
from tesm.services import kea
from tesm_core.extension import core
ext = core(tesm_app)
with ext.database.session() as conn, ext.database.transaction(conn):
device_id = _geraet(conn, "autodarts", "4c:52:62:25:b9:e4", "192.168.80.137")
conn.execute(
"INSERT INTO dhcp_subnets (interface, range_start, range_end, enabled, "
"created_at, updated_at) VALUES ('eth0','192.168.80.100','192.168.80.200',1,"
"datetime('now'),datetime('now'))"
)
option_id = conn.execute(
"INSERT INTO dhcp_options (code, name, type, is_standard) VALUES (225,'url','string',0)"
).lastrowid
conn.execute(
"INSERT INTO dhcp_option_values (option_id, device_id, value) VALUES (?,?,?)",
(option_id, device_id, "http://autodarts.local"),
)
conn.execute(
"INSERT INTO dhcp_option_values (option_id, device_id, value) VALUES (?,NULL,?)",
(option_id, "http://global.local"),
)
with ext.database.session() as conn:
config, _ = kea.render_config(conn, networks=[_netz()], settings={})
subnet = config["Dhcp4"]["subnet4"][0]
reservierung = subnet["reservations"][0]
assert reservierung["ip-address"] == "192.168.80.137"
assert reservierung["option-data"] == [{"name": "url", "data": "http://autodarts.local"}]
# Der globale Wert bleibt global und wird nicht verdraengt.
assert {"name": "url", "data": "http://global.local"} in config["Dhcp4"]["option-data"]
def test_reservation_state_from_leases():
"""Wie im Windows-DHCP: eine Reservierung ist aktiv, wenn sie benutzt wird.
Ohne diese Angabe sah eine nie abgeholte Reservierung genauso aus wie eine
benutzte -- und ein Geraet, das noch auf einer alten Adresse haengt und die
Reservierung deshalb gar nicht uebernimmt, fiel ueberhaupt nicht auf.
"""
import time
from tesm.services.kea import Lease, annotate_reservations
jetzt = int(time.time())
benutzt = Lease("192.168.80.10", "aa:bb:cc:00:00:01", "a", jetzt + 3600, "0")
veraltet = Lease("192.168.80.99", "aa:bb:cc:00:00:03", "c", jetzt + 3600, "0")
abgelaufen = Lease("192.168.80.12", "aa:bb:cc:00:00:04", "d", jetzt - 60, "0")
eintraege = annotate_reservations(
[
{"mac": "aa:bb:cc:00:00:01", "ip": "192.168.80.10"},
{"mac": "aa:bb:cc:00:00:02", "ip": "192.168.80.11"},
{"mac": "aa:bb:cc:00:00:03", "ip": "192.168.80.30"},
{"mac": "aa:bb:cc:00:00:04", "ip": "192.168.80.12"},
],
[benutzt, veraltet, abgelaufen],
)
assert [e["state"] for e in eintraege] == [
"aktiv",
"unbenutzt",
"andere_adresse",
"abgelaufen",
]
# Der Hinweis erklaert den Zustand, statt ihn nur zu benennen.
assert all(e["state_hint"] for e in eintraege)
assert eintraege[2]["lease"].address == "192.168.80.99"
def test_lease_expiry_is_readable():
"""Der rohe Unix-Zeitstempel stand vorher unveraendert in der Tabelle."""
import time
from datetime import datetime
from tesm.services.kea import Lease
jetzt = int(time.time())
lease = Lease("192.168.80.10", "aa:bb:cc:00:00:01", "a", jetzt + 5400, "0")
erwartet = datetime.fromtimestamp(jetzt + 5400).strftime("%d.%m.%Y %H:%M")
assert lease.expires_text == erwartet
assert lease.remaining_text.startswith("in 1 h")
assert str(jetzt) not in lease.expires_text
assert Lease("x", "aa:bb:cc:00:00:01", "a", 0, "0").expires_text == "unbegrenzt"
assert Lease("x", "aa:bb:cc:00:00:01", "a", jetzt - 5, "0").remaining_text == "abgelaufen"
def test_custom_option_used_only_in_reservation_is_declared(tesm_app):
"""Eine eigene Option muss deklariert sein, auch ohne globalen Wert.
Genau das hat den Dienst umgebracht: der Wert stand nur in einer
Reservierung, die ``option-def`` wurde aber nur fuer Optionen mit globalem
Wert erzeugt. Kea startet dann nicht mehr --
"definition for the option 'dhcp4.url' does not exist" -- und weil das erst
beim naechsten Start auffiel, lag der DHCP-Server stundenlang.
"""
from tesm.services import kea
from tesm_core.extension import core
ext = core(tesm_app)
with ext.database.session() as conn, ext.database.transaction(conn):
device_id = _geraet(conn, "autodarts", "4c:52:62:25:b9:e4", "192.168.80.137")
conn.execute(
"INSERT INTO dhcp_subnets (interface, range_start, range_end, enabled, "
"created_at, updated_at) VALUES ('eth0','192.168.80.100','192.168.80.200',1,"
"datetime('now'),datetime('now'))"
)
option_id = conn.execute(
"INSERT INTO dhcp_options (code, name, type, is_standard) "
"VALUES (225,'url','string',0)"
).lastrowid
# Bewusst *kein* globaler Wert -- nur der Wert fuer dieses eine Geraet.
conn.execute(
"INSERT INTO dhcp_option_values (option_id, device_id, value) VALUES (?,?,?)",
(option_id, device_id, "https://play.autodarts.com"),
)
with ext.database.session() as conn:
config, _ = kea.render_config(conn, networks=[_netz()], settings={})
dhcp4 = config["Dhcp4"]
benutzt = dhcp4["subnet4"][0]["reservations"][0]["option-data"]
assert benutzt == [{"name": "url", "data": "https://play.autodarts.com"}]
assert "option-data" not in dhcp4, "kein globaler Wert erwartet"
definitionen = {entry["name"]: entry for entry in dhcp4["option-def"]}
assert "url" in definitionen, "ohne option-def verweigert Kea den Start"
assert definitionen["url"] == {
"name": "url",
"code": 225,
"type": "string",
"space": "dhcp4",
}
def test_restart_marks_device_as_restarting(tesm_app):
"""Nach einem Neustart muss die Kachel das sofort zeigen.
Der Zustand wechselte vorher erst beim naechsten Pruefdurchlauf -- Vorgabe
fuenf Minuten. Ein Neustart ueber SSH ist in vierzig Sekunden durch: das
Geraet war weg und wieder da, ohne dass es je jemand gesehen haette. Die
Kachel blieb die ganze Zeit auf "online".
"""
from tesm.services import monitor
from tesm_core.extension import core
ext = core(tesm_app)
with ext.database.session() as conn, ext.database.transaction(conn):
device_id = _geraet(conn, "kasse", "aa:bb:cc:00:00:07", "192.168.80.7")
conn.execute(
"INSERT INTO device_status (device_id, state, last_change_at) "
"VALUES (?, 'online', datetime('now'))",
(device_id,),
)
monitor.record_restart(
conn,
device_id=device_id,
action="reboot",
result="ok",
trigger="manuell",
actor="test",
duration_ms=100,
detail="",
method=monitor.METHOD_SSH,
)
with ext.database.session() as conn:
zeile = conn.execute(
"SELECT state, restart_count FROM device_status WHERE device_id=?", (device_id,)
).fetchone()
assert zeile["state"] == monitor.STATE_RESTARTING
assert zeile["restart_count"] == 1
assert monitor.dashboard_counts(conn)["restarting"] == 1
# Solange es schweigt, bleibt es "startet neu" -- nicht "ausgefallen".
vorher, nachher = monitor.record_status(
conn, device_id, monitor.PingResult(ok=False, latency_ms=None, detail="")
)
assert nachher == monitor.STATE_RESTARTING, "ein laufender Neustart ist kein Ausfall"
# Antwortet es wieder, ist es online.
_, nachher = monitor.record_status(
conn, device_id, monitor.PingResult(ok=True, latency_ms=3, detail="")
)
assert nachher == "online"
def test_restart_state_gives_up_after_the_grace_period(tesm_app):
"""Ein Neustart, der fuenf Minuten dauert, ist keiner mehr."""
from datetime import datetime, timedelta, timezone
from tesm.services import monitor
from tesm_core.extension import core
ext = core(tesm_app)
lange_her = (
datetime.now(timezone.utc) - timedelta(seconds=monitor.RESTART_GRACE_SECONDS + 60)
).isoformat()
with ext.database.session() as conn, ext.database.transaction(conn):
device_id = _geraet(conn, "haenger", "aa:bb:cc:00:00:08", "192.168.80.8")
conn.execute(
"INSERT INTO device_status (device_id, state, last_restart_at, last_change_at) "
"VALUES (?, ?, ?, datetime('now'))",
(device_id, monitor.STATE_RESTARTING, lange_her),
)
with ext.database.session() as conn:
_, nachher = monitor.record_status(
conn, device_id, monitor.PingResult(ok=False, latency_ms=None, detail="")
)
assert nachher == "offline"
tests/test_webserver.py
+521
View File
@@ -0,0 +1,521 @@
"""Die nginx-Vorlage und die Trennung mehrerer Instanzen auf einem Host.
Zwei Fehlerklassen werden hier festgenagelt:
1. Der alias-Fehler des Vorgaengers -- statische Dateien zeigten auf ein
Verzeichnis, das es nicht gab, und die Oberflaeche kam jahrelang ohne
Design. Die erzeugte Datei muss die richtigen Pfade enthalten.
2. Zwei Installationen derselben Anwendung auf einem Host. Solange alles
ueber den ``app_key`` benannt wurde, hat eine Testinstanz beim Speichern
der Webserver-Einstellungen die Site-Datei der Produktion ueberschrieben
-- und beide teilten sich dasselbe Sitzungscookie, weil Cookies keine
Ports unterscheiden.
"""
from __future__ import annotations
import re
import pytest
from tesm_core.system import webserver
from tesm_core.sysops import HELPER_VERBS, Result, SysOps
def make_config(**overrides) -> webserver.WebConfig:
base = {
"app_key": "tesm",
"upstream": "127.0.0.1:5000",
"static_root": "/srv/tesm/static",
}
return webserver.WebConfig(**{**base, **overrides})
# --------------------------------------------------------------------------- #
# Nur HTTP
# --------------------------------------------------------------------------- #
def test_http_only_has_no_tls_block():
text = webserver.render(make_config())
assert "listen 80 default_server;" in text
assert "ssl_certificate" not in text
assert "return 308" not in text
def test_static_aliases_point_into_the_installation():
text = webserver.render(make_config())
assert "alias /srv/tesm/static/;" in text
assert "alias /srv/tesm/static-core/;" in text
def test_proxy_passes_the_host_including_port():
"""``$host`` verwirft den Port -- damit scheitert die Origin-Pruefung.
Genau daran ist die Anmeldung hinter dem Reverse Proxy auf einem
abweichenden Port real gescheitert.
"""
text = webserver.render(make_config())
assert "proxy_set_header Host $http_host;" in text
assert "proxy_set_header Host $host;" not in text
assert "proxy_set_header X-Forwarded-Port $server_port;" in text
def test_acme_path_is_always_reachable():
text = webserver.render(
make_config(https_enabled=True, cert_path="/c.pem", key_path="/k.pem")
)
# Auch bei aktiver Umleitung -- sonst kann Let's Encrypt nie erneuern.
assert "/.well-known/acme-challenge/" in text
challenge = text.index("acme-challenge")
redirect = text.index("return 308")
assert challenge < redirect
# --------------------------------------------------------------------------- #
# HTTPS
# --------------------------------------------------------------------------- #
def test_https_listens_on_443_and_redirects_80():
text = webserver.render(
make_config(https_enabled=True, cert_path="/c.pem", key_path="/k.pem")
)
assert "listen 443 ssl default_server http2;" in text
assert "listen [::]:443 ssl default_server http2;" in text
assert "return 308 https://$host$request_uri;" in text
assert "ssl_protocols TLSv1.2 TLSv1.3;" in text
def test_hsts_only_when_asked():
without = webserver.render(
make_config(https_enabled=True, cert_path="/c.pem", key_path="/k.pem")
)
assert "Strict-Transport-Security" not in without
with_hsts = webserver.render(
make_config(
https_enabled=True, hsts_enabled=True, cert_path="/c.pem", key_path="/k.pem"
)
)
assert "Strict-Transport-Security" in with_hsts
# HSTS ohne TLS waere ein Selbstschuss: der Browser merkt sich, dass er die
# Seite nur noch ueber HTTPS aufrufen darf.
assert "ssl_certificate" in with_hsts
def test_hsts_is_ignored_without_https():
text = webserver.render(make_config(hsts_enabled=True))
assert "Strict-Transport-Security" not in text
def test_redirect_can_be_switched_off():
text = webserver.render(
make_config(
https_enabled=True, redirect_http=False, cert_path="/c.pem", key_path="/k.pem"
)
)
assert "return 308" not in text
# Dann muss Port 80 die Anwendung weiter bedienen.
assert text.count("proxy_pass http://127.0.0.1:5000;") >= 2
def test_deviating_ports_are_rendered_and_redirect_keeps_the_port():
text = webserver.render(
make_config(
https_enabled=True,
cert_path="/c.pem",
key_path="/k.pem",
http_port=8080,
https_port=8443,
)
)
assert "listen 8080 default_server;" in text
assert "listen 8443 ssl default_server http2;" in text
# Ohne den Port im Ziel liefe die Umleitung ins Leere.
assert "return 308 https://$host:8443$request_uri;" in text
def test_standard_port_leaves_no_suffix():
assert make_config(https_port=443).https_suffix == ""
assert make_config(https_port=8443).https_suffix == ":8443"
def test_websocket_location_only_when_configured():
assert "proxy_set_header Upgrade" not in webserver.render(make_config())
text = webserver.render(make_config(websocket_path="/ws/"))
assert "location /ws/ {" in text
assert "proxy_read_timeout 3600s;" in text
def _location_blocks(text: str) -> list[list[str]]:
"""Zerlegt die erzeugte Datei in ihre ``location``-Bloecke."""
blocks: list[list[str]] = []
current: list[str] | None = None
for line in text.splitlines():
stripped = line.strip()
if stripped.startswith("location ") and stripped.endswith("{"):
current = []
elif stripped == "}" and current is not None:
blocks.append(current)
current = None
elif current is not None and stripped and not stripped.startswith("#"):
current.append(stripped)
return blocks
@pytest.mark.parametrize(
"config",
[
make_config(websocket_path="/ws/"),
make_config(
websocket_path="/ws/", https_enabled=True, cert_path="/c.pem", key_path="/k.pem"
),
make_config(https_enabled=True, hsts_enabled=True, cert_path="/c.pem", key_path="/k.pem"),
],
)
def test_no_directive_appears_twice_in_one_block(config: webserver.WebConfig):
"""nginx weist eine doppelte Direktive im selben Block ab.
Auf POETEST real passiert: der Websocket-Block erbte ``proxy_read_timeout
120s`` aus dem gemeinsamen Teil und setzte danach ``3600s``. nginx meldete
"directive is duplicate" und liess sich nicht mehr starten -- nicht nur
fuer diese Site, sondern fuer alle.
"""
for block in _location_blocks(webserver.render(config)):
directives = [line.split()[0] for line in block if line.endswith(";")]
repeated = {
name
for name in directives
# Wiederholung ist nur bei den Direktiven zulaessig, die nginx als
# Liste versteht.
if directives.count(name) > 1 and name not in {"proxy_set_header", "add_header"}
}
assert not repeated, f"Mehrfach im selben Block: {sorted(repeated)}"
def test_websocket_block_keeps_the_long_timeout():
"""Die lange Wartezeit ist der Zweck des Blocks -- sie darf nicht verlorengehen."""
blocks = {
block[0]: block
for block in _location_blocks(webserver.render(make_config(websocket_path="/ws/")))
}
websocket = next(b for b in blocks.values() if "proxy_set_header Upgrade $http_upgrade;" in b)
assert "proxy_read_timeout 3600s;" in websocket
assert "proxy_read_timeout 120s;" not in websocket
def test_http2_uses_the_old_spelling_by_default():
"""``http2 on;`` kennt erst nginx 1.25.1.
Auf POETEST (nginx 1.24) fuehrte die neue Direktive zu "unknown directive"
-- und damit zu einem Webserver, der sich nicht mehr starten liess. Die
alte Schreibweise laeuft ueberall.
"""
text = webserver.render(
make_config(https_enabled=True, cert_path="/c.pem", key_path="/k.pem")
)
assert "listen 443 ssl default_server http2;" in text
assert "http2 on;" not in text
def test_http2_directive_on_request():
text = webserver.render(
make_config(
https_enabled=True, cert_path="/c.pem", key_path="/k.pem", http2_directive=True
)
)
assert " http2 on;" in text
assert "listen 443 ssl default_server;" in text
assert "ssl http2;" not in text
def test_stapling_only_with_a_real_certificate():
"""Ein selbstsigniertes Zertifikat hat keinen OCSP-Responder."""
self_signed = webserver.render(
make_config(https_enabled=True, cert_path="/c.pem", key_path="/k.pem")
)
assert "ssl_stapling" not in self_signed
public = webserver.render(
make_config(
https_enabled=True,
domain="tesm.example.org",
cert_path="/c.pem",
key_path="/k.pem",
)
)
assert "ssl_stapling on;" in public
# --------------------------------------------------------------------------- #
# Domain
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"value", ["tesm.example.org", "a.b.c.de", "TESM.Example.ORG", "tesm.example.org."]
)
def test_valid_domains(value: str):
assert webserver.validate_domain(value) == value.lower().rstrip(".")
@pytest.mark.parametrize(
"value", ["kein-punkt", "-start.de", "ende-.de", "raum .de", "a..b", "http://x.de"]
)
def test_invalid_domains(value: str):
with pytest.raises(ValueError):
webserver.validate_domain(value)
def test_empty_domain_becomes_catch_all():
assert make_config().server_name == "_"
assert make_config(domain="tesm.example.org").server_name == "tesm.example.org"
def test_only_the_nameless_site_claims_default_server():
"""Sonst entscheidet der Dateiname in sites-enabled, wer die Vorgabe wird.
``server_name _;`` ist bei nginx kein Sonderfall -- ohne
``default_server`` gewinnt schlicht die zuerst geladene Datei. Bei zwei
Anwendungen auf einem Host waere das reiner Zufall.
"""
nameless = webserver.render(make_config())
assert "listen 80 default_server;" in nameless
named = webserver.render(make_config(domain="lizenz.example.org"))
assert "default_server" not in named
assert "server_name lizenz.example.org;" in named
# --------------------------------------------------------------------------- #
# Selbstsigniertes Zertifikat
# --------------------------------------------------------------------------- #
class FakeSysOps:
"""Merkt sich den Aufruf, statt den Helfer wirklich zu starten."""
def __init__(self, result: Result) -> None:
self.result = result
self.calls: list[tuple[str, ...]] = []
self.available = True
self.site_key = "tesm-opus"
def issue_self_signed(self, common_name: str, *alt_names: str) -> Result:
self.calls.append((common_name, *alt_names))
return self.result
def test_self_signed_returns_both_paths():
sysops = FakeSysOps(Result(True, "/etc/tesm/certs/tesm/fullchain.pem\n/etc/tesm/certs/tesm/privkey.pem\n"))
result = webserver.create_self_signed(
sysops, common_name="poetest", alt_names=["poetest.local", "192.168.82.202"]
)
assert result["ok"]
assert result["cert_path"].endswith("fullchain.pem")
assert result["key_path"].endswith("privkey.pem")
assert sysops.calls == [("poetest", "poetest.local", "192.168.82.202")]
def test_self_signed_rejects_a_hostile_name():
"""Der Name landet in einem openssl-Argument -- er wird hier geprueft."""
sysops = FakeSysOps(Result(True, "a\nb\n"))
for hostile in ("a b", "a;rm -rf /", "$(whoami)", "../../etc", ""):
result = webserver.create_self_signed(sysops, common_name=hostile)
assert not result["ok"], hostile
assert sysops.calls == []
def test_self_signed_drops_unusable_alternative_names():
sysops = FakeSysOps(Result(True, "a\nb\n"))
webserver.create_self_signed(
sysops, common_name="poetest", alt_names=["poetest", "", "bad name", "ok.local"]
)
assert sysops.calls == [("poetest", "ok.local")]
def test_self_signed_reports_a_failing_helper():
sysops = FakeSysOps(Result(False, "", "openssl fehlt", 2))
result = webserver.create_self_signed(sysops, common_name="poetest")
assert not result["ok"]
assert "openssl fehlt" in result["detail"]
def test_self_signed_rejects_an_unexpected_answer():
sysops = FakeSysOps(Result(True, "nur eine Zeile\n"))
result = webserver.create_self_signed(sysops, common_name="poetest")
assert not result["ok"]
def test_certificate_info_survives_an_unreadable_file(tmp_path, monkeypatch):
"""Ein unlesbares Zertifikat darf die Einstellungsseite nicht abwerfen.
Auf POETEST real passiert: das Zertifikatsverzeichnis war 0750 root:root,
der Dienstbenutzer bekam beim blossen ``stat`` einen PermissionError -- und
die Seite "Webserver und TLS" antwortete mit HTTP 500.
"""
target = tmp_path / "fullchain.pem"
target.write_text("egal", encoding="utf-8")
def deny(*_args, **_kwargs):
raise PermissionError(13, "Permission denied")
monkeypatch.setattr("pathlib.Path.is_file", deny)
info = webserver.certificate_info(str(target))
assert info["present"] is True
assert "nicht lesen" in info["error"]
def test_certificate_info_reports_a_missing_file():
assert webserver.certificate_info("/gibt/es/nicht.pem") == {"present": False}
def test_certificate_info_reports_garbage_without_raising(tmp_path):
target = tmp_path / "fullchain.pem"
target.write_text("kein Zertifikat", encoding="utf-8")
info = webserver.certificate_info(str(target))
assert info["present"] is True
assert info["error"]
def test_local_names_are_usable_in_a_certificate():
names = webserver.local_names()
assert names, "Es sollte mindestens der Hostname gefunden werden."
for name in names:
assert re.fullmatch(r"[A-Za-z0-9]([A-Za-z0-9.-]{0,251}[A-Za-z0-9])?", name), name
# --------------------------------------------------------------------------- #
# Instanztrennung
# --------------------------------------------------------------------------- #
def test_apply_config_is_one_call_with_rollback():
"""Getrennte Aufrufe liessen eine abgelehnte Konfiguration aktiviert liegen.
``nginx-write`` verlinkt sofort nach ``sites-enabled``. Schlug danach
``nginx -t`` fehl, blieb die kaputte Datei aktiv -- der laufende nginx
merkte nichts, aber der naechste Reload durch irgendwen legte den
Webserver lahm. Die Uebernahme muss deshalb *eine* privilegierte Operation
sein, die sich selbst zuruecknimmt.
"""
calls: list[str] = []
class Recording(SysOps):
def run(self, verb: str, *args: str, stdin: str = "", timeout: int | None = None) -> Result:
calls.append(verb)
return Result(True, "uebernommen")
result = webserver.apply_config(Recording(app_key="tesm"), "server {}")
assert result["ok"]
assert calls == ["nginx-apply"]
assert "nginx-write" not in calls
def test_apply_config_reports_a_rejection_without_claiming_success():
class Failing(SysOps):
def run(self, verb: str, *args: str, stdin: str = "", timeout: int | None = None) -> Result:
return Result(False, "", "unknown directive \"http2\"", 2)
result = webserver.apply_config(Failing(app_key="tesm"), "server {}")
assert not result["ok"]
assert "alte Stand" in result["message"]
assert "http2" in result["detail"]
def test_sysops_writes_system_files_under_the_site_key():
"""Sonst ueberschreibt eine Testinstanz die Systemdateien der Produktion."""
calls: list[tuple[str, ...]] = []
class Recording(SysOps):
def run(self, verb: str, *args: str, stdin: str = "", timeout: int | None = None) -> Result:
calls.append((verb, *args))
return Result(True)
sysops = Recording(app_key="tesm", site_key="tesm-opus")
sysops.write_nginx("x")
sysops.write_netplan("x")
sysops.write_logrotate("x")
sysops.issue_self_signed("poetest")
assert calls == [
("nginx-write", "tesm-opus"),
("netplan-write", "tesm-opus"),
("logrotate-write", "tesm-opus"),
("selfsigned-issue", "tesm-opus", "poetest"),
]
def test_upstream_follows_the_configured_bind_address(monkeypatch):
"""Die nginx-Site muss auf *diese* Instanz zeigen, nicht auf Port 5000.
Mit fest verdrahteter 5000 erzeugte eine Instanz auf Port 5100 eine
Konfiguration, die Anfragen an die *andere* Installation weiterreichte --
auf POETEST reproduziert: HTTPS stand, lieferte aber die alte Anwendung.
"""
from tesm_core.config import load_core_config
monkeypatch.setenv("TESM_BIND", "127.0.0.1:5100")
config = load_core_config(app_key="tesm", app_name="TESM", env_prefix="TESM_")
assert config.bind == "127.0.0.1:5100"
text = webserver.render(make_config(upstream=config.bind))
assert "proxy_pass http://127.0.0.1:5100;" in text
assert "127.0.0.1:5000" not in text
def test_bind_address_is_validated(monkeypatch):
from tesm_core.config import load_core_config
monkeypatch.setenv("TESM_BIND", "127.0.0.1:5100; rm -rf /")
with pytest.raises(ValueError):
load_core_config(app_key="tesm", app_name="TESM", env_prefix="TESM_")
def test_site_key_is_validated(monkeypatch):
from tesm_core.config import load_core_config
monkeypatch.setenv("TESM_SITE_KEY", "../../etc/nginx")
with pytest.raises(ValueError):
load_core_config(app_key="tesm", app_name="TESM", env_prefix="TESM_")
def test_site_key_defaults_to_the_app_key():
assert SysOps(app_key="tesm").site_key == "tesm"
def test_self_signed_verb_is_known():
assert "selfsigned-issue" in HELPER_VERBS
def test_service_state_finds_unit_without_suffix(monkeypatch):
"""``list-unit-files`` braucht ein Muster mit Endung.
Ohne sie fand die Abfrage nichts und der Zustand lautete
"installiert: nein" -- auch bei laufendem Dienst. Auf der DHCP-Seite lud
das zum Klick auf "Kea installieren" ein.
"""
import tesm_core.sysops as sysops_modul
from tesm_core.sysops import Result, SysOps
# Ohne systemctl bricht die Abfrage vorher ab -- auf dem Entwicklungsrechner
# gibt es keines.
monkeypatch.setattr(sysops_modul.shutil, "which", lambda name: "/bin/" + name)
gesehen: list[list[str]] = []
class Attrappe(SysOps):
def __init__(self):
pass
def read_command(self, argv):
gesehen.append(list(argv))
if "list-unit-files" in argv:
treffer = "kea-dhcp4-server.service" in argv
return Result(True, "kea-dhcp4-server.service enabled" if treffer else "", "")
return Result(True, "active", "")
zustand = Attrappe().service_state("kea-dhcp4-server")
assert zustand["installed"] == "ja", gesehen
# Eine bereits vollstaendige Unit bleibt unveraendert.
assert Attrappe().service_state("kea-dhcp4-server.service")["installed"] == "ja"