TESM-Lizenzserver 2.0.0 -- Neubau

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

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

Die wichtigsten Unterschiede zum Vorgaenger:

* Keine doppelte licensing.py -- ein Paket, das beide Anwendungen
  installieren, statt zweier Dateien, die byte-identisch bleiben sollen.
* 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.
* Keine Lizenz ohne master_endpoint: eine Ausstellung ohne Endpunkt wird
  abgelehnt statt eine Lizenz zu erzeugen, die sich nie aktivieren kann.
* Offline-Aktivierung in beide Richtungen; die Lizenz bleibt als
  "Aktivierung offen" markiert, bis sie zurueckkommt.
* Getrennte Signaturkontexte je Nachrichtenart, Nonce gegen Wiedereinspielung,
  seq gegen das Zurueckrollen auf eine aeltere Lizenz.
* Kein Hostname im Maschinen-Fingerabdruck.
* Verschachtelte Datenbankverbindungen sind ein Fehler, kein Deadlock.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
2026-09-02 18:01:23 +02:00
co-authored by Claude Opus 5
parent c130c85b17
commit f7805a2180
211 changed files with 31099 additions and 14759 deletions
+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
+493
View File
@@ -0,0 +1,493 @@
"""End-to-End: der komplette Lizenz-Lebenszyklus ueber beide Anwendungen hinweg.
Laeuft auf dem Zielhost gegen zwei echte, laufende Installationen:
Lizenzserver: Kunde -> Ticket -> Lizenz ausstellen -> Bundle herunterladen
TESM : Bundle einspielen -> aktivieren (online gegen den Lizenzserver)
-> Module freigeschaltet -> Heartbeat -> Revision -> Widerruf
Das ist der Pfad, der im Vorgaengerprojekt nur von Hand pruefbar war -- und
genau dort sassen die teuren Fehler. Geprueft wird deshalb nicht nur "es geht",
sondern auch: Doppelausstellung wird verhindert, Module sind vor der
Aktivierung wirklich gesperrt, eine neue Revision nimmt Module wieder weg, und
ein Widerruf kommt beim Client an.
Aufruf:
python3 e2e_license.py http://127.0.0.1:8081 http://127.0.0.1:8080 admin 'Passwort'
"""
from __future__ import annotations
import http.cookiejar
import json
import re
import ssl
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="([^"]+)"')
PASSED: list[str] = []
FAILED: list[str] = []
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}")
class Response:
def __init__(self, status: int, body: bytes, url: str, headers: dict[str, str]) -> None:
self.status = status
self.raw = body
self.url = url
self.headers = headers
@property
def text(self) -> str:
return self.raw.decode("utf-8", "replace")
class Client:
def __init__(self, base: str, label: str) -> None:
self.base = base.rstrip("/")
self.label = label
jar = http.cookiejar.CookieJar()
# Der Test laeuft auf dem Zielhost gegen dessen eigenes, selbst
# ausgestelltes Zertifikat. Geprueft wird, dass TLS steht und die
# richtige Anwendung antwortet -- nicht die Vertrauenskette.
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
self.opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(jar),
urllib.request.HTTPSHandler(context=context),
)
self.opener.addheaders = [("User-Agent", "tesm-e2e-license/2.0")]
def _open(self, request: urllib.request.Request) -> Response:
try:
with self.opener.open(request, timeout=60) as response:
return Response(
response.status, response.read(), response.geturl(), dict(response.headers)
)
except urllib.error.HTTPError as exc:
return Response(exc.code, exc.read(), exc.url, dict(exc.headers))
def get(self, path: str) -> Response:
return self._open(urllib.request.Request(self.base + path))
def text(self, path: str) -> str:
return self.get(path).text
def post(self, path: str, data: dict[str, object]) -> Response:
pairs = [
(key, str(item))
for key, value in data.items()
for item in (value if isinstance(value, (list, tuple)) else [value])
]
body = urllib.parse.urlencode(pairs).encode("utf-8")
request = urllib.request.Request(
self.base + path,
data=body,
headers={
"Content-Type": "application/x-www-form-urlencoded",
"Origin": self.base,
"Referer": self.base + path,
},
)
return self._open(request)
def post_file(
self, path: str, field: str, filename: str, content: bytes, extra: dict[str, str]
) -> Response:
boundary = "----tesm-e2e-boundary-3f88"
parts: list[bytes] = []
for key, value in extra.items():
parts.append(
f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n'
f"{value}\r\n".encode()
)
parts.append(
f'--{boundary}\r\nContent-Disposition: form-data; name="{field}"; '
f'filename="{filename}"\r\nContent-Type: application/json\r\n\r\n'.encode()
+ content
+ b"\r\n"
)
parts.append(f"--{boundary}--\r\n".encode())
request = urllib.request.Request(
self.base + path,
data=b"".join(parts),
headers={
"Content-Type": f"multipart/form-data; boundary={boundary}",
"Origin": self.base,
"Referer": self.base + path,
},
)
return self._open(request)
def token(self, path: str) -> str:
body = self.text(path)
match = CSRF_RE.search(body) or BODY_CSRF_RE.search(body)
if not match:
raise AssertionError(f"{self.label}: kein CSRF-Token auf {path}")
return match.group(1)
def login(self, username: str, password: str) -> bool:
token = self.token("/login")
response = self.post(
"/login", {"username": username, "password": password, "csrf_token": token}
)
return response.status == 200 and "/login" not in response.url
def main(
server_base: str,
client_base: str,
username: str,
password: str,
endpoint: str = "",
) -> int:
# Der Endpunkt, den TESM spaeter anruft, muss nicht die Adresse sein, unter
# der ein Mensch den Lizenzserver bedient. Auf einem Host, der beide
# Anwendungen traegt, ist der direkte Weg ueber die Loopback-Adresse der
# ehrlichere -- er umgeht Namensaufloesung und Zertifikatspruefung, die
# zwischen zwei Diensten derselben Maschine nichts beitragen.
endpoint = (endpoint or server_base).rstrip("/")
server = Client(server_base, "Lizenzserver")
tesm = Client(client_base, "TESM")
print(f"\n== Lizenz-Lebenszyklus: {server_base} -> {client_base} ==\n")
print("Anmeldung:")
check("Lizenzserver", server.login(username, password), "fehlgeschlagen")
check("TESM", tesm.login(username, password), "fehlgeschlagen")
if FAILED:
print("\nOhne Anmeldung ist der Rest sinnlos -- Abbruch.")
return 1
# -- Anbieterdaten: der Endpunkt landet spaeter in jeder Lizenz -----------
print("\nAnbieterdaten und Endpunkt:")
token = server.token("/einstellungen/anbieter")
response = server.post(
"/einstellungen/anbieter",
{
"csrf_token": token,
"vendor_name": "WiS GmbH",
"vendor_email": "lizenzen@wis.invalid",
"vendor_phone": "+49 000 000000",
"vendor_address": "Musterweg 1, 12345 Musterstadt",
"master_endpoint": endpoint,
},
)
check(
"Anbieterdaten gespeichert",
"gespeichert" in response.text or "Keine Aenderungen" in response.text,
response.text[:150],
)
token = server.token("/einstellungen/anbieter")
response = server.post(
"/einstellungen/anbieter", {"csrf_token": token, "master_endpoint": "ftp://falsch"}
)
check("Ungueltiger Endpunkt abgelehnt", "muss mit http" in response.text, "keine Meldung")
# -- Kunde ---------------------------------------------------------------
print("\nKunde:")
token = server.token("/kunden/")
response = server.post(
"/kunden/neu",
{
"csrf_token": token,
"name": "Beispiel AG",
"contact_email": "it@beispiel.invalid",
"contact_person": "Frau Muster",
"street": "Hauptstrasse",
"house_number": "7",
"postal_code": "12345",
"city": "Musterstadt",
},
)
check("Kunde angelegt", "angelegt" in response.text, response.text[:150])
match = re.search(r"/kunden/(\d+)$", response.url)
customer_id = match.group(1) if match else ""
check("Kunden-Nummer erhalten", bool(customer_id), response.url)
# -- Ticket ---------------------------------------------------------------
print("\nTicket:")
token = server.token("/tickets/neu")
response = server.post(
"/tickets/neu",
{
"csrf_token": token,
"customer_id": customer_id,
"type": "enterprise",
"modules": ["dhcp", "fileshare", "maintenance"],
"valid_days": "365",
"role": "customer",
"notes": "E2E-Test",
},
)
check("Ticket angelegt", "angelegt" in response.text, response.text[:150])
match = re.search(r"/tickets/([^/]+)$", response.url)
ticket_id = match.group(1) if match else ""
check("Ticket-Kennung erhalten", bool(ticket_id), response.url)
token = server.token("/tickets/neu")
response = server.post(
"/tickets/neu",
{
"csrf_token": token,
"customer_id": customer_id,
"type": "trial",
"valid_days": "30",
"lifetime": "1",
"role": "customer",
},
)
check(
"Testlizenz kann nicht unbefristet sein",
"nicht unbefristet" in response.text,
"Regel greift nicht",
)
# -- Lizenz ausstellen ----------------------------------------------------
print("\nLizenz ausstellen:")
token = server.token(f"/tickets/{ticket_id}")
response = server.post(f"/lizenzen/aus-ticket/{ticket_id}", {"csrf_token": token})
check("Lizenz ausgestellt", "ausgestellt" in response.text, response.text[:200])
match = re.search(r"/lizenzen/([0-9a-f-]{36})$", response.url)
license_id = match.group(1) if match else ""
check("Lizenz-Kennung erhalten", bool(license_id), response.url)
token = server.token(f"/tickets/{ticket_id}")
response = server.post(f"/lizenzen/aus-ticket/{ticket_id}", {"csrf_token": token})
check(
"Doppelklick stellt keine zweite Lizenz aus",
"bereits eine Lizenz offen" in response.text,
"zweite Lizenz entstanden",
)
# -- Bundle herunterladen -------------------------------------------------
print("\nAuslieferung:")
response = server.get(f"/lizenzen/{license_id}/datei")
check("Bundle heruntergeladen", response.status == 200, f"HTTP {response.status}")
check(
"Als Anhang ausgeliefert",
"attachment" in response.headers.get("Content-Disposition", ""),
response.headers.get("Content-Disposition", ""),
)
check(
"Nicht zwischenspeicherbar",
"no-store" in response.headers.get("Cache-Control", ""),
response.headers.get("Cache-Control", ""),
)
bundle_bytes = response.raw
try:
bundle = json.loads(bundle_bytes.decode("utf-8"))
except ValueError:
check("Bundle ist gueltiges JSON", False, bundle_bytes[:120].decode("latin1"))
return 1
check("Bundle ist gueltiges JSON", True)
check(
"Dokument und privater Clientschluessel enthalten",
"license" in bundle and "client_key" in bundle,
str(sorted(bundle))[:100],
)
document = bundle.get("license", {})
check(
"Endpunkt eingebettet",
document.get("master_endpoint") == endpoint,
str(document.get("master_endpoint")),
)
check(
"Alle drei Module enthalten",
set(document.get("modules", [])) == {"dhcp", "fileshare", "maintenance"},
str(document.get("modules")),
)
check("Rolle ist customer", document.get("role") == "customer", str(document.get("role")))
body = server.text(f"/lizenzen/{license_id}")
check("Zustellprotokoll vermerkt den Download", "ownload" in body, "kein Eintrag")
# -- Einspielen in TESM ---------------------------------------------------
print("\nEinspielen in TESM:")
body = tesm.text("/lizenz/")
if "keine Lizenz installiert" in body:
check("Vorher keine Lizenz installiert", True)
else:
# Bei einem wiederholten Lauf liegt noch die Lizenz des Vorlaufs. Das ist
# kein Fehler -- das Einspielen ersetzt sie, und genau das wird gleich
# geprueft.
print(" [info] Es liegt bereits eine Lizenz -- wiederholter Lauf, wird ersetzt.")
token = tesm.token("/lizenz/")
response = tesm.post_file(
"/lizenz/einspielen", "license_file", "lizenz.json", bundle_bytes, {"csrf_token": token}
)
check("Lizenz eingespielt", "eingespielt" in response.text, response.text[:250])
check(
"Aussteller beim ersten Mal gepinnt",
"Aussteller gepinnt" in response.text,
"kein Pinning erkennbar",
)
check("Noch nicht aktiviert", "nicht aktiviert" in response.text, "Zustand unerwartet")
print("\nModule vor der Aktivierung:")
for path in ("/dhcp/", "/wartung/", "/dateifreigaben/"):
check(f"{path} gesperrt", tesm.get(path).status == 404, "erreichbar")
# -- Manipuliertes Bundle -------------------------------------------------
print("\nManipuliertes Bundle:")
tampered = json.loads(bundle_bytes.decode("utf-8"))
tampered["license"]["modules"] = ["dhcp", "fileshare", "maintenance", "erfunden"]
token = tesm.token("/lizenz/")
response = tesm.post_file(
"/lizenz/einspielen",
"license_file",
"boese.json",
json.dumps(tampered).encode("utf-8"),
{"csrf_token": token},
)
check(
"Veraenderte Lizenz wird abgewiesen",
"eingespielt" not in response.text,
"Manipulation akzeptiert!",
)
# -- Aktivierung ----------------------------------------------------------
print("\nAktivierung (online, echter HTTP-Aufruf an den Lizenzserver):")
token = tesm.token("/lizenz/")
response = tesm.post("/lizenz/aktivieren", {"csrf_token": token})
check("Aktivierung erfolgreich", "jetzt aktiv" in response.text, response.text[:300])
check("Zustand aktiviert", "nicht aktiviert" not in response.text, "Zustand unerwartet")
print("\nModule nach der Aktivierung:")
for path, needle in (
("/dhcp/", "DHCP"),
("/wartung/", "Wartung"),
("/dateifreigaben/", "Dateifreigaben"),
):
response = tesm.get(path)
check(
f"{path} freigeschaltet",
response.status == 200 and needle in response.text,
f"HTTP {response.status}",
)
print("\nSicht des Lizenzservers:")
body = server.text(f"/lizenzen/{license_id}")
check("Maschine gebunden", "noch nicht gebunden" not in body, "kein Fingerprint")
check("Aktivierungszeitpunkt vermerkt", "Aktiviert" in body, "fehlt")
# -- Heartbeat ------------------------------------------------------------
print("\nHeartbeat:")
token = tesm.token("/lizenz/")
response = tesm.post("/lizenz/heartbeat", {"csrf_token": token})
check("Heartbeat bestaetigt", "bestaetigt" in response.text, response.text[:250])
# -- Neue Revision: Modul entziehen ---------------------------------------
print("\nNeue Revision (Modul entziehen):")
token = server.token(f"/lizenzen/{license_id}")
response = server.post(
f"/lizenzen/{license_id}/neu-signieren",
{"csrf_token": token, "modules": ["dhcp"], "extend_days": ""},
)
check("Revision signiert", "Revision" in response.text, response.text[:250])
token = tesm.token("/lizenz/")
response = tesm.post("/lizenz/heartbeat", {"csrf_token": token})
check("Heartbeat holt die Revision", "bestaetigt" in response.text, response.text[:200])
check("Entzogenes Modul gesperrt", tesm.get("/wartung/").status == 404, "weiter offen")
check("Verbliebenes Modul offen", tesm.get("/dhcp/").status == 200, "gesperrt")
# -- Deaktivierung und erneute Aktivierung --------------------------------
print("\nDeaktivierung (Systemwechsel):")
token = tesm.token("/lizenz/")
response = tesm.post("/lizenz/deaktivieren", {"csrf_token": token})
check("Deaktivierung erfolgreich", "wurde deaktiviert" in response.text, response.text[:250])
check("Module wieder gesperrt", tesm.get("/dhcp/").status == 404, "weiter offen")
body = server.text(f"/lizenzen/{license_id}")
check("Server kennt die Deaktivierung", "deaktiviert" in body, "unerwartet")
print("\nErneute Aktivierung derselben Maschine:")
token = tesm.token("/lizenz/")
response = tesm.post("/lizenz/aktivieren", {"csrf_token": token})
check("Wieder aktiv", "jetzt aktiv" in response.text, response.text[:250])
check("Modul wieder offen", tesm.get("/dhcp/").status == 200, "gesperrt")
# -- Widerruf -------------------------------------------------------------
print("\nWiderruf durch den Anbieter:")
token = server.token(f"/lizenzen/{license_id}")
response = server.post(
f"/lizenzen/{license_id}/widerrufen", {"csrf_token": token, "reason": "E2E-Test"}
)
check("Widerruf gespeichert", "widerrufen" in response.text, response.text[:250])
token = tesm.token("/lizenz/")
response = tesm.post("/lizenz/heartbeat", {"csrf_token": token})
check("Client erfaehrt den Widerruf", "widerrufen" in response.text, response.text[:250])
check("Module nach Widerruf gesperrt", tesm.get("/dhcp/").status == 404, "weiter offen")
check("Lizenzseite bleibt erreichbar", tesm.get("/lizenz/").status == 200, "gesperrt")
# -- Aufraeumen -------------------------------------------------------
# Der Test endet mit einem Widerruf -- danach sind DHCP, Wartung und
# Dateifreigaben gesperrt. Eine Pruefung, die die Installation unbrauchbar
# zuruecklaesst, ist keine gute Pruefung: wer danach die Wartung sucht,
# findet einen 404 und haelt ihn fuer einen Fehler. Also wird zum Schluss
# eine frische Lizenz ausgestellt und aktiviert.
print("\nAufraeumen -- Installation wieder betriebsbereit machen:")
token = server.token(f"/tickets/{ticket_id}")
response = server.post(f"/lizenzen/aus-ticket/{ticket_id}", {"csrf_token": token})
check("Ersatzlizenz ausgestellt", "ausgestellt" in response.text, response.text[:200])
match = re.search(r"/lizenzen/([0-9a-f-]{36})$", response.url)
replacement_id = match.group(1) if match else ""
check("Kennung der Ersatzlizenz", bool(replacement_id), response.url)
response = server.get(f"/lizenzen/{replacement_id}/datei")
check("Ersatzbundle heruntergeladen", response.status == 200, f"HTTP {response.status}")
token = tesm.token("/lizenz/")
response = tesm.post_file(
"/lizenz/einspielen", "license_file", "lizenz.json", response.raw, {"csrf_token": token}
)
check("Ersatzlizenz eingespielt", "eingespielt" in response.text, response.text[:200])
token = tesm.token("/lizenz/")
response = tesm.post("/lizenz/aktivieren", {"csrf_token": token})
check("Ersatzlizenz aktiv", "jetzt aktiv" in response.text, response.text[:250])
for path in ("/dhcp/", "/wartung/", "/dateifreigaben/"):
check(f"{path} wieder erreichbar", tesm.get(path).status == 200, "weiter gesperrt")
# -- Protokolle -----------------------------------------------------------
print("\nProtokolle und Diagnose:")
body = server.text("/protokolle/aenderungen")
for action in ("license.issued", "license.downloaded", "license.resigned", "license.revoked"):
check(f"Eintrag {action}", action in body, "fehlt")
check(
"Diagnose Lizenzserver: Kette unversehrt",
"unversehrt" in server.text("/diagnose/"),
"gebrochen",
)
body = tesm.text("/protokolle/aenderungen")
for action in ("license.activate", "license.deactivate", "license.heartbeat"):
check(f"TESM-Eintrag {action}", action in body, "fehlt")
check("Diagnose TESM: Kette unversehrt", "unversehrt" in tesm.text("/diagnose/"), "gebrochen")
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) not in (5, 6):
print(__doc__)
raise SystemExit(2)
raise SystemExit(main(*sys.argv[1:]))
+251
View File
@@ -0,0 +1,251 @@
"""End-to-End-Rauchtest des Lizenzservers gegen eine laufende Installation.
Gegenstueck zu ``e2e_smoke.py`` (das prueft TESM). Beide Anwendungen teilen
sich den Kern; hier geht es um die Seiten und Schranken, die nur der
Lizenzserver hat -- Kunden, Tickets, Lizenzen, Kundenportal, Selbstbedienung
und die Maschinen-API.
Aufruf:
python3 e2e_license_server.py https://lizenz.example.org admin 'Passwort'
"""
from __future__ import annotations
import http.cookiejar
import json
import re
import ssl
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="([^"]+)"')
PASSED: list[str] = []
FAILED: list[str] = []
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}")
class Client:
def __init__(self, base: str) -> None:
self.base = base.rstrip("/")
self.https = self.base.startswith("https://")
jar = http.cookiejar.CookieJar()
context = ssl.create_default_context()
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
tls = urllib.request.HTTPSHandler(context=context)
self.opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(jar), tls
)
self.opener.addheaders = [("User-Agent", "tesm-e2e-lic/2.0")]
class _NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, *_args, **_kwargs): # type: ignore[override]
return None
self.plain = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(jar), _NoRedirect(), tls
)
self.plain.addheaders = [("User-Agent", "tesm-e2e-lic/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=30) 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:
try:
with self.plain.open(urllib.request.Request(self.base + path), timeout=30) 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=45) 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 post_json(self, path: str, payload: dict[str, object]) -> tuple[int, str]:
request = urllib.request.Request(
self.base + path,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"},
)
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)
def main(base: str, username: str, password: str) -> int:
client = Client(base)
print(f"\n== Rauchtest Lizenzserver gegen {base} ==\n")
print("Oeffentlich erreichbar:")
status, body, headers = client.get("/gesundheit")
check(
"Gesundheitsendpunkt",
status == 200 and json.loads(body)["status"] == "ok",
body[:80],
)
check(
"Antwortet als Lizenzserver",
json.loads(body).get("version") == "tesm-license",
body[:80],
)
status, body, headers = client.get("/login")
check("Anmeldeseite", status == 200 and "Anmelden" in body, f"HTTP {status}")
csp = headers.get("Content-Security-Policy", "")
check("CSP mit Nonce", "nonce-" in csp, csp[:80])
check(
"X-Frame-Options",
headers.get("X-Frame-Options") == "DENY",
str(headers.get("X-Frame-Options")),
)
if client.https:
cookie = headers.get("Set-Cookie", "")
check("Sitzungscookie mit Secure", "Secure" in cookie, cookie[:80] or "kein Cookie")
print("\nKundenportal (ohne Anmeldung erreichbar):")
status, body, _ = client.get("/portal/")
check("Portalseite", status == 200 and "Kundenportal" in body, f"HTTP {status}")
check("Fragt nach der Kennung, nicht nach einem Passwort", "Passwort" not in body, "unerwartet")
print("\nSchranken ohne Anmeldung:")
for path in ("/kunden/", "/tickets/", "/lizenzen/", "/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}")
print("\nMaschinen-API (eigene Authentisierung, kein Cookie):")
status, body = client.post_json("/api/v1/licenses/activate", {"unsinn": True})
check(
"Unsinniger Antrag wird abgewiesen",
status in (400, 401, 403, 404, 422),
f"HTTP {status}",
)
check(
"Antwort ist maschinenlesbar, nicht HTML",
body.lstrip().startswith("{") and "error" in body,
body[:100],
)
check("Keine Interna in der Antwort", "Traceback" not in body, "Stacktrace nach aussen")
check(
"Kein CSRF-Token noetig (Maschinenschnittstelle)",
"Sicherheitstoken" not in body,
"API verlangt ein Formulartoken",
)
print("\nAnmeldung:")
token = client.token("/login")
status, body = client.post(
"/login", {"username": username, "password": password, "csrf_token": token}
)
check("Anmeldung erfolgreich", status == 200 and "Anmelden" not in body, f"HTTP {status}")
print("\nSeiten (angemeldet):")
pages = [
("/", "Uebersicht"),
("/kunden/", "Kunden"),
("/tickets/", "Tickets"),
("/tickets/neu", "Ticket"),
("/lizenzen/", "Lizenzen"),
("/lizenzen/offline-code", "Offline"),
("/lizenzen/sub-lizenzserver", "Lizenzserver"),
("/eigene-lizenz/", "Lizenz"),
("/einstellungen/system", "System"),
("/einstellungen/anbieter", "Anbieter"),
("/einstellungen/email", "Mail"),
("/einstellungen/netzwerk", "Netzwerk"),
("/einstellungen/webserver", "Webserver"),
("/einstellungen/verzeichnisdienst", "Verzeichnisdienst"),
("/protokolle/aenderungen", "Aenderungsprotokoll"),
("/sicherung/", "Sicherung"),
("/papierkorb/", "Papierkorb"),
("/diagnose/", "Diagnose"),
("/verwaltung/benutzer", "Benutzer"),
("/verwaltung/gruppen", "Gruppen"),
("/konto/sicherheit", "Sicherheit"),
]
for path, needle in pages:
status, body, _ = client.get(path)
check(
f"{path}",
status == 200 and needle in body and "Interner Fehler" not in body,
f"HTTP {status}",
)
print("\nEigene Master-Lizenz:")
status, body, _ = client.get("/eigene-lizenz/")
check("Master-Rolle erkannt", "master" in body.lower(), "Rolle fehlt")
print("\nDiagnose:")
status, body, _ = client.get("/diagnose/")
check("Audit-Kette unversehrt", "unversehrt" in body, "Kette gebrochen?")
check("Signaturschluessel vorhanden", "Signatur" in body, "kein Hinweis")
print("\nAbmeldung:")
token = client.token("/konto/")
client.post("/logout", {"csrf_token": token})
check("Nach Abmeldung gesperrt", client.get_raw("/kunden/") in (302, 401), "weiter offen")
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]))
+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"
+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
+703
View File
@@ -0,0 +1,703 @@
"""Tests des Lizenzservers: Ausstellung, Aktivierung, Portal, Rollen.
Deckt die Regeln ab, die der Vorgaenger schmerzhaft gelernt hat -- und die
Haertungen, die hier neu sind (Replay, Nonce, Rollen-Sperre, Portal-TTL).
"""
from __future__ import annotations
import json
from typing import Any
import pytest
import tesm_licensing as lic
@pytest.fixture()
def ls_app(instance_env: Any):
from tesm_license import create_app
return create_app(TESTING=True)
@pytest.fixture()
def ls_client(ls_app: Any):
return ls_app.test_client()
@pytest.fixture()
def ls_admin(ls_app: Any):
from tesm_core.extension import core
from tesm_core.security import passwords
extension = core(ls_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 ls_login(ls_client: Any, ls_admin: dict[str, str]):
def do() -> Any:
return ls_client.post(
"/login",
data={
"username": ls_admin["username"],
"password": ls_admin["password"],
"csrf_token": _token(ls_client),
},
follow_redirects=True,
)
return do
@pytest.fixture()
def bootstrapped(ls_app: Any, ls_client: Any, ls_login: Any):
"""Angemeldeter Admin mit Signaturschluessel und Vendor-Angaben."""
from tesm_core.extension import core
ls_login()
extension = core(ls_app)
with ls_app.app_context(), extension.database.session() as conn:
with extension.database.transaction(conn):
extension.settings.set(conn, "vendor_name", "WiS GmbH", actor="test")
extension.settings.set(
conn, "master_endpoint", "https://lizenz.example.invalid", actor="test"
)
return ls_client
def _create_customer(client: Any, name: str = "Beispiel AG") -> int:
token = _token(client, "/kunden/")
client.post(
"/kunden/neu",
data={
"csrf_token": token,
"name": name,
"contact_email": "it@example.invalid",
"contact_person": "Frau Muster",
},
follow_redirects=True,
)
return 1
def _create_ticket(client: Any, customer_id: int = 1, **overrides: Any) -> str:
token = _token(client, "/tickets/neu")
data = {
"csrf_token": token,
"customer_id": str(customer_id),
"type": "standard",
"valid_days": "365",
}
data.update(overrides)
response = client.post("/tickets/neu", data=data, follow_redirects=False)
assert response.status_code == 302, response.get_data(as_text=True)[:400]
return response.headers["Location"].rstrip("/").split("/")[-1]
# --------------------------------------------------------------------------- #
# Start
# --------------------------------------------------------------------------- #
def test_app_boots(ls_app):
from tesm_core.extension import core
components = {row["component"] for row in core(ls_app).migration_status()}
assert components == {"core", "tesm_license"}
def test_dashboard_requires_login(ls_client):
"""Anders als TESMs Geraeteboard ist diese Startseite intern."""
response = ls_client.get("/")
assert response.status_code == 302
assert "/login" in response.headers["Location"]
LS_PAGES = [
"/",
"/kunden/",
"/tickets/",
"/tickets/neu",
"/lizenzen/",
"/lizenzen/offline-code",
"/lizenzen/sub-lizenzserver",
"/einstellungen/system",
"/einstellungen/anbieter",
"/einstellungen/email",
"/einstellungen/netzwerk",
"/einstellungen/webserver",
"/einstellungen/verzeichnisdienst",
"/eigene-lizenz/",
"/eigene-lizenz/signaturschluessel",
"/sicherung/",
"/papierkorb/",
"/diagnose/",
"/protokolle/laufend",
"/protokolle/aenderungen",
"/verwaltung/benutzer",
"/verwaltung/gruppen",
]
@pytest.mark.parametrize("path", LS_PAGES)
def test_pages_render(bootstrapped, path):
response = bootstrapped.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_portal_pages_public(ls_client):
assert ls_client.get("/portal/").status_code == 200
assert ls_client.get("/api/v1/licenses/hinweis").get_json()["service"] == "tesm-license"
# --------------------------------------------------------------------------- #
# Kunde, Ticket, Lizenz
# --------------------------------------------------------------------------- #
def test_customer_ticket_license_flow(bootstrapped, ls_app):
from tesm_core.extension import core
from tesm_license.services import issuance
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client, type="enterprise")
detail = client.get(f"/tickets/{ticket_id}").get_data(as_text=True)
assert "Lizenz ausstellen" in detail
response = client.post(
f"/lizenzen/aus-ticket/{ticket_id}",
data={"csrf_token": _token(client, f"/tickets/{ticket_id}")},
follow_redirects=True,
)
assert "ausgestellt" in response.get_data(as_text=True)
extension = core(ls_app)
with extension.database.session() as conn:
rows = issuance.list_licenses(conn)
assert len(rows) == 1
document = json.loads(str(rows[0]["license_document"]))
lic.verify_license(document)
assert set(document["modules"]) == set(lic.ALL_MODULES)
assert document["role"] == "customer"
assert rows[0]["client_key_encrypted"]
assert "client_key" not in json.dumps(document)
def test_only_one_open_license_per_ticket(bootstrapped, ls_app):
from tesm_core.extension import core
from tesm_license.services import issuance
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client)
token = _token(client, f"/tickets/{ticket_id}")
client.post(f"/lizenzen/aus-ticket/{ticket_id}", data={"csrf_token": token}, follow_redirects=True)
second = client.post(
f"/lizenzen/aus-ticket/{ticket_id}", data={"csrf_token": token}, follow_redirects=True
)
assert "bereits eine Lizenz offen" in second.get_data(as_text=True)
extension = core(ls_app)
with extension.database.session() as conn:
assert len(issuance.list_licenses(conn)) == 1
def test_trial_override_does_not_change_ticket(bootstrapped, ls_app):
from tesm_core.extension import core
from tesm_license.services import issuance
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client, type="standard", valid_days="730")
client.post(
f"/lizenzen/aus-ticket/{ticket_id}",
data={"csrf_token": _token(client, f"/tickets/{ticket_id}"), "as_trial": "1"},
follow_redirects=True,
)
extension = core(ls_app)
with extension.database.session() as conn:
ticket = issuance.get_ticket(conn, ticket_id)
row = issuance.list_licenses(conn)[0]
assert ticket["type"] == "standard"
assert ticket["valid_days"] == 730
assert row["type"] == "trial"
assert set(json.loads(str(row["modules"]))) == set(lic.ALL_MODULES)
def test_trial_lifetime_rejected(bootstrapped):
client = bootstrapped
_create_customer(client)
response = client.post(
"/tickets/neu",
data={
"csrf_token": _token(client, "/tickets/neu"),
"customer_id": "1",
"type": "trial",
"valid_days": "30",
"lifetime": "1",
},
follow_redirects=True,
)
assert "nicht unbefristet" in response.get_data(as_text=True)
def test_ticket_delete_blocked_with_open_license(bootstrapped):
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client)
token = _token(client, f"/tickets/{ticket_id}")
client.post(f"/lizenzen/aus-ticket/{ticket_id}", data={"csrf_token": token}, follow_redirects=True)
response = client.post(
f"/tickets/{ticket_id}/loeschen", data={"csrf_token": token}, follow_redirects=True
)
assert "offene Lizenz" in response.get_data(as_text=True)
def test_customer_delete_blocked_with_tickets(bootstrapped):
client = bootstrapped
_create_customer(client)
_create_ticket(client)
response = client.post(
"/kunden/1/loeschen",
data={"csrf_token": _token(client, "/kunden/1")},
follow_redirects=True,
)
assert "Ticket" in response.get_data(as_text=True)
def test_blocked_ticket_prevents_issuing(bootstrapped):
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client)
token = _token(client, f"/tickets/{ticket_id}")
client.post(f"/tickets/{ticket_id}/sperre", data={"csrf_token": token}, follow_redirects=True)
response = client.post(
f"/lizenzen/aus-ticket/{ticket_id}", data={"csrf_token": token}, follow_redirects=True
)
assert "gesperrt" in response.get_data(as_text=True)
# --------------------------------------------------------------------------- #
# Rollen-Sperre
# --------------------------------------------------------------------------- #
def test_license_server_role_downgraded_without_master_license(bootstrapped, ls_app):
"""Ohne eigene Master-Lizenz wird die Rolle still auf 'customer' gesetzt."""
from tesm_core.extension import core
from tesm_license.services import issuance
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client, role="license_server")
extension = core(ls_app)
with extension.database.session() as conn:
ticket = issuance.get_ticket(conn, ticket_id)
assert ticket["role"] == "customer"
def test_ticket_new_hides_license_server_option(bootstrapped):
body = bootstrapped.get("/tickets/neu").get_data(as_text=True)
assert "Lizenzserver-Lizenz" not in body
# --------------------------------------------------------------------------- #
# Aktivierungsprotokoll
# --------------------------------------------------------------------------- #
@pytest.fixture()
def issued_license(bootstrapped, ls_app):
from tesm_core.extension import core
from tesm_license.services import issuance
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client, type="enterprise")
client.post(
f"/lizenzen/aus-ticket/{ticket_id}",
data={"csrf_token": _token(client, f"/tickets/{ticket_id}")},
follow_redirects=True,
)
extension = core(ls_app)
with ls_app.app_context(), extension.database.session() as conn:
row = issuance.list_licenses(conn)[0]
bundle = issuance.delivery_bundle(
conn, str(row["license_id"]), keystore=extension.keystore
)
document, client_key = lic.split_bundle(bundle)
return {"document": document, "client_key": client_key, "row": row}
def test_activation_round_trip_over_api(ls_client, issued_license):
document = issued_license["document"]
request = lic.build_client_request(
"activate", document, issued_license["client_key"], hostname="kunde-01"
)
response = ls_client.post("/api/v1/licenses/activate", json=request)
assert response.status_code == 200, response.get_json()
payload = response.get_json()
lic.verify_master_response(
payload,
document["master_pubkey"],
expected_license_id=document["license_id"],
expected_fingerprint=request["fingerprint"],
expected_action="activate",
expected_nonce=request["nonce"],
)
assert payload["status"] == "ok"
def test_replayed_activation_is_rejected(ls_client, issued_license):
document = issued_license["document"]
request = lic.build_client_request("activate", document, issued_license["client_key"])
assert ls_client.post("/api/v1/licenses/activate", json=request).status_code == 200
replay = ls_client.post("/api/v1/licenses/activate", json=request)
assert replay.status_code == 409
assert replay.get_json()["error"] == "replay"
def test_forged_signature_rejected(ls_client, issued_license):
document = issued_license["document"]
other_private, _ = lic.generate_keypair()
request = lic.build_client_request("activate", document, other_private)
response = ls_client.post("/api/v1/licenses/activate", json=request)
assert response.status_code == 400
assert response.get_json()["error"] == "invalid_signature"
def test_activation_bound_to_fingerprint(ls_client, ls_app, issued_license):
from tesm_core.extension import core
document = issued_license["document"]
request = lic.build_client_request("activate", document, issued_license["client_key"])
ls_client.post("/api/v1/licenses/activate", json=request)
# Zweite Maschine: anderer Fingerprint, gueltige Signatur.
second = lic.build_client_request("activate", document, issued_license["client_key"])
second["fingerprint"] = "b" * 64
second["signature"] = lic.sign(
{k: v for k, v in second.items() if k != "signature"},
issued_license["client_key"],
lic.SIG_CONTEXT_REQUEST,
)
response = ls_client.post("/api/v1/licenses/activate", json=second)
assert response.status_code == 409
assert response.get_json()["error"] == "already_bound_elsewhere"
assert core(ls_app) is not None
def test_heartbeat_requires_active_license(ls_client, issued_license):
document = issued_license["document"]
request = lic.build_client_request("heartbeat", document, issued_license["client_key"])
response = ls_client.post("/api/v1/licenses/heartbeat", json=request)
assert response.status_code == 409
assert response.get_json()["error"] == "not_active"
def test_heartbeat_after_activation_delivers_update(ls_client, ls_app, issued_license):
from tesm_core.extension import core
from tesm_license.services import issuance
from tesm_license.services import master_key
document = issued_license["document"]
activate = lic.build_client_request("activate", document, issued_license["client_key"])
ls_client.post("/api/v1/licenses/activate", json=activate)
# Der Anbieter erweitert die Lizenz nachtraeglich.
extension = core(ls_app)
with ls_app.app_context(), extension.database.session() as conn:
key = master_key.ensure(extension.registry["master_key_path"])
with extension.database.transaction(conn):
issuance.resign_license(
conn,
document["license_id"],
master_private_key=key.private_key,
changes={"modules": ["dhcp"]},
)
beat = lic.build_client_request("heartbeat", document, issued_license["client_key"])
response = ls_client.post("/api/v1/licenses/heartbeat", json=beat)
assert response.status_code == 200
payload = response.get_json()
assert payload["status"] == "ok"
assert payload["license_update"]["seq"] == document["seq"] + 1
lic.verify_license(payload["license_update"])
def test_revoked_license_reported_signed(ls_client, ls_app, issued_license):
from tesm_core.extension import core
from tesm_license.services import issuance
document = issued_license["document"]
activate = lic.build_client_request("activate", document, issued_license["client_key"])
ls_client.post("/api/v1/licenses/activate", json=activate)
extension = core(ls_app)
with ls_app.app_context(), extension.database.session() as conn:
with extension.database.transaction(conn):
issuance.revoke_license(conn, document["license_id"], reason="Test")
beat = lic.build_client_request("heartbeat", document, issued_license["client_key"])
response = ls_client.post("/api/v1/licenses/heartbeat", json=beat)
assert response.status_code == 200
payload = response.get_json()
assert payload["status"] == "revoked"
# Die Antwort ist signiert -- der Client loescht erst nach eigener Pruefung.
lic.verify_master_response(
payload,
document["master_pubkey"],
expected_license_id=document["license_id"],
expected_fingerprint=beat["fingerprint"],
expected_action="heartbeat",
expected_nonce=beat["nonce"],
)
def test_offline_code_path_matches_api(bootstrapped, issued_license):
"""Der Offline-Weg laeuft durch dieselbe Verarbeitung wie die API."""
document = issued_license["document"]
request = lic.build_client_request("activate", document, issued_license["client_key"])
code = lic.encode_code(request)
response = bootstrapped.post(
"/lizenzen/offline-code",
data={"csrf_token": _token(bootstrapped, "/lizenzen/offline-code"), "code": code},
follow_redirects=True,
)
body = response.get_data(as_text=True)
assert "Antwortcode" in body
start = body.index('id="response-code">') + len('id="response-code">')
answer = body[start : body.index("</pre>", start)].strip()
decoded = lic.decode_code(answer)
lic.verify_master_response(
decoded,
document["master_pubkey"],
expected_license_id=document["license_id"],
expected_fingerprint=request["fingerprint"],
expected_action="activate",
expected_nonce=request["nonce"],
)
assert decoded["status"] == "ok"
def test_unknown_license_rejected(ls_client, issued_license):
document = dict(issued_license["document"])
document["license_id"] = "gibt-es-nicht"
request = lic.build_client_request("activate", document, issued_license["client_key"])
response = ls_client.post("/api/v1/licenses/activate", json=request)
assert response.status_code == 404
# --------------------------------------------------------------------------- #
# Self-Service und Portal
# --------------------------------------------------------------------------- #
def test_self_service_page_and_issue(bootstrapped, ls_app):
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client)
fresh = ls_app.test_client() # ohne Anmeldung
page = fresh.get(f"/lizenz/{ticket_id}")
assert page.status_code == 200
assert "Lizenz erstellen" in page.get_data(as_text=True)
token = _token(fresh, f"/lizenz/{ticket_id}")
created = fresh.post(
f"/lizenz/{ticket_id}/ausstellen", data={"csrf_token": token}, follow_redirects=True
)
assert "Lizenz wurde erstellt" in created.get_data(as_text=True)
def test_self_service_404_for_license_server_ticket(bootstrapped, ls_app):
from tesm_core.extension import core
from tesm_license.services import issuance
client = bootstrapped
_create_customer(client)
ticket_id = _create_ticket(client)
extension = core(ls_app)
with extension.database.session() as conn, extension.database.transaction(conn):
conn.execute(
"UPDATE license_tickets SET role='license_server' WHERE id=?", (ticket_id,)
)
fresh = ls_app.test_client()
assert fresh.get(f"/lizenz/{ticket_id}").status_code == 404
def test_portal_answer_is_identical_for_unknown_email(ls_client):
token = _token(ls_client, "/portal/")
known = ls_client.post(
"/portal/", data={"csrf_token": token, "email": "gibt@es.nicht"}, follow_redirects=True
).get_data(as_text=True)
assert "Falls diese E-Mail-Adresse bei uns hinterlegt ist" in known
def test_portal_token_is_single_use(bootstrapped, ls_app):
from tesm_core.extension import core
client = bootstrapped
_create_customer(client)
# Token direkt anlegen (Mailversand ist im Test nicht konfiguriert).
extension = core(ls_app)
with extension.database.session() as conn, extension.database.transaction(conn):
conn.execute(
"INSERT INTO portal_access_tokens (token, customer_id, created_at, expires_at) "
"VALUES ('testtoken', 1, datetime('now'), datetime('now', '+30 minutes'))"
)
fresh = ls_app.test_client()
first = fresh.get("/portal/zugang/testtoken", follow_redirects=True)
assert "Ihre Lizenzen" in first.get_data(as_text=True)
second = ls_app.test_client()
again = second.get("/portal/zugang/testtoken", follow_redirects=True)
assert "ungueltig oder abgelaufen" in again.get_data(as_text=True)
def test_portal_expired_token_rejected(bootstrapped, ls_app):
from tesm_core.extension import core
_create_customer(bootstrapped)
extension = core(ls_app)
with extension.database.session() as conn, extension.database.transaction(conn):
conn.execute(
"INSERT INTO portal_access_tokens (token, customer_id, created_at, expires_at) "
"VALUES ('alt', 1, datetime('now','-2 hours'), datetime('now','-1 hours'))"
)
fresh = ls_app.test_client()
response = fresh.get("/portal/zugang/alt", follow_redirects=True)
assert "ungueltig oder abgelaufen" in response.get_data(as_text=True)
# --------------------------------------------------------------------------- #
# Ampel
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize(
"days,expected",
[(120, "green"), (91, "green"), (90, "orange"), (31, "orange"), (29, "red"), (-1, "red")],
)
def test_traffic_light_boundaries(days, expected):
from datetime import timedelta
from tesm_license.services.issuance import display_state
now = lic.utcnow()
row = {"expires_at": lic.iso(now + timedelta(days=days, hours=1)), "lifetime": 0}
assert display_state(row, now=now)["color"] == expected
def test_lifetime_is_always_green():
from tesm_license.services.issuance import display_state
state = display_state({"expires_at": "2000-01-01T00:00:00Z", "lifetime": 1})
assert state["color"] == "green"
assert state["is_expired"] is False
def test_unparsable_date_fails_safe():
from tesm_license.services.issuance import display_state
assert display_state({"expires_at": "kaputt", "lifetime": 0})["color"] == "red"
# --------------------------------------------------------------------------- #
# Endpunkt: ohne ihn keine Lizenz
# --------------------------------------------------------------------------- #
def test_issuing_without_an_endpoint_is_refused():
"""Der Endpunkt wird mitsigniert -- fehlt er, ist die Lizenz dauerhaft offline.
Nachtraeglich laesst er sich nicht eintragen: er steht in der signierten
Nutzlast. Eine Warnung waere zu spaet, denn der Schaden entsteht in dem
Moment, in dem die Datei den Server verlaesst.
"""
import pytest as _pytest
from tesm_license.services import issuance
assert "master_endpoint" in issuance.issue_from_ticket.__code__.co_varnames
source = __import__("inspect").getsource(issuance.issue_from_ticket)
assert 'if not str(master_endpoint or "").strip():' in source
assert "IssuanceError" in source
assert _pytest is not None
def test_offline_activation_has_a_navigation_entry():
"""Ohne Eintrag findet niemand die Seite -- der Offline-Weg waere unbenutzbar."""
from tesm_license.navigation import NAVIGATION
endpoints = set()
def walk(items):
for item in items:
if item.endpoint:
endpoints.add(item.endpoint)
if item.children:
walk(item.children)
walk(NAVIGATION.items)
assert "licenses.manual_code" in endpoints
def test_revision_pending_survives_a_partial_row():
"""Die Funktion sitzt in Listen, Detailseiten und im Portal.
Nicht jede Abfrage waehlt jede Spalte. Eine fehlende darf keine Ausnahme
werfen -- sonst faellt eine ganze Seite wegen einer Nebensaechlichkeit aus.
"""
from tesm_license.services.issuance import revision_pending
assert revision_pending({"revision_at": "2026-01-02", "status": "active"}) is True
assert revision_pending({"revision_at": None}) is False
assert revision_pending({}) is False
assert revision_pending({"revision_at": "2026-01-02", "status": "revoked"}) is False
# Abgeholt: der Kontakt ist juenger als die Revision.
assert (
revision_pending(
{
"revision_at": "2026-01-02T00:00:00+00:00",
"status": "active",
"last_heartbeat_at": "2026-01-03T00:00:00+00:00",
}
)
is False
)
# Noch nicht abgeholt.
assert (
revision_pending(
{
"revision_at": "2026-01-04T00:00:00+00:00",
"status": "active",
"last_heartbeat_at": "2026-01-03T00:00:00+00:00",
}
)
is True
)
+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)
+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}"
+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"