Files
Key-service-Server/Website/main.py
T
2026-08-28 00:18:26 +02:00

3563 lines
128 KiB
Python

from flask import Flask, render_template, request, jsonify, flash, redirect, url_for, get_flashed_messages, session, send_file, after_this_request, abort, send_from_directory, current_app
import os
import json
import atexit
import calendar
import re
import subprocess
import hashlib
import shutil
import tarfile
import tempfile
import threading
import unicodedata
import zipfile
from datetime import timedelta, datetime, date
from functools import wraps
from io import BytesIO
from urllib.parse import parse_qs, quote, urlparse
from werkzeug.security import generate_password_hash, check_password_hash
from werkzeug.utils import secure_filename
import bleach
from markupsafe import escape
from pymongo import MongoClient
from pymongo.errors import PyMongoError
from bson.objectid import ObjectId
import user as user_store
import buchungen
import server_steering as steering
from modules.emailservice.email import send_register_token, send_accreditation_email
import secrets
import requests
from server_steering import Instance
from datetime import timezone, datetime
import string
import traceback
app = Flask(__name__)
app.secret_key = "ASDfhbsdfseiufhgildsrfrjg874368546987s6e8468f4!?FAUS/&s"
app.config["SESSION_COOKIE_HTTPONLY"] = True
app.config["SESSION_COOKIE_SAMESITE"] = "Strict"
app.config["SESSION_COOKIE_SECURE"] = os.environ.get("SESSION_COOKIE_SECURE", "0") == "1"
app.config["PREFERRED_URL_SCHEME"] = "https" if os.environ.get("SESSION_COOKIE_SECURE") == "1" else "http"
BOOKING_PACKAGE_MAP = {
"startpaket": "Startpaket (Inventar & Bücherei)",
"advanced_package": "Advanced-Paket",
"inventarsystem": "Inventarsystem",
"buecherei": "Bücherei",
"terminverwaltung": "Terminverwaltung",
"emailversand": "E-Mail Versand",
"schultraeger": "Schulträger Angebot",
}
BOOKING_FLOW_MAP = {
"consultation": {
"title": "Buchung als Gesprächstermin prüfen",
"intro": "Bitte prüfen Sie die Angaben noch einmal. Nach Ihrer Bestätigung wird die Anfrage als Gesprächstermin an das Team gesendet.",
"confirm_label": "Gesprächstermin bestätigen",
"message_prefix": "Ich möchte für das Paket",
"message_suffix": "einen Gesprächstermin vereinbaren. Bitte kontaktieren Sie mich für die nächsten Schritte.",
},
"payment": {
"title": "Buchung vor dem Zahlungsschritt prüfen",
"intro": "Bitte prüfen Sie die Angaben noch einmal. Sie bekommen bei bestätigung ihre Rechnung per Email versendet.",
"confirm_label": "Jetzt kostenpflichtig buchen",
"message_prefix": "Ich möchte das Paket",
"message_suffix": "kostenpflichtig buchen. Bitte prüfen Sie die Angaben und senden Sie mir die nächsten Schritte.",
},
}
BOOKING_PRICE_MAP = {
"startpaket": 420.0,
"advanced_package": 0.0,
"inventarsystem": 350.0,
"buecherei": 100.0,
"terminverwaltung": 100.0,
"emailversand": 50.0,
"schultraeger": 0.0,
}
BOOKING_ADMIN_EMAIL_FALLBACK = os.environ.get("BOOKING_ADMIN_EMAIL", "").strip()
@app.after_request
def set_security_headers(response):
response.headers["X-Content-Type-Options"] = "nosniff"
response.headers["X-Frame-Options"] = "SAMEORIGIN"
response.headers["X-XSS-Protection"] = "1; mode=block"
response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin"
# Fonts no longer loaded from Google Fonts; remove font hosts
response.headers["Content-Security-Policy"] = "default-src 'self'; script-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; style-src 'self' 'unsafe-inline'; font-src 'self'; img-src 'self' data:; connect-src 'self'; frame-src 'self' https://www.youtube.com https://youtube.com https://www.youtube-nocookie.com https://youtube-nocookie.com; media-src 'self' https:;"
return response
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
INVOICE_UPLOAD_DIR = os.path.join(BASE_DIR, "static", "uploads", "invoices")
TUTORIAL_UPLOAD_DIR = os.path.join(BASE_DIR, "static", "uploads", "tutorials")
LEGAL_DOWNLOAD_DIR = os.path.join(BASE_DIR, "static", "downloads", "legal")
LEGAL_DOWNLOAD_FILES = (
(
"nutzungsbedingungen.pdf",
"Nutzungsbedingungen",
[
"Invario UG Modul Suite",
"PDF-Export der Nutzungsbedingungen",
"",
"Die aktuelle HTML-Version finden Sie unter /nutzungsbedingungen.",
"Dieser Download wird vom Server als PDF bereitgestellt.",
],
),
(
"datenschutz.pdf",
"Datenschutzerklaerung",
[
"Invario UG Modul Suite",
"PDF-Export der Datenschutzerklaerung",
"",
"Die aktuelle HTML-Version finden Sie unter /datenschutz.",
"Dieser Download wird vom Server als PDF bereitgestellt.",
],
),
)
MONGO_URI = os.environ.get("MONGO_URI", "mongodb://localhost:27017")
MONGO_DB_NAME = os.environ.get("MONGO_DB_NAME", "Invario_Website")
INSTANCE_REPO_URL = os.environ.get("INSTANCE_REPO_URL", "https://git.invario-software.eu/Invario/Inventarsystem")
INSTANCE_PARENT_DOMAIN = os.environ.get("INSTANCE_PARENT_DOMAIN", "invario-software.de")
INSTANCE_BASE_DIR = os.environ.get("INSTANCE_BASE_DIR", "/opt/inventarsystem-instances")
INSTANCE_TLS_MODE = os.environ.get("INSTANCE_TLS_MODE", "development")
INSTANCE_VERSION_OPTIONS = os.environ.get("INSTANCE_VERSION_OPTIONS", "latest,v0.3.1")
INSTANCE_WILDCARD_CERT_FILE = os.environ.get(
"INSTANCE_WILDCARD_CERT_FILE",
"/etc/nginx/certs/wildcard.meine-domain.crt",
)
INSTANCE_WILDCARD_KEY_FILE = os.environ.get(
"INSTANCE_WILDCARD_KEY_FILE",
"/etc/nginx/certs/wildcard.meine-domain.key",
)
INSTANCE_PROVISION_SCRIPT = os.environ.get(
"INSTANCE_PROVISION_SCRIPT",
os.path.abspath(os.path.join(BASE_DIR, "provision_instance.sh")),
)
TRIGGER_DIR = "/app/triggers"
def _env_int(name: str, default: int) -> int:
value = os.environ.get(name)
if value is None:
return default
try:
return int(value)
except (TypeError, ValueError):
return default
MONGO_MAX_POOL_SIZE = max(_env_int("MONGO_MAX_POOL_SIZE", 12), 1)
MONGO_MIN_POOL_SIZE = max(_env_int("MONGO_MIN_POOL_SIZE", 0), 0)
MONGO_MAX_IDLE_MS = max(_env_int("MONGO_MAX_IDLE_MS", 60000), 1000)
MONGO_CONNECT_TIMEOUT_MS = max(_env_int("MONGO_CONNECT_TIMEOUT_MS", 1500), 500)
MONGO_SOCKET_TIMEOUT_MS = max(_env_int("MONGO_SOCKET_TIMEOUT_MS", 30000), 1000)
MONGO_WAIT_QUEUE_TIMEOUT_MS = max(_env_int("MONGO_WAIT_QUEUE_TIMEOUT_MS", 2000), 500)
_MONGO_CLIENT: MongoClient | None = None
_MONGO_LOCK = threading.Lock()
class _NoopMongoClientHandle:
def close(self):
# Backward-compatible no-op so existing finally blocks stay harmless.
return None
def _utc_now_iso() -> str:
return datetime.utcnow().isoformat(timespec="seconds") + "Z"
def _is_allowed_invoice_filename(filename: str) -> bool:
return bool(filename) and filename.lower().endswith(".pdf")
def _is_allowed_image_filename(filename: str) -> bool:
if not filename:
return False
lowered = filename.lower()
return lowered.endswith(".jpg") or lowered.endswith(".jpeg") or lowered.endswith(".png") or lowered.endswith(".webp")
def _is_allowed_tutorial_video_filename(filename: str) -> bool:
if not filename:
return False
lowered = filename.lower()
return lowered.endswith(".mp4") or lowered.endswith(".webm") or lowered.endswith(".ogv") or lowered.endswith(".ogg") or lowered.endswith(".m4v")
def _pdf_safe_text(value: str) -> str:
text = unicodedata.normalize("NFKD", value or "")
text = text.encode("ascii", "ignore").decode("ascii")
return text.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
def _build_simple_pdf_bytes(title: str, lines: list[str]) -> bytes:
buffer = BytesIO()
def write(text: str) -> None:
buffer.write(text.encode("latin-1"))
content_lines = ["BT", "/F1 20 Tf", "72 760 Td", f"({_pdf_safe_text(title)}) Tj", "/F1 11 Tf"]
y_position = 730
for line in lines:
if not line:
y_position -= 14
continue
content_lines.append(f"1 0 0 1 72 {y_position} Tm")
content_lines.append(f"({_pdf_safe_text(line)}) Tj")
y_position -= 16
content_lines.append("ET")
content = "\n".join(content_lines).encode("latin-1")
write("%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
offsets = [0]
def write_object(object_number: int, body: str) -> None:
offsets.append(buffer.tell())
write(f"{object_number} 0 obj\n{body}\nendobj\n")
write_object(1, "<< /Type /Catalog /Pages 2 0 R >>")
write_object(2, "<< /Type /Pages /Kids [3 0 R] /Count 1 >>")
write_object(
3,
"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",
)
write_object(4, "<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>")
offsets.append(buffer.tell())
write(f"5 0 obj\n<< /Length {len(content)} >>\nstream\n")
buffer.write(content)
write("\nendstream\nendobj\n")
xref_start = buffer.tell()
write("xref\n0 6\n")
write("0000000000 65535 f \n")
for offset in offsets[1:]:
write(f"{offset:010d} 00000 n \n")
write("trailer\n<< /Size 6 /Root 1 0 R >>\nstartxref\n")
write(f"{xref_start}\n%%EOF\n")
return buffer.getvalue()
def _ensure_legal_pdf_files() -> None:
os.makedirs(LEGAL_DOWNLOAD_DIR, exist_ok=True)
for filename, title, lines in LEGAL_DOWNLOAD_FILES:
target_path = os.path.join(LEGAL_DOWNLOAD_DIR, filename)
if os.path.exists(target_path):
continue
with open(target_path, "wb") as pdf_file:
pdf_file.write(_build_simple_pdf_bytes(title, lines))
def _build_legal_pdf_bundle() -> BytesIO:
_ensure_legal_pdf_files()
bundle = BytesIO()
with zipfile.ZipFile(bundle, mode="w", compression=zipfile.ZIP_DEFLATED) as archive:
for filename, _, _ in LEGAL_DOWNLOAD_FILES:
archive.write(os.path.join(LEGAL_DOWNLOAD_DIR, filename), arcname=filename)
bundle.seek(0)
return bundle
def _save_invoice_pdf(file_obj, invoice_number: str) -> str | None:
if not file_obj or not file_obj.filename:
return None
original_name = secure_filename(file_obj.filename)
if not _is_allowed_invoice_filename(original_name):
return None
os.makedirs(INVOICE_UPLOAD_DIR, exist_ok=True)
safe_invoice = secure_filename(invoice_number or "invoice")
unique_name = f"{datetime.utcnow().strftime('%Y%m%d%H%M%S')}_{safe_invoice}_{original_name}"
target = os.path.join(INVOICE_UPLOAD_DIR, unique_name)
file_obj.save(target)
return f"uploads/invoices/{unique_name}"
def _save_tutorial_video_file(file_obj, title: str) -> str | None:
if not file_obj or not file_obj.filename:
return None
original_name = secure_filename(file_obj.filename)
if not _is_allowed_tutorial_video_filename(original_name):
return None
os.makedirs(TUTORIAL_UPLOAD_DIR, exist_ok=True)
safe_title = secure_filename(title or "tutorial")
unique_name = f"{datetime.utcnow().strftime('%Y%m%d%H%M%S')}_{safe_title}_{original_name}"
target = os.path.join(TUTORIAL_UPLOAD_DIR, unique_name)
file_obj.save(target)
return f"/static/uploads/tutorials/{unique_name}"
def _get_mongo_client() -> MongoClient:
global _MONGO_CLIENT
if _MONGO_CLIENT is not None:
return _MONGO_CLIENT
with _MONGO_LOCK:
if _MONGO_CLIENT is None:
_MONGO_CLIENT = MongoClient(
MONGO_URI,
serverSelectionTimeoutMS=MONGO_CONNECT_TIMEOUT_MS,
connectTimeoutMS=MONGO_CONNECT_TIMEOUT_MS,
socketTimeoutMS=MONGO_SOCKET_TIMEOUT_MS,
maxPoolSize=MONGO_MAX_POOL_SIZE,
minPoolSize=MONGO_MIN_POOL_SIZE,
maxIdleTimeMS=MONGO_MAX_IDLE_MS,
waitQueueTimeoutMS=MONGO_WAIT_QUEUE_TIMEOUT_MS,
)
return _MONGO_CLIENT
@atexit.register
def _shutdown_mongo_client() -> None:
global _MONGO_CLIENT
client = _MONGO_CLIENT
_MONGO_CLIENT = None
if client is not None:
client.close()
def _get_mongo_db():
client = _get_mongo_client()
return _NoopMongoClientHandle(), client[MONGO_DB_NAME]
def _normalize_user_doc(doc: dict | None) -> dict | None:
if not doc:
return None
return {
"username": doc.get("Username") or doc.get("username"),
"display_name": doc.get("name") or doc.get("display_name") or doc.get("Username"),
"is_admin": bool(doc.get("Admin") or doc.get("is_admin", False)),
"email": doc.get("email") or doc.get("Email") or "",
"created_at": doc.get("created_at"),
"verified": doc.get("verified", True),
"verification_token": doc.get("verification_token"),
}
def _get_collection(name: str):
client, db = _get_mongo_db()
return client, db[name]
def _list_users_for_admin() -> list:
docs = user_store.get_all_users() or []
users = []
for doc in docs:
normalized = _normalize_user_doc(doc)
if normalized and normalized.get("username"):
users.append(normalized)
users.sort(key=lambda item: (item.get("is_admin", False), item.get("username", "")), reverse=True)
return users
def _list_admin_users() -> list:
return [user for user in _list_users_for_admin() if user.get("is_admin")]
def _select_booking_admin() -> dict | None:
admins = _list_admin_users()
if admins:
return admins[0]
return None
def _sanitize_text(text: str, max_length: int = 255) -> str:
"""Sanitize user text input: strip, limit length, and escape HTML."""
text = (text or "").strip()
if len(text) > max_length:
text = text[:max_length]
return text
def _normalize_tutorial_video_url(url: str, origin: str = "") -> str:
"""Normalize user-provided tutorial URLs to safe embed URLs where possible."""
raw = _sanitize_text(url, 800)
if not raw:
return ""
if raw.startswith("/"):
return raw
try:
parsed = urlparse(raw)
except Exception:
return ""
host = (parsed.netloc or "").lower()
path = parsed.path or ""
video_id = ""
if host in {"youtu.be", "www.youtu.be"}:
video_id = path.strip("/").split("/")[0]
elif "youtube.com" in host:
if path.startswith("/watch"):
video_id = (parse_qs(parsed.query).get("v") or [""])[0]
elif path.startswith("/embed/"):
video_id = path.split("/embed/", 1)[1].split("/")[0]
elif path.startswith("/shorts/"):
video_id = path.split("/shorts/", 1)[1].split("/")[0]
elif "youtube-nocookie.com" in host and path.startswith("/embed/"):
video_id = path.split("/embed/", 1)[1].split("/")[0]
if video_id and re.fullmatch(r"[A-Za-z0-9_-]{6,20}", video_id):
origin_value = quote((origin or "").rstrip("/"), safe=":/")
if origin_value:
return (
f"https://www.youtube-nocookie.com/embed/{video_id}"
f"?rel=0&modestbranding=1&playsinline=1&origin={origin_value}"
)
return f"https://www.youtube-nocookie.com/embed/{video_id}?rel=0&modestbranding=1&playsinline=1"
if parsed.scheme in {"http", "https"} and host:
return raw
return ""
def _slugify_subdomain(value: str) -> str:
cleaned = (value or "").strip().lower()
cleaned = cleaned.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss")
cleaned = re.sub(r"[^a-z0-9-]+", "-", cleaned)
cleaned = re.sub(r"-{2,}", "-", cleaned).strip("-")
return cleaned[:63]
def _is_valid_subdomain(value: str) -> bool:
if not value or len(value) < 3 or len(value) > 63:
return False
return bool(re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{1,61}[a-z0-9])", value))
def _parse_key_value_output(raw: str) -> dict:
parsed = {}
for line in (raw or "").splitlines():
row = line.strip()
if not row or "=" not in row:
continue
key, value = row.split("=", 1)
parsed[key.strip()] = value.strip()
return parsed
def _parse_instance_version_options() -> list[str]:
values = []
for item in (INSTANCE_VERSION_OPTIONS or "").split(","):
token = _sanitize_text(item or "", 80)
if token and token not in values:
values.append(token)
if not values:
values = ["latest"]
return values
def _list_school_instances() -> list:
rows = []
client = None
try:
# Greife auf "instances" zu (passend zum Async-Provisioning)
client, col = _get_collection("instances")
rows = list(col.find().sort([("updated_at", -1), ("created_at", -1)]))
except PyMongoError:
return []
finally:
if client:
client.close()
normalized = []
for row in rows:
normalized.append(
{
"id": str(row.get("_id") or ""),
"school_name": _sanitize_text(row.get("school_name") or "", 120),
"owner_username": _sanitize_text(row.get("owner_username") or row.get("username") or "", 80),
"subdomain": _sanitize_text(row.get("subdomain") or "", 63),
"domain": _sanitize_text(row.get("domain") or "", 190),
"https_port": int(row.get("https_port") or row.get("port") or 0),
"instance_dir": _sanitize_text(row.get("instance_dir") or "", 300),
"app_image_tag": _sanitize_text(row.get("app_image_tag") or "latest", 80),
"library_enabled": bool(row.get("library_enabled", False)),
"status": _sanitize_text(row.get("status") or "Unbekannt", 40),
"nginx_status": _sanitize_text(row.get("nginx_status") or "unbekannt", 80),
"last_message": _sanitize_text(row.get("last_message") or "", 500),
"updated_at": row.get("updated_at") or row.get("created_at") or "",
}
)
return normalized
def _list_available_instance_users() -> list:
users = _list_users_for_admin()
instances = _list_school_instances()
assigned = {
_sanitize_text(item.get("owner_username") or "", 80).lower()
for item in instances
if _sanitize_text(item.get("owner_username") or "", 80)
}
return [u for u in users if _sanitize_text(u.get("username") or "", 80).lower() not in assigned]
def _list_instances_grouped_by_owner() -> dict[str, list[dict]]:
grouped: dict[str, list[dict]] = {}
for item in _list_school_instances():
owner = _sanitize_text(item.get("owner_username") or "", 80)
domain = _sanitize_text(item.get("domain") or "", 190)
subdomain = _sanitize_text(item.get("subdomain") or "", 63)
if not owner or not domain:
continue
grouped.setdefault(owner, []).append(
{
"domain": domain,
"subdomain": subdomain,
"status": _sanitize_text(item.get("status") or "", 40),
}
)
return grouped
def _parse_iso_timestamp(value: str) -> datetime | None:
raw = (value or "").strip()
if not raw:
return None
try:
return datetime.fromisoformat(raw.replace("Z", "+00:00"))
except Exception:
return None
def _build_instance_dashboard(instances: list[dict]) -> dict:
total = len(instances)
running = len([item for item in instances if _sanitize_text(item.get("status") or "", 40) == "Läuft"])
error = len(
[
item
for item in instances
if _sanitize_text(item.get("status") or "", 40) == "Fehler"
or _sanitize_text(item.get("nginx_status") or "", 80).lower() == "error"
]
)
library_on = len([item for item in instances if bool(item.get("library_enabled"))])
assigned_users = len(
{
_sanitize_text(item.get("owner_username") or "", 80)
for item in instances
if _sanitize_text(item.get("owner_username") or "", 80)
}
)
version_counts: dict[str, int] = {}
owner_counts: dict[str, int] = {}
for item in instances:
version = _sanitize_text(item.get("app_image_tag") or "latest", 80) or "latest"
owner = _sanitize_text(item.get("owner_username") or "Unzugewiesen", 80) or "Unzugewiesen"
version_counts[version] = version_counts.get(version, 0) + 1
owner_counts[owner] = owner_counts.get(owner, 0) + 1
version_items = sorted(version_counts.items(), key=lambda row: row[1], reverse=True)
owner_items = sorted(owner_counts.items(), key=lambda row: row[1], reverse=True)[:8]
day_keys = []
today = date.today()
for back in range(6, -1, -1):
day_keys.append((today - timedelta(days=back)).isoformat())
updates_by_day = {day: 0 for day in day_keys}
for item in instances:
parsed = _parse_iso_timestamp(_sanitize_text(item.get("updated_at") or "", 40))
if not parsed:
continue
day_key = parsed.date().isoformat()
if day_key in updates_by_day:
updates_by_day[day_key] += 1
labels = [datetime.fromisoformat(day).strftime("%d.%m") for day in day_keys]
values = [updates_by_day[day] for day in day_keys]
return {
"kpis": {
"total": total,
"running": running,
"error": error,
"library_on": library_on,
"assigned_users": assigned_users,
},
"status": {
"labels": ["Läuft", "Fehler", "Sonstige"],
"values": [running, error, max(total - running - error, 0)],
},
"library": {
"labels": ["Aktiv", "Inaktiv"],
"values": [library_on, max(total - library_on, 0)],
},
"versions": {
"labels": [item[0] for item in version_items],
"values": [item[1] for item in version_items],
},
"owners": {
"labels": [item[0] for item in owner_items],
"values": [item[1] for item in owner_items],
},
"activity": {
"labels": labels,
"values": values,
},
}
def _parse_size_to_mib(value: str) -> float:
text = (value or "").strip()
if not text:
return 0.0
match = re.match(r"^([0-9]*\.?[0-9]+)\s*([kmgt]?i?b)$", text.lower())
if not match:
return 0.0
number = float(match.group(1))
unit = match.group(2)
factor_map = {
"b": 1.0 / (1024.0 * 1024.0),
"kib": 1.0 / 1024.0,
"kb": 1.0 / 1024.0,
"mib": 1.0,
"mb": 1.0,
"gib": 1024.0,
"gb": 1024.0,
"tib": 1024.0 * 1024.0,
"tb": 1024.0 * 1024.0,
}
return number * factor_map.get(unit, 0.0)
def _read_meminfo_mib() -> tuple[float, float]:
total_kib = 0.0
available_kib = 0.0
try:
with open("/proc/meminfo", "r", encoding="utf-8") as handle:
for row in handle:
if row.startswith("MemTotal:"):
total_kib = float(row.split()[1])
elif row.startswith("MemAvailable:"):
available_kib = float(row.split()[1])
except Exception:
return 0.0, 0.0
return total_kib / 1024.0, available_kib / 1024.0
def _collect_runtime_stats(instances: list[dict]) -> dict:
ok, output = _run_command(
["docker", "stats", "--no-stream", "--format", "{{.Name}}|{{.CPUPerc}}|{{.MemUsage}}"],
timeout=120,
)
container_stats = {}
total_used_mib = 0.0
if ok:
for row in (output or "").splitlines():
line = row.strip()
if not line or "|" not in line:
continue
parts = line.split("|")
if len(parts) < 3:
continue
name = parts[0].strip()
cpu_raw = parts[1].strip().replace("%", "")
mem_raw = parts[2].strip()
used_raw = mem_raw.split("/")[0].strip() if "/" in mem_raw else mem_raw
try:
cpu_percent = float(cpu_raw) if cpu_raw else 0.0
except ValueError:
cpu_percent = 0.0
mem_mib = _parse_size_to_mib(used_raw)
total_used_mib += mem_mib
container_stats[name] = {
"cpu_percent": round(cpu_percent, 2),
"mem_mib": round(mem_mib, 2),
"mem_raw": mem_raw,
}
managed_prefixes = [f"{_sanitize_text(item.get('subdomain') or '', 63)}-" for item in instances if _sanitize_text(item.get('subdomain') or '', 63)]
managed_prefixes.append("website-")
managed_container_stats = {
name: stats
for name, stats in container_stats.items()
if any(name.startswith(prefix) for prefix in managed_prefixes)
}
managed_used_mib = round(sum(item.get("mem_mib", 0.0) for item in managed_container_stats.values()), 2)
per_instance = []
for item in instances:
subdomain = _sanitize_text(item.get("subdomain") or "", 63)
domain = _sanitize_text(item.get("domain") or "", 190)
if not subdomain:
continue
prefix = f"{subdomain}-"
related = []
for container_name, stats in container_stats.items():
if container_name.startswith(prefix):
related.append({"name": container_name, **stats})
mem_sum = round(sum(entry.get("mem_mib", 0.0) for entry in related), 2)
cpu_sum = round(sum(entry.get("cpu_percent", 0.0) for entry in related), 2)
per_instance.append(
{
"subdomain": subdomain,
"domain": domain,
"status": _sanitize_text(item.get("status") or "Unbekannt", 40),
"nginx_status": _sanitize_text(item.get("nginx_status") or "unbekannt", 80),
"containers": related,
"container_count": len(related),
"mem_mib": mem_sum,
"cpu_percent": cpu_sum,
}
)
per_instance.sort(key=lambda row: row.get("mem_mib", 0.0), reverse=True)
total_mem_mib, available_mem_mib = _read_meminfo_mib()
used_mem_mib = max(total_mem_mib - available_mem_mib, 0.0)
used_mem_pct = (used_mem_mib / total_mem_mib * 100.0) if total_mem_mib > 0 else 0.0
return {
"generated_at": _utc_now_iso(),
"host": {
"total_mem_mib": round(total_mem_mib, 2),
"available_mem_mib": round(available_mem_mib, 2),
"used_mem_mib": round(used_mem_mib, 2),
"used_mem_pct": round(used_mem_pct, 2),
},
"docker": {
"all_container_count": len(container_stats),
"all_used_mem_mib": round(total_used_mib, 2),
"managed_container_count": len(managed_container_stats),
"managed_used_mem_mib": managed_used_mib,
},
"instances": per_instance,
}
def _read_uptime_load() -> dict:
uptime_seconds = 0.0
load_1 = 0.0
load_5 = 0.0
load_15 = 0.0
try:
with open("/proc/uptime", "r", encoding="utf-8") as handle:
uptime_seconds = float((handle.read().strip().split() or ["0"])[0])
except Exception:
uptime_seconds = 0.0
try:
with open("/proc/loadavg", "r", encoding="utf-8") as handle:
parts = handle.read().strip().split()
load_1 = float(parts[0])
load_5 = float(parts[1])
load_15 = float(parts[2])
except Exception:
pass
return {
"uptime_seconds": round(uptime_seconds, 1),
"load_1": round(load_1, 2),
"load_5": round(load_5, 2),
"load_15": round(load_15, 2),
}
def _read_disk_usage(path: str) -> dict:
try:
usage = shutil.disk_usage(path)
except Exception:
return {"path": path, "total_gib": 0.0, "used_gib": 0.0, "free_gib": 0.0, "used_pct": 0.0}
total_gib = usage.total / (1024.0 ** 3)
used_gib = usage.used / (1024.0 ** 3)
free_gib = usage.free / (1024.0 ** 3)
used_pct = (used_gib / total_gib * 100.0) if total_gib > 0 else 0.0
return {
"path": path,
"total_gib": round(total_gib, 2),
"used_gib": round(used_gib, 2),
"free_gib": round(free_gib, 2),
"used_pct": round(used_pct, 2),
}
def _collect_ops_counts() -> dict:
appointments_pending = 0
tickets_open = 0
users_total = 0
client = None
try:
client, col = _get_collection("appointments")
appointments_pending = col.count_documents({"status": "Angefragt"})
except PyMongoError:
appointments_pending = 0
finally:
if client:
client.close()
client = None
try:
client, col = _get_collection("support_tickets")
tickets_open = col.count_documents({"status": {"$in": ["Offen", "In Bearbeitung"]}})
except PyMongoError:
tickets_open = 0
finally:
if client:
client.close()
try:
users_total = len(_list_users_for_admin())
except Exception:
users_total = 0
return {
"appointments_pending": appointments_pending,
"tickets_open": tickets_open,
"users_total": users_total,
}
def _build_server_management_snapshot(instances: list[dict]) -> dict:
runtime = _collect_runtime_stats(instances)
root_disk = _read_disk_usage("/")
instance_disk = _read_disk_usage(INSTANCE_BASE_DIR)
ops_counts = _collect_ops_counts()
uptime_load = _read_uptime_load()
return {
"generated_at": _utc_now_iso(),
"runtime": runtime,
"root_disk": root_disk,
"instance_disk": instance_disk,
"ops": ops_counts,
"system": uptime_load,
}
def _upsert_school_instance(data: dict) -> None:
subdomain = _sanitize_text(data.get("subdomain") or "", 63)
if not subdomain:
return
client = None
try:
client, col = _get_collection("school_instances")
existing = col.find_one({"subdomain": subdomain}) or {}
def _pick_text(field: str, max_len: int, fallback: str = "") -> str:
if field in data:
raw = data.get(field)
if raw is not None and (not isinstance(raw, str) or raw.strip()):
return _sanitize_text(raw, max_len)
return _sanitize_text(existing.get(field) or fallback, max_len)
def _pick_int(field: str, fallback: int = 0) -> int:
if field in data:
raw = data.get(field)
if raw not in (None, ""):
try:
return int(raw)
except (TypeError, ValueError):
pass
try:
return int(existing.get(field) or fallback)
except (TypeError, ValueError):
return fallback
if "library_enabled" in data:
library_enabled = bool(data.get("library_enabled", False))
else:
library_enabled = bool(existing.get("library_enabled", False))
payload = {
"school_name": _pick_text("school_name", 120),
"owner_username": _pick_text("owner_username", 80),
"subdomain": subdomain,
"domain": _pick_text("domain", 190),
"https_port": _pick_int("https_port", 0),
"instance_dir": _pick_text("instance_dir", 300),
"app_image_tag": _pick_text("app_image_tag", 80, "latest") or "latest",
"library_enabled": library_enabled,
"status": _pick_text("status", 40, "Unbekannt") or "Unbekannt",
"nginx_status": _pick_text("nginx_status", 80, "unbekannt") or "unbekannt",
"last_message": _pick_text("last_message", 500),
"updated_at": _utc_now_iso(),
}
col.update_one(
{"subdomain": subdomain},
{"$set": payload, "$setOnInsert": {"created_at": _utc_now_iso()}},
upsert=True,
)
except PyMongoError:
return
finally:
if client:
client.close()
def _run_instance_provision(
action: str,
school_name: str,
subdomain: str,
app_image_tag: str = "latest",
library_enabled: bool = False,
) -> tuple[bool, str, dict]:
script_path = (INSTANCE_PROVISION_SCRIPT or "").strip()
if not script_path:
return False, "INSTANCE_PROVISION_SCRIPT ist nicht gesetzt.", {}
if not os.path.isfile(script_path):
return False, f"Provisioning-Skript nicht gefunden: {script_path}", {}
if not os.access(script_path, os.X_OK):
return False, f"Provisioning-Skript ist nicht ausführbar: {script_path}", {}
command = [
script_path,
"--action",
action,
"--repo",
INSTANCE_REPO_URL,
"--base-dir",
INSTANCE_BASE_DIR,
"--domain",
INSTANCE_PARENT_DOMAIN,
"--tls-mode",
INSTANCE_TLS_MODE,
"--wildcard-cert-file",
INSTANCE_WILDCARD_CERT_FILE,
"--wildcard-key-file",
INSTANCE_WILDCARD_KEY_FILE,
"--app-image-tag",
_sanitize_text(app_image_tag or "latest", 80),
"--library-enabled",
"1" if library_enabled else "0",
"--school-name",
school_name,
"--subdomain",
subdomain,
]
try:
result = subprocess.run(
command,
capture_output=True,
text=True,
timeout=900,
check=False,
env={**os.environ, "LC_ALL": "C.UTF-8"},
)
except subprocess.TimeoutExpired:
return False, "Provisioning-Zeitlimit erreicht (15 Minuten).", {}
except Exception as exc:
return False, f"Provisioning konnte nicht gestartet werden: {exc}", {}
output = _parse_key_value_output(result.stdout)
message = output.get("MESSAGE") or output.get("ERROR") or (result.stderr or "").strip()
if result.returncode != 0:
if not message:
message = "Provisioning fehlgeschlagen. Details im Server-Log prüfen."
return False, message, output
return True, message or "Instanz erfolgreich gestartet.", output
def _instance_dir_path(subdomain: str) -> str | None:
key = _sanitize_text(subdomain or "", 63)
if not _is_valid_subdomain(key):
return None
target = os.path.abspath(os.path.join(INSTANCE_BASE_DIR, key))
base_abs = os.path.abspath(INSTANCE_BASE_DIR)
if not target.startswith(base_abs + os.sep):
return None
return target
def _resolve_instance_dir(subdomain: str) -> str | None:
key = _sanitize_text(subdomain or "", 63)
if not _is_valid_subdomain(key):
return None
target = os.path.abspath(os.path.join(INSTANCE_BASE_DIR, key))
base_abs = os.path.abspath(INSTANCE_BASE_DIR)
if not target.startswith(base_abs + os.sep):
return None
if not os.path.isdir(target):
return None
return target
def _run_command(command: list[str], cwd: str | None = None, timeout: int = 900) -> tuple[bool, str]:
try:
result = subprocess.run(
command,
cwd=cwd,
capture_output=True,
text=True,
timeout=timeout,
check=False,
env={**os.environ, "LC_ALL": "C.UTF-8"},
)
except subprocess.TimeoutExpired:
return False, "Befehl hat das Zeitlimit erreicht."
except Exception as exc:
return False, f"Befehl konnte nicht ausgeführt werden: {exc}"
output = ((result.stdout or "") + "\n" + (result.stderr or "")).strip()
if result.returncode != 0:
return False, output or "Befehl ist fehlgeschlagen."
return True, output or "OK"
def _instance_compose_file(instance_dir: str) -> str | None:
preferred = os.path.join(instance_dir, "docker-compose-multitenant.yml")
fallback = os.path.join(instance_dir, "docker-compose.yml")
if os.path.isfile(preferred):
return preferred
if os.path.isfile(fallback):
return fallback
return None
def _instance_compose_cmd(instance_dir: str, args: list[str]) -> list[str] | None:
compose_file = _instance_compose_file(instance_dir)
if not compose_file:
return None
cmd = ["docker", "compose", "-f", compose_file]
env_file = os.path.join(instance_dir, ".docker-build.env")
if os.path.isfile(env_file):
cmd.extend(["--env-file", ".docker-build.env"])
cmd.extend(args)
return cmd
def _collect_command_candidates(base_binaries: list[str], args: list[str]) -> list[list[str]]:
candidates: list[list[str]] = []
seen: set[tuple[str, ...]] = set()
has_sudo = shutil.which("sudo") is not None
for binary in base_binaries:
resolved = ""
if "/" in binary:
if os.path.isfile(binary) and os.access(binary, os.X_OK):
resolved = binary
else:
found = shutil.which(binary)
if found:
resolved = found
if not resolved:
continue
regular = tuple([resolved, *args])
if regular not in seen:
candidates.append(list(regular))
seen.add(regular)
if has_sudo:
privileged = tuple(["sudo", resolved, *args])
if privileged not in seen:
candidates.append(list(privileged))
seen.add(privileged)
return candidates
def _reload_host_nginx() -> tuple[bool, str]:
test_candidates = _collect_command_candidates(
["nginx", "/usr/sbin/nginx", "/usr/bin/nginx"],
["-t"],
)
if test_candidates:
test_errors: list[str] = []
test_ok = False
for command in test_candidates:
ok, output = _run_command(command, timeout=60)
if ok:
test_ok = True
break
test_errors.append(f"{' '.join(command)} -> {_tail_output(output, 3)}")
if not test_ok:
details = " | ".join(test_errors[-2:]) if test_errors else "Kein Detail vorhanden."
return False, f"Nginx-Konfigurationstest fehlgeschlagen. {details}"
reload_candidates: list[list[str]] = []
reload_candidates.extend(
_collect_command_candidates(
["nginx", "/usr/sbin/nginx", "/usr/bin/nginx"],
["-s", "reload"],
)
)
reload_candidates.extend(_collect_command_candidates(["systemctl"], ["reload", "nginx"]))
if not reload_candidates:
return False, "Kein Nginx-Reload-Befehl verfügbar (nginx/systemctl nicht gefunden)."
reload_errors: list[str] = []
for command in reload_candidates:
ok, output = _run_command(command, timeout=60)
if ok:
return True, f"Nginx neu geladen via: {' '.join(command)}"
reload_errors.append(f"{' '.join(command)} -> {_tail_output(output, 3)}")
details = " | ".join(reload_errors[-2:]) if reload_errors else "Kein Detail vorhanden."
return False, f"Nginx-Reload fehlgeschlagen. {details}"
def _host_reload_hint() -> str:
return "Bitte auf dem Host ausführen: sudo nginx -t && sudo systemctl reload nginx"
def _promote_manual_nginx_status(reload_message: str) -> int:
client = None
try:
client, col = _get_collection("school_instances")
result = col.update_many(
{"nginx_status": "manual_required"},
{
"$set": {
"nginx_status": "ok",
"last_message": _sanitize_text(reload_message, 500),
"updated_at": _utc_now_iso(),
}
},
)
return int(result.modified_count or 0)
except PyMongoError:
return 0
finally:
if client:
client.close()
def _tail_output(text: str, lines: int = 24) -> str:
rows = [line for line in (text or "").splitlines() if line.strip()]
if not rows:
return "Keine Ausgabe"
return "\n".join(rows[-lines:])
def _collect_core_logs() -> tuple[bool, str]:
compose_file = os.path.join(BASE_DIR, "docker-compose.yml")
command = [
"docker",
"compose",
"-f",
compose_file,
"logs",
"--no-color",
"--tail",
"500",
"website",
"mongodb",
]
return _run_command(command, cwd=BASE_DIR, timeout=180)
def _truncate_log_blob(text: str, max_lines: int = 220, max_chars: int = 32000) -> str:
rows = (text or "").splitlines()
if len(rows) > max_lines:
rows = rows[-max_lines:]
clipped = "\n".join(rows).strip()
if len(clipped) > max_chars:
clipped = clipped[-max_chars:]
return clipped or "Keine Ausgabe"
def _run_first_success(candidates: list[list[str]], cwd: str | None = None, timeout: int = 120) -> tuple[bool, str, str]:
if not candidates:
return False, "Kein Befehl verfügbar.", ""
failures: list[str] = []
for command in candidates:
ok, output = _run_command(command, cwd=cwd, timeout=timeout)
if ok:
return True, output, " ".join(command)
failures.append(f"{' '.join(command)} -> {_tail_output(output, 3)}")
return False, " | ".join(failures[-2:]) if failures else "Kein Detail vorhanden.", ""
def _collect_core_live_logs(lines: int = 220) -> dict:
compose_file = os.path.join(BASE_DIR, "docker-compose.yml")
command = [
"docker",
"compose",
"-f",
compose_file,
"logs",
"--no-color",
"--tail",
str(max(lines, 20)),
"website",
"mongodb",
]
ok, output = _run_command(command, cwd=BASE_DIR, timeout=180)
return {
"ok": ok,
"label": "Docker Compose (website, mongodb)",
"logs": _truncate_log_blob(output),
}
def _collect_systemd_service_snapshot(service_name: str, lines: int = 160) -> dict:
status_ok, status_out, status_cmd = _run_first_success(
_collect_command_candidates(["systemctl"], ["is-active", service_name]),
timeout=60,
)
status = (status_out or "").strip().splitlines()[-1] if status_ok else "unavailable"
logs_ok, logs_out, logs_cmd = _run_first_success(
_collect_command_candidates(
["journalctl"],
["-u", service_name, "--no-pager", "-n", str(max(lines, 40))],
),
timeout=120,
)
if logs_ok:
logs_text = _truncate_log_blob(logs_out)
else:
logs_text = _truncate_log_blob(f"Service-Logs konnten nicht geladen werden.\n{logs_out}")
return {
"service": service_name,
"status": status,
"status_ok": status_ok,
"status_command": status_cmd,
"logs_ok": logs_ok,
"logs_command": logs_cmd,
"logs": logs_text,
}
def _collect_homepage_service_logs() -> dict:
services = [
"invario-hosts-sync.service",
"invario-stack-autostart.service",
"nginx.service",
]
snapshots = [_collect_systemd_service_snapshot(service) for service in services]
return {
"generated_at": _utc_now_iso(),
"core": _collect_core_live_logs(),
"services": snapshots,
}
def _collect_instance_logs(subdomain: str) -> tuple[bool, str]:
instance_dir = _resolve_instance_dir(subdomain)
if not instance_dir:
return False, "Instanzverzeichnis nicht gefunden."
command = _instance_compose_cmd(
instance_dir,
["logs", "--no-color", "--tail", "500"],
)
if not command:
return False, "Compose-Datei der Instanz wurde nicht gefunden."
return _run_command(command, cwd=instance_dir, timeout=180)
SLIM_BACKUP_EXCLUDE_DIRS = {
".git",
".github",
"logs",
"certs",
"test-data",
"__pycache__",
".pytest_cache",
".mypy_cache",
"node_modules",
".venv",
"venv",
"tmp",
"temp",
}
SLIM_BACKUP_EXCLUDE_FILES = {
".env",
".env.local",
".env.production",
}
SLIM_BACKUP_EXCLUDE_SUFFIXES = (
".log",
".tmp",
".cache",
".pid",
)
def _is_excluded_from_slim_backup(rel_path: str, is_dir: bool) -> bool:
parts = [p for p in rel_path.split(os.sep) if p and p != "."]
if not parts:
return False
if any(part in SLIM_BACKUP_EXCLUDE_DIRS for part in parts):
return True
name = parts[-1]
if not is_dir and name in SLIM_BACKUP_EXCLUDE_FILES:
return True
if not is_dir and name.lower().endswith(SLIM_BACKUP_EXCLUDE_SUFFIXES):
return True
return False
def _build_instance_backup_archive(subdomain: str) -> tuple[bool, str, str | None]:
instance_dir = _resolve_instance_dir(subdomain)
if not instance_dir:
return False, "Instanzverzeichnis nicht gefunden.", None
safe_name = _slugify_subdomain(subdomain)
stamp = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
archive_path = os.path.join(tempfile.gettempdir(), f"instance-backup-{safe_name}-{stamp}.tar.gz")
try:
with tarfile.open(archive_path, "w:gz") as tf:
for root, dirs, files in os.walk(instance_dir):
rel_root = os.path.relpath(root, instance_dir)
if rel_root == ".":
rel_root = ""
kept_dirs = []
for dirname in dirs:
rel_dir = os.path.join(rel_root, dirname) if rel_root else dirname
if _is_excluded_from_slim_backup(rel_dir, is_dir=True):
continue
kept_dirs.append(dirname)
dirs[:] = kept_dirs
for filename in files:
rel_file = os.path.join(rel_root, filename) if rel_root else filename
if _is_excluded_from_slim_backup(rel_file, is_dir=False):
continue
full_path = os.path.join(root, filename)
tf.add(full_path, arcname=rel_file)
except Exception as exc:
return False, f"Backup-Archiv konnte nicht erstellt werden: {exc}", None
filename = f"instance-{safe_name}-slim-backup-{stamp}.tar.gz"
return True, filename, archive_path
def _safe_extract_tar_archive(archive_path: str, target_dir: str) -> tuple[bool, str]:
target_abs = os.path.abspath(target_dir)
try:
with tarfile.open(archive_path, "r:gz") as tf:
for member in tf.getmembers():
member_path = os.path.abspath(os.path.join(target_abs, member.name))
if not member_path.startswith(target_abs + os.sep) and member_path != target_abs:
return False, "Unsicherer Pfad im Backup-Archiv erkannt."
tf.extractall(path=target_abs)
except Exception as exc:
return False, f"Backup konnte nicht entpackt werden: {exc}"
return True, "Backup erfolgreich entpackt."
def _set_instance_library_enabled(instance_dir: str, enabled: bool) -> tuple[bool, str]:
config_path = os.path.join(instance_dir, "config.json")
if not os.path.isfile(config_path):
return False, "config.json nicht gefunden."
try:
with open(config_path, "r", encoding="utf-8") as handle:
config = json.load(handle)
except Exception as exc:
return False, f"config.json konnte nicht gelesen werden: {exc}"
modules = config.get("modules")
if not isinstance(modules, dict):
modules = {}
config["modules"] = modules
library = modules.get("library")
if not isinstance(library, dict):
library = {}
modules["library"] = library
library["enabled"] = bool(enabled)
try:
with open(config_path, "w", encoding="utf-8") as handle:
json.dump(config, handle, indent=2, ensure_ascii=False)
handle.write("\n")
except Exception as exc:
return False, f"config.json konnte nicht geschrieben werden: {exc}"
return True, "Bibliothekseinstellung gespeichert."
def _restart_instance_stack(instance_dir: str) -> tuple[bool, str]:
restart_cmd = _instance_compose_cmd(instance_dir, ["restart", "app", "nginx", "mongodb", "redis"])
if not restart_cmd:
return False, "Compose-Datei der Instanz wurde nicht gefunden."
up_cmd = _instance_compose_cmd(instance_dir, ["up", "-d", "--remove-orphans"])
if not up_cmd:
return False, "Compose-Datei der Instanz wurde nicht gefunden."
# restart may fail for stopped services; ensure desired state with up -d afterwards.
_run_command(restart_cmd, cwd=instance_dir, timeout=420)
ok, output = _run_command(up_cmd, cwd=instance_dir, timeout=900)
if ok:
return True, output
# Missing app image is a common restart failure after tag changes.
if "local app image not found" in (output or "").lower() and os.path.isfile(os.path.join(instance_dir, "update.sh")):
upd_ok, upd_out = _run_command(["bash", "./update.sh"], cwd=instance_dir, timeout=2400)
if upd_ok:
ok2, out2 = _run_command(up_cmd, cwd=instance_dir, timeout=900)
if ok2:
return True, f"{upd_out}\n{out2}".strip()
return False, out2
return False, upd_out
return False, output
def _delete_instance_stack(subdomain: str) -> tuple[bool, str]:
target_dir = _instance_dir_path(subdomain)
if not target_dir:
return False, "Ungültige Subdomain für Löschung."
details: list[str] = []
if os.path.isdir(target_dir):
compose_file = _instance_compose_file(target_dir)
if compose_file:
down_cmd = _instance_compose_cmd(
target_dir,
["down", "--remove-orphans", "--volumes", "--timeout", "40"],
)
if not down_cmd:
return False, "Compose-Datei der Instanz wurde nicht gefunden."
down_ok, down_out = _run_command(down_cmd, cwd=target_dir, timeout=900)
if not down_ok:
return False, f"Docker-Stack konnte nicht gestoppt werden.\n{_tail_output(down_out, 12)}"
details.append("Docker-Stack gestoppt und Volumes entfernt.")
try:
shutil.rmtree(target_dir)
details.append("Instanzverzeichnis gelöscht.")
except Exception as exc:
return False, f"Instanzverzeichnis konnte nicht gelöscht werden: {exc}"
else:
details.append("Instanzverzeichnis war bereits entfernt.")
nginx_sites_available = "/etc/nginx/sites-available"
nginx_sites_enabled = "/etc/nginx/sites-enabled"
site_name = f"inventarsystem-{subdomain}.conf"
avail_file = os.path.join(nginx_sites_available, site_name)
enabled_file = os.path.join(nginx_sites_enabled, site_name)
removed_nginx_file = False
for path in (enabled_file, avail_file):
try:
if os.path.lexists(path):
os.remove(path)
removed_nginx_file = True
except Exception:
continue
if removed_nginx_file:
if os.path.isfile("/usr/sbin/nginx") or os.path.isfile("/usr/bin/nginx"):
test_ok, _ = _run_command(["nginx", "-t"], timeout=60)
if test_ok:
_run_command(["nginx", "-s", "reload"], timeout=60)
details.append("Nginx-Site entfernt.")
return True, " ".join(details) if details else "Instanz gelöscht."
def _delete_school_instance(subdomain: str) -> bool:
key = _sanitize_text(subdomain or "", 63)
if not key:
return False
client = None
try:
client, col = _get_collection("school_instances")
result = col.delete_one({"subdomain": key})
return result.deleted_count > 0
except PyMongoError:
return False
finally:
if client:
client.close()
def _instance_db_name(instance_dir: str) -> str:
config_path = os.path.join(instance_dir, "config.json")
try:
with open(config_path, "r", encoding="utf-8") as handle:
cfg = json.load(handle)
value = ((cfg or {}).get("mongodb") or {}).get("db")
if isinstance(value, str) and value.strip():
return value.strip()
except Exception:
pass
return "Inventarsystem"
def _create_instance_admin_user(
subdomain: str,
username: str,
password: str,
first_name: str,
last_name: str,
) -> tuple[bool, str]:
instance_dir = _resolve_instance_dir(subdomain)
if not instance_dir:
return False, "Instanzverzeichnis nicht gefunden."
user_name = _sanitize_text(username, 80)
pwd = (password or "").strip()
first = _sanitize_text(first_name or "Admin", 80) or "Admin"
last = _sanitize_text(last_name or "User", 80) or "User"
if not _validate_username(user_name):
return False, "Ungültiger Benutzername für Instanz-Admin."
if len(pwd) < 8:
return False, "Passwort muss mindestens 8 Zeichen haben."
pwd_hash = hashlib.sha512(pwd.encode("utf-8")).hexdigest()
db_name = _instance_db_name(instance_dir)
js_user = json.dumps(user_name)
js_hash = json.dumps(pwd_hash)
js_first = json.dumps(first)
js_last = json.dumps(last)
eval_script = (
"const username=" + js_user + ";"
"const hash=" + js_hash + ";"
"const first=" + js_first + ";"
"const last=" + js_last + ";"
"const existing=db.users.findOne({Username:username});"
"if(existing){"
"db.users.updateOne({Username:username},{$set:{Password:hash,Admin:true,name:first,last_name:last,updated_at:new Date().toISOString()}});"
"print('UPDATED');"
"}else{"
"db.users.insertOne({Username:username,Password:hash,Admin:true,active_ausleihung:null,name:first,last_name:last,favorites:[]});"
"print('CREATED');"
"}"
)
up_cmd = _instance_compose_cmd(instance_dir, ["up", "-d", "mongodb"])
if not up_cmd:
return False, "Compose-Datei der Instanz wurde nicht gefunden."
up_ok, up_out = _run_command(up_cmd, cwd=instance_dir, timeout=180)
if not up_ok:
return False, f"MongoDB der Instanz konnte nicht gestartet werden: {_tail_output(up_out, 10)}"
exec_cmd = _instance_compose_cmd(
instance_dir,
[
"exec",
"-T",
"mongodb",
"mongosh",
"--quiet",
"--eval",
eval_script,
db_name,
],
)
if not exec_cmd:
return False, "Compose-Datei der Instanz wurde nicht gefunden."
ok, output = _run_command(exec_cmd, cwd=instance_dir, timeout=240)
if not ok:
return False, f"Instanz-Admin konnte nicht angelegt werden: {_tail_output(output, 12)}"
state = "aktualisiert" if "UPDATED" in output else "erstellt"
return True, f"Instanz-Admin '{user_name}' wurde {state}."
def _get_school_instance_by_subdomain(subdomain: str) -> dict | None:
key = _sanitize_text(subdomain or "", 63)
if not key:
return None
client = None
try:
client, col = _get_collection("school_instances")
return col.find_one({"subdomain": key})
except PyMongoError:
return None
finally:
if client:
client.close()
def _get_instance_for_user(username, display_name=None, user_id=None):
# Ensure we fall back to session if user_id isn't explicitly passed
user_id = user_id or session.get("user_id")
client, col = _get_collection("instances")
try:
query = {
"$or": [
{"owner_username": username},
{"username": username},
{"owner_id": user_id}
]
}
return col.find_one(query)
finally:
client.close()
def _with_public_id(doc: dict | None) -> dict | None:
if not doc:
return doc
if not doc.get("id") and doc.get("_id") is not None:
doc["id"] = str(doc.get("_id"))
return doc
def _appointment_query_from_id(appointment_id: str) -> dict | None:
lookup = (appointment_id or "").strip()
if not lookup:
return None
if lookup.startswith("a-"):
return {"id": lookup}
try:
return {"_id": ObjectId(lookup)}
except Exception:
return {"id": lookup}
def _post_query_from_id(post_id: str) -> dict | None:
lookup = (post_id or "").strip()
if not lookup:
return None
if lookup.startswith("p-"):
return {"id": lookup}
try:
return {"_id": ObjectId(lookup)}
except Exception:
return {"id": lookup}
def _get_blocked_days() -> list:
client = None
entries = []
try:
client, col = _get_collection("blocked_days")
entries = list(col.find({}, {"_id": 0}).sort("date", 1))
except PyMongoError:
return []
finally:
if client:
client.close()
normalized = []
for item in entries:
day = (item.get("date") or "").strip()
if not day:
continue
normalized.append(
{
"date": day,
"reason": (item.get("reason") or "").strip()[:200],
"blocked_by": (item.get("blocked_by") or "").strip()[:80],
"created_at": item.get("created_at") or "",
}
)
normalized.sort(key=lambda value: value.get("date", ""))
return normalized
def _get_blocked_day_map() -> dict:
blocked = {}
for item in _get_blocked_days():
blocked[item.get("date", "")] = item
return blocked
def _sanitize_html(html_content: str, max_length: int = 50000) -> str:
"""Sanitize HTML content: allow safe tags only."""
if not html_content:
return ""
html_content = html_content[:max_length]
allowed_tags = ["p", "br", "strong", "em", "u", "h1", "h2", "h3", "h4", "h5", "h6", "ul", "ol", "li", "a", "blockquote", "code", "pre"]
allowed_attrs = {"a": ["href", "title"]}
return bleach.clean(html_content, tags=allowed_tags, attributes=allowed_attrs, strip=True)
def _validate_username(username: str) -> bool:
"""Validate username format: alphanumeric, underscore, dash, 3-30 chars."""
if not username or not isinstance(username, str):
return False
username = username.strip()
if len(username) < 3 or len(username) > 30:
return False
return all(c.isalnum() or c in "_-" for c in username)
def _validate_email(email: str) -> bool:
"""Simple email validation for registration input."""
value = (email or "").strip()
if len(value) < 5 or len(value) > 254:
return False
if "@" not in value or value.count("@") != 1:
return False
local, domain = value.split("@", 1)
if not local or not domain or "." not in domain:
return False
return True
def _find_user(username: str):
username_key = (username or "").strip()
if not username_key:
return None
try:
# First try an exact lookup using the original username.
raw_user = user_store.get_user(username_key)
normalized = _normalize_user_doc(raw_user)
if normalized:
return normalized
# Fallback: case-insensitive lookup to avoid role-check failures
# when session username casing differs from stored Username casing.
all_users = user_store.get_all_users() or []
for entry in all_users:
candidate = (entry.get("Username") or entry.get("username") or "").strip()
if candidate.lower() == username_key.lower():
return _normalize_user_doc(entry)
return None
except Exception:
return None
def login_required(view_func):
@wraps(view_func)
def wrapped(*args, **kwargs):
if "username" not in session:
flash("Bitte zuerst einloggen.", "error")
return redirect(url_for("login"))
return view_func(*args, **kwargs)
return wrapped
def admin_required(view_func):
@wraps(view_func)
def wrapped(*args, **kwargs):
if "username" not in session:
flash("Bitte zuerst einloggen.", "error")
return redirect(url_for("login"))
current_user = _find_user(session.get("username"))
if not current_user or not current_user.get("is_admin", False):
flash("Zugriff verweigert. Admin-Rechte erforderlich.", "error")
return redirect(url_for("default"))
return view_func(*args, **kwargs)
return wrapped
"""-----------------------------------------------------------------------------------------------------------------"""
@app.route("/", methods=["GET", "POST"])
def default():
return render_template("main.html")
@app.route('/health')
def health():
return jsonify({"status": "ok", "time": _utc_now_iso()}), 200
@app.route('/robots.txt')
def robots_txt():
path = os.path.join(BASE_DIR, 'static', 'robots.txt')
if os.path.exists(path):
return send_file(path, mimetype='text/plain')
return (
"User-agent: *\nDisallow: /admin/\nAllow: /\nSitemap: /sitemap.xml\n",
200,
{"Content-Type": "text/plain"},
)
@app.route('/google<token>.html')
def google_verification(token):
# Serve files named like google<alphanumeric>.html from the static/ directory
filename = f"google{token}.html"
path = os.path.join(BASE_DIR, 'static', filename)
if os.path.exists(path):
return send_file(path, mimetype='text/html')
abort(404)
@app.route('/sitemap.xml')
def sitemap_xml():
site_root = request.url_root.rstrip('/')
sitemap_path = os.path.join(BASE_DIR, 'static', 'sitemap.xml')
if os.path.exists(sitemap_path):
try:
with open(sitemap_path, 'r', encoding='utf-8') as fh:
content = fh.read().replace('{{ SITE_ROOT }}', site_root)
return app.response_class(content, mimetype='application/xml')
except Exception:
pass
xml = (
'<?xml version="1.0" encoding="UTF-8"?>\n'
'<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">\n'
' <url>\n'
f' <loc>{site_root}</loc>\n'
' <changefreq>daily</changefreq>\n'
' <priority>1.0</priority>\n'
' </url>\n'
'</urlset>'
)
return app.response_class(xml, mimetype='application/xml')
@app.route('/dienstleistungen')
def dienstleistungen():
return render_template("dienstleistungen.html")
@app.route('/projekte')
def projekte():
return render_template("projekte.html")
@app.route('/inventarsystem')
def inventarsystem():
return render_template("inventarsystem.html")
@app.route('/kontakt')
def kontakt():
return render_template("kontakt.html")
@app.route('/login', methods=['GET', 'POST'])
def login():
if 'username' in session:
return redirect(url_for('default'))
if request.method == 'POST':
data = request.get_json(silent=True) or {}
username = (data.get("username") or request.form.get('username') or "").strip()
password = data.get("password") or request.form.get('password') or ""
if not username or not password:
if request.is_json:
return jsonify({"error": "Missing login data"}), 400
flash('Bitte Benutzername und Passwort eingeben.', 'error')
return redirect(url_for('login'))
try:
stored_user_raw = user_store.check_nm_pwd(username, password)
except Exception:
stored_user_raw = None
stored_user = _normalize_user_doc(stored_user_raw)
if stored_user:
if stored_user.get("verified", True) is False:
flash('Bitte verifizieren Sie Ihr Konto mit dem Code aus der E-Mail, bevor Sie sich einloggen.', 'error')
return redirect(url_for('verify_account', username=stored_user.get("username")))
session['username'] = stored_user.get("username")
session['display_name'] = stored_user.get("display_name") or stored_user.get("username")
session['is_admin'] = stored_user.get("is_admin", False)
if request.is_json:
return jsonify({"status": "ok"}), 200
return redirect(url_for('default'))
if request.is_json:
return jsonify({"error": "Invalid credentials"}), 401
flash('Login fehlgeschlagen. Bitte prüfen Sie Ihre Eingaben.', 'error')
return render_template('login.html')
@app.route("/register", methods=["GET", "POST"])
def register():
if request.method == "POST":
username = (request.form.get("username") or "").strip()
school_name = (request.form.get("school_name") or "").strip()
contact_person = (request.form.get("contact_person") or "").strip()
email = (request.form.get("email") or "").strip().lower()
password = request.form.get("password") or ""
password_repeat = request.form.get("password_repeat") or ""
marketing_opt_in = True if request.form.get("marketing_opt_in") == "on" else False
if not username or not school_name or not contact_person or not email or not password or not password_repeat:
flash("Bitte alle Felder ausfüllen.", "error")
return redirect(url_for("register"))
if not _validate_username(username):
flash("Benutzername muss 3-30 Zeichen lang sein und nur Buchstaben, Zahlen, - und _ enthalten.", "error")
return redirect(url_for("register"))
if len(password) < 8:
flash("Das Passwort muss mindestens 8 Zeichen haben.", "error")
return redirect(url_for("register"))
if password != password_repeat:
flash("Die Passwörter stimmen nicht überein.", "error")
return redirect(url_for("register"))
if _find_user(username):
flash("Benutzername bereits vergeben.", "error")
return redirect(url_for("register"))
if not _validate_email(email):
flash("Bitte eine gültige E-Mail-Adresse angeben.", "error")
return redirect(url_for("register"))
school_name = _sanitize_text(school_name, 120)
contact_person = _sanitize_text(contact_person, 120)
email = _sanitize_text(email, 254)
from modules.emailservice.email import send_register_token
import secrets
token = secrets.token_hex(3).upper()
try:
existing_users = user_store.get_all_users() or []
is_first_user = len(existing_users) == 0
if not user_store.add_user(
username, password, school_name, contact_person, email,
marketing_opt_in=marketing_opt_in,
verified=False,
verification_token=token
):
flash("Benutzer konnte nicht erstellt werden.", "error")
return redirect(url_for("register"))
if is_first_user:
user_store.make_admin(username)
email_sent = send_register_token(email, token)
if not email_sent:
flash("Konto erstellt, aber E-Mail konnte nicht gesendet werden. Bitte Support kontaktieren.", "error")
else:
flash("Schulregistrierung erfolgreich. Bitte überprüfen Sie Ihre E-Mails für den Bestätigungscode.", "success")
except Exception:
flash("MongoDB ist derzeit nicht erreichbar.", "error")
return redirect(url_for("register"))
return redirect(url_for("verify_account", username=username))
return render_template("register.html")
@app.route("/verify", methods=["GET", "POST"])
def verify_account():
if request.method == "POST":
username = (request.form.get("username") or "").strip()
token = (request.form.get("token") or "").strip().upper()
if not username or not token:
flash("Bitte Benutzername und Token eingeben.", "error")
return redirect(url_for("verify_account", username=username))
user = _find_user(username)
if not user:
flash("Benutzer nicht gefunden.", "error")
return redirect(url_for("verify_account"))
if user.get("verified"):
flash("Konto ist bereits verifiziert. Bitte einloggen.", "success")
return redirect(url_for("login"))
is_valid = user_store.verify_user_token(username, token)
if is_valid:
flash("Konto erfolgreich verifiziert! Sie können sich jetzt einloggen.", "success")
return redirect(url_for("login"))
else:
flash("Ungültiger Bestätigungscode. Bitte erneut versuchen.", "error")
return redirect(url_for("verify_account", username=username))
prefilled_username = request.args.get("username", "")
return render_template("verify.html", username=prefilled_username)
@app.route("/resend-token", methods=["POST"])
def resend_token():
username = (request.form.get("username") or "").strip()
user = _find_user(username)
if not user:
flash("Benutzer nicht gefunden.", "error")
return redirect(url_for("verify_account"))
if user.get("verified"):
flash("Konto ist bereits verifiziert.", "info")
return redirect(url_for("login"))
new_token = secrets.token_hex(3).upper()
user_store.update_verification_token(user["username"], new_token)
if send_register_token(user["email"], new_token):
flash("Ein neuer Bestätigungscode wurde an Ihre E-Mail-Adresse gesendet.", "success")
else:
flash("E-Mail konnte nicht gesendet werden. Bitte versuchen Sie es später erneut.", "error")
return redirect(url_for("verify_account", username=username))
@app.route("/change-email", methods=["POST"])
def change_email_and_resend():
username = (request.form.get("username") or "").strip()
new_email = (request.form.get("new_email") or "").strip().lower()
if not username or not new_email or "@" not in new_email:
flash("Bitte geben Sie eine gültige E-Mail-Adresse ein.", "error")
return redirect(url_for("verify_account", username=username))
user = _find_user(username)
if not user:
flash("Benutzer nicht gefunden.", "error")
return redirect(url_for("verify_account"))
if user_store.email_exists(new_email):
flash("Diese E-Mail-Adresse wird bereits verwendet.", "error")
return redirect(url_for("verify_account", username=username))
new_token = secrets.token_hex(3).upper()
user_store.update_user_email_and_token(username, new_email, new_token)
if send_register_token(new_email, new_token):
flash("E-Mail-Adresse erfolgreich geändert und neuer Code gesendet.", "success")
else:
flash("E-Mail-Adresse geändert, aber der Code konnte nicht gesendet werden.", "error")
return redirect(url_for("verify_account", username=username))
@app.route('/user/profile', methods=['GET', 'POST'])
def user_profile():
if 'username' not in session:
flash("Bitte melden Sie sich an, um auf Ihr Profil zuzugreifen.", "warning")
return redirect(url_for('login'))
username = session['username']
user = user_store.get_user(username)
if not user:
flash("Benutzer nicht gefunden. Bitte loggen Sie sich erneut ein.", "danger")
session.clear()
return redirect(url_for('login'))
if request.method == 'POST':
action = request.form.get('action')
if action == 'update_info':
email = request.form.get('email')
full_name = request.form.get('full_name')
user_store.update_user_profile(username, full_name, email)
session['email'] = email
session['full_name'] = full_name
flash("Ihre Profilinformationen wurden erfolgreich aktualisiert.", "success")
return redirect(url_for('user_profile'))
elif action == 'change_password':
current_password = request.form.get('current_password')
new_password = request.form.get('new_password')
confirm_password = request.form.get('confirm_password')
if not user_store.check_nm_pwd(username, current_password):
flash("Das aktuelle Passwort ist nicht korrekt.", "danger")
return redirect(url_for('user_profile'))
if new_password != confirm_password:
flash("Die neuen Passwörter stimmen nicht überein.", "danger")
return redirect(url_for('user_profile'))
if user_store.update_password(username, new_password):
flash("Ihr Passwort wurde erfolgreich geändert.", "success")
else:
flash("Das Passwort ist zu schwach. Bitte verwenden Sie mindestens 6 Zeichen.", "warning")
return redirect(url_for('user_profile'))
elif action == 'delete_account':
current_password = request.form.get('current_password')
if not user_store.check_nm_pwd(username, current_password):
flash("Das Passwort ist nicht korrekt. Konto konnte nicht gelöscht werden.", "danger")
return redirect(url_for('user_profile'))
if user_store.delete_user(username):
session.clear()
flash("Ihr Konto wurde erfolgreich gelöscht.", "info")
return redirect(url_for('default'))
else:
flash("Fehler beim Löschen des Kontos. Bitte versuchen Sie es später erneut.", "danger")
return redirect(url_for('user_profile'))
return render_template('user_profile.html', user=user)
@app.route('/download-sample')
def download_sample():
# Ensure the file is only served from the specific directory
directory = os.path.join(app.root_path, 'static', 'downloads')
try:
return send_from_directory(directory, 'download_sample.pdf', as_attachment=True)
except FileNotFoundError:
abort(404)
@app.route('/preise', methods=['GET'])
def preise():
return render_template("preise.html")
@app.route('/preise/book-option', methods=['POST'])
@login_required
def book_option_package():
# Supports both "packages" (multi-select/cart) and "package" (single form)
packages_raw = _sanitize_text(request.form.get("packages") or request.form.get("package") or "", 200).lower()
booking_flow = _sanitize_text(request.form.get("flow") or "payment", 24).lower()
if booking_flow not in BOOKING_FLOW_MAP:
booking_flow = "payment"
if not packages_raw:
flash("Keine Buchungsoption ausgewählt.", "error")
return redirect(url_for("preise"))
return redirect(url_for("booking_payment", packages=packages_raw, flow=booking_flow))
ALL_MODULE_KEYS = ["library", "inventory", "student_cards", "terminplan"]
PACKAGE_MODULE_MAP = {
"buecherei": ["library"],
"inventarsystem": ["inventory"],
"terminverwaltung": ["terminplan"],
"starter": ["library", "inventory", "student_cards"],
"startpaket": ["library", "inventory", "student_cards"],
"advanced": ["library", "inventory", "student_cards", "terminplan"],
"komplettpaket": ["library", "inventory", "student_cards", "terminplan"],
}
def build_module_config_string(package_list: list) -> str:
active_modules = set()
for pkg in package_list:
pkg_key = pkg.strip().lower()
# 1. Direkter Modulname
if pkg_key in ALL_MODULE_KEYS:
active_modules.add(pkg_key)
# 2. Exaktes Paket-Mapping
elif pkg_key in PACKAGE_MODULE_MAP:
active_modules.update(PACKAGE_MODULE_MAP[pkg_key])
# 3. Fallback für Teilstrings (z.B. "startpaket" -> "starter")
elif "start" in pkg_key:
active_modules.update(["library", "inventory", "student_cards"])
elif "advanced" in pkg_key or "komplett" in pkg_key:
active_modules.update(ALL_MODULE_KEYS)
config_parts = [
f"{mod}={'on' if mod in active_modules else 'off'}"
for mod in ALL_MODULE_KEYS
]
return " ".join(config_parts)
def background_booking_provisioning(
app_instance,
prov_id,
subdomain,
package_list,
user_id,
school_name,
user_email,
username,
address,
price,
software_name="Invario Inventarsystem"
):
"""Hintergrund-Worker, der die Tenant-Bereitstellung über das Host-Watcher-Triggersystem auslöst."""
with app_instance.app_context():
req_client, req_col = _get_collection("instance_requests")
try:
# 1. Status auf "in_progress" setzen
req_col.update_one({"_id": ObjectId(prov_id)}, {"$set": {"provision_status": "in_progress"}})
# 2. Nächsten Port und Admin-Passwort ermitteln
inst_client_check, inst_col_check = _get_collection("instances")
try:
existing_instances = list(inst_col_check.find({}, {"https_port": 1, "port": 1}))
used_ports = [
doc.get("https_port") or doc.get("port")
for doc in existing_instances
if doc.get("https_port") or doc.get("port")
]
finally:
inst_client_check.close()
next_port = max(used_ports) + 1 if used_ports else 10002
admin_password = generate_secure_password()
# Generiere den korrekten Modul-String
module_config_str = build_module_config_string(package_list)
# 3. Trigger-Datei für den Host-Watcher schreiben
os.makedirs(TRIGGER_DIR, exist_ok=True)
trigger_file = os.path.join(TRIGGER_DIR, f"{subdomain}.json")
trigger_payload = {
"action": "add",
"request_id": str(prov_id),
"slug": subdomain,
"port": next_port,
"password": admin_password,
"module_config": module_config_str
}
with open(trigger_file, "w") as f:
json.dump(trigger_payload, f)
print(f"[INFO] Trigger-Datei für Host-Watcher erstellt: {trigger_file} mit Payload: {trigger_payload}")
# 4. Instanz-Eintrag in der Datenbank registrieren
domain = f"{subdomain}.invario-software.de"
instance_doc = {
"owner_id": user_id,
"owner_username": username,
"school_name": school_name,
"subdomain": subdomain,
"domain": domain,
"https_port": next_port,
"status": "ready",
"nginx_status": "active",
"admin_username": "admin",
"admin_password": admin_password,
"created_at": _utc_now_iso()
}
inst_client, inst_col = _get_collection("instances")
try:
created_inst_id = str(inst_col.insert_one(instance_doc).inserted_id)
finally:
inst_client.close()
# 5. Bestätigungs-E-Mail inkl. Vertrag & Anhängen versenden
try:
if user_email:
formatted_date = datetime.today().strftime('%d.%m.%Y')
send_accreditation_email(
recipient=user_email,
domain=domain,
username="admin",
password=admin_password,
school_name=school_name,
address=address,
price=price,
date=formatted_date,
software_name=software_name,
user_id=username
)
except Exception as mail_err:
print(f"[WARNING] E-Mail-Versand fehlgeschlagen: {mail_err}")
# 6. Status in instance_requests auf "ready" setzen
req_col.update_one(
{"_id": ObjectId(prov_id)},
{"$set": {
"provision_status": "ready",
"provision_port": next_port,
"instance_id": created_inst_id,
"url": f"https://{domain}",
"username": "admin",
"password": admin_password,
"updated_at": _utc_now_iso()
}}
)
print(f"[SUCCESS] Provisioning-Trigger erfolgreich gesendet für {subdomain}")
except Exception as e:
print(f"[ERROR] Async Provisioning fehlgeschlagen für {subdomain}:")
traceback.print_exc()
req_col.update_one({"_id": ObjectId(prov_id)}, {"$set": {"provision_status": "failed", "last_error": str(e)}})
finally:
req_client.close()
@app.route('/booking/payment', methods=['GET', 'POST'])
@login_required
def booking_payment():
packages_str = _sanitize_text(request.values.get("packages") or request.values.get("package") or "", 200).lower()
booking_flow = _sanitize_text(request.values.get("flow") or "payment", 24).lower()
if booking_flow not in BOOKING_FLOW_MAP:
booking_flow = "payment"
is_json_request = "application/json" in request.headers.get("Accept", "") or request.is_json
if not packages_str:
if is_json_request:
return jsonify({"success": False, "error": "Keine Buchungsoption ausgewählt."}), 400
flash("Keine Buchungsoption ausgewählt.", "error")
return redirect(url_for("preise"))
package_list = [p.strip() for p in packages_str.split(',') if p.strip()]
selected_package_names = []
total_price = 0.0
for pkg in package_list:
name = BOOKING_PACKAGE_MAP.get(pkg)
if name:
selected_package_names.append(name)
total_price += BOOKING_PRICE_MAP.get(pkg, 0.0)
else:
err_msg = f"Ungültige Buchungsoption: {pkg}"
if is_json_request:
return jsonify({"success": False, "error": err_msg}), 400
flash(err_msg, "error")
return redirect(url_for("preise"))
selected_package = ", ".join(selected_package_names)
package_raw = packages_str
flow_config = BOOKING_FLOW_MAP[booking_flow]
form_data = {
"contact_person": _sanitize_text(request.values.get("contact_person") or "", 120),
"school_name": _sanitize_text(request.values.get("school_name") or "", 160),
"school_number": _sanitize_text(request.values.get("school_number") or "", 60),
"contact_email": _sanitize_text(request.values.get("contact_email") or "", 160).lower(),
"contact_phone": _sanitize_text(request.values.get("contact_phone") or "", 60),
"billing_organization": _sanitize_text(request.values.get("billing_organization") or "", 160),
"billing_street": _sanitize_text(request.values.get("billing_street") or "", 160),
"billing_zip": _sanitize_text(request.values.get("billing_zip") or "", 20),
"billing_city": _sanitize_text(request.values.get("billing_city") or "", 120),
"billing_reference": _sanitize_text(request.values.get("billing_reference") or "", 100),
"tenant_slug": _sanitize_text(request.values.get("tenant_slug") or "", 63).lower(),
"notes": _sanitize_text(request.values.get("notes") or "", 600),
"agreements_accepted": (request.values.get("agreements_accepted") or "").strip().lower() in {"1", "on", "true", "yes"},
}
if request.method == "GET" and not form_data["contact_person"]:
form_data["contact_person"] = session.get("display_name") or session.get("username") or ""
review_data = {
"package": selected_package,
"package_key": package_raw,
"package_price": total_price,
"flow": booking_flow,
"title": flow_config["title"],
"intro": flow_config["intro"],
"confirm_label": flow_config["confirm_label"],
"username": session.get("username"),
"display_name": session.get("display_name") or session.get("username"),
"form_data": form_data,
}
if request.method == "GET":
return render_template("schnell_buchung.html", **review_data)
# --- POST Validation ---
required_fields = {
"contact_person": "Ansprechpartner",
"school_name": "Name der Schule",
"school_number": "Dienststellennummer",
"contact_email": "Dienstliche E-Mail",
"billing_organization": "Einrichtung / Amt",
"billing_street": "Straße",
"billing_zip": "PLZ",
"billing_city": "Ort",
"tenant_slug": "Gewünschte Schul-URL",
}
missing_labels = [label for key, label in required_fields.items() if not form_data.get(key)]
if missing_labels:
err_msg = f"Bitte folgende Pflichtfelder ausfüllen: {', '.join(missing_labels)}."
if is_json_request:
return jsonify({"success": False, "error": err_msg}), 400
flash(err_msg, "error")
return render_template("schnell_buchung.html", **review_data)
if not form_data["agreements_accepted"]:
err_msg = "Bitte Nutzungsbedingungen, Lizenzvereinbarung und Datenschutzerklärung bestätigen."
if is_json_request:
return jsonify({"success": False, "error": err_msg}), 400
flash(err_msg, "error")
return render_template("schnell_buchung.html", **review_data)
if not _validate_email(form_data["contact_email"]):
err_msg = "Bitte eine gültige E-Mail-Adresse angeben."
if is_json_request:
return jsonify({"success": False, "error": err_msg}), 400
flash(err_msg, "error")
return render_template("schnell_buchung.html", **review_data)
assigned_admin = _select_booking_admin()
assigned_admin_username = assigned_admin.get("username") if assigned_admin else ""
assigned_admin_display_name = assigned_admin.get("display_name") if assigned_admin else ""
assigned_admin_email = (assigned_admin.get("email") if assigned_admin else "") or BOOKING_ADMIN_EMAIL_FALLBACK
try:
# FLOW 1: Consultation / Chat
if booking_flow == "consultation":
client, col = _get_collection("chat_messages")
try:
col.insert_one(
{
"username": session.get("username"),
"sender": session.get("display_name") or session.get("username"),
"sender_role": "user",
"message": f"Gesprächswunsch bezüglich: {selected_package}",
"created_at": _utc_now_iso(),
"booking_flow": booking_flow,
"booking_package": package_raw,
"booking_data": form_data,
}
)
finally:
client.close()
if is_json_request:
return jsonify({"success": True, "redirect_url": url_for("user_chat")}), 200
flash(f"Gesprächstermin für {selected_package} wurde an das Team gesendet.", "success")
return redirect(url_for("user_chat"))
# FLOW 2: Direktbuchung mit Provisioning via Host-Watcher
if booking_flow == "payment":
booking_number = f"BOOK-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
raw_base = form_data.get("tenant_slug") or form_data.get("school_name") or session.get("username") or booking_number
base = _slugify_subdomain(raw_base).lower()[:51]
inst_check_client, inst_col_check = _get_collection("instances")
try:
existing_docs = list(inst_col_check.find({}, {"subdomain": 1}))
existing_subdomains = [d.get("subdomain") for d in existing_docs if d.get("subdomain")]
except Exception as e:
print(f"[WARNING] Konnte bestehende Subdomains nicht prüfen: {e}")
existing_subdomains = []
finally:
inst_check_client.close()
if base in existing_subdomains:
err_msg = "Die gewünschte Subdomain existiert bereits. Bitte wähle eine andere Subdomain."
if is_json_request:
return jsonify({"success": False, "error": err_msg}), 400
flash(err_msg, "error")
return render_template("schnell_buchung.html", **review_data)
subdomain = base[:63]
provision_payload = {
"user_id": session.get("user_id"),
"username": session.get("username"),
"booking_number": booking_number,
"package": package_raw,
"package_label": selected_package,
"amount_eur": total_price,
"booking_flow": booking_flow,
"booking_data": form_data,
"assigned_admin_username": assigned_admin_username,
"assigned_admin_display_name": assigned_admin_display_name,
"assigned_admin_email": assigned_admin_email,
"subdomain": subdomain,
"tenant_slug": subdomain,
"provision_status": "pending",
"created_at": _utc_now_iso(),
"source": "booking_payment",
}
req_client, req_col = _get_collection("instance_requests")
try:
prov_id = str(req_col.insert_one(provision_payload).inserted_id)
finally:
req_client.close()
app_obj = current_app._get_current_object()
user_id = session.get("user_id")
username = session.get("username")
user_email = session.get("email") or form_data.get("contact_email")
address_process = f"{form_data['billing_street']}, {form_data['billing_zip']} {form_data['billing_city']}"
price_str = f"{total_price:.2f}".replace('.', ',')
# Startet den asynchronen Provisioning-Thread mit exakten Parametern
thread = threading.Thread(
target=background_booking_provisioning,
args=(
app_obj,
prov_id,
subdomain,
package_list,
user_id,
form_data["school_name"],
user_email,
username,
address_process,
price_str,
selected_package,
)
)
thread.start()
print(f"[INFO] Background Provisioning Trigger gestartet für Subdomain: {subdomain}")
if is_json_request:
return jsonify({
"success": True,
"slug": subdomain,
"message": "Instanz wird eingerichtet...",
"estimated_seconds": 45,
"status_poll_interval_ms": 3000
}), 200
flash(f"Buchung für {selected_package} wurde erfasst. Deine Instanz wird jetzt automatisch eingerichtet.", "success")
return redirect(url_for("my_instance_management", wait_for_slug=subdomain))
except Exception as e:
print(f"[FATAL ERROR] In booking_payment Route:")
traceback.print_exc()
err_msg = "Buchungsanfrage konnte nicht gesendet werden. Ein unerwarteter Fehler ist aufgetreten."
if is_json_request:
return jsonify({"success": False, "error": err_msg}), 500
flash(err_msg, "error")
return render_template("schnell_buchung.html", **review_data)
@app.route('/instance_request/delete', methods=['POST'])
@login_required
def instance_request_delete():
prov_id = _sanitize_text(request.form.get("prov_id") or "", 64)
if not prov_id:
flash("Ungültige Anfrage.", "error")
return redirect(url_for("my_invoices"))
req_client = None
try:
req_client, req_col = _get_collection("instance_requests")
prov = req_col.find_one({"_id": ObjectId(prov_id)})
if not prov:
flash("Instanz-Anfrage nicht gefunden.", "error")
return redirect(url_for("my_invoices"))
# Only owner or admin can delete
if prov.get("username") != session.get("username") and not session.get("is_admin"):
flash("Keine Berechtigung zum Löschen dieser Anfrage.", "error")
return redirect(url_for("my_invoices"))
req_col.delete_one({"_id": prov.get("_id")})
flash("Instanz-Anfrage gelöscht.", "success")
except Exception:
flash("Instanz-Anfrage konnte nicht gelöscht werden.", "error")
finally:
if req_client:
req_client.close()
return redirect(url_for("my_invoices"))
@app.route('/admin/instances', methods=['GET'])
@admin_required
def admin_instances():
instances = _list_school_instances()
return render_template(
"admin_instances.html",
instances=instances,
parent_domain=INSTANCE_PARENT_DOMAIN,
)
@app.route('/admin/blog', methods=['GET', 'POST'])
@admin_required
def admin_blog():
if request.method == 'POST':
action = request.form.get("action", "").strip()
if action == "create":
title = _sanitize_text(request.form.get("title") or "", 200)
content = _sanitize_html(request.form.get("content") or "", 50000)
excerpt = _sanitize_text(request.form.get("excerpt") or "", 500)
if not title or not content:
flash("Bitte Titel und inhalt ausfüllen.", "error")
return redirect(url_for("admin_blog"))
client = None
try:
client, col = _get_collection("posts")
col.insert_one(
{
"id": f"p-{int(datetime.utcnow().timestamp() * 1000)}",
"title": title,
"excerpt": excerpt or (content[:150] + "...") if len(content) > 150 else content,
"content": content,
"author": session.get("username"),
"created_at": datetime.utcnow().isoformat(timespec="seconds") + "Z",
"published": True,
}
)
except PyMongoError:
flash("Beitrag konnte nicht gespeichert werden.", "error")
return redirect(url_for("admin_blog"))
finally:
if client:
client.close()
flash("Beitrag veroeffentlicht.", "success")
return redirect(url_for("admin_blog"))
elif action == "delete":
post_id = (request.form.get("post_id") or "").strip()
query = _post_query_from_id(post_id)
if not query:
flash("Ungültige Beitrag-ID.", "error")
return redirect(url_for("admin_blog"))
client = None
try:
client, col = _get_collection("posts")
result = col.delete_one(query)
except PyMongoError:
flash("Beitrag konnte nicht gelöscht werden.", "error")
return redirect(url_for("admin_blog"))
finally:
if client:
client.close()
if not result.deleted_count:
flash("Beitrag nicht gefunden.", "error")
return redirect(url_for("admin_blog"))
flash("Beitrag gelöscht.", "success")
return redirect(url_for("admin_blog"))
posts = []
client = None
try:
client, col = _get_collection("posts")
posts = list(col.find().sort("created_at", -1))
for item in posts:
_with_public_id(item)
except PyMongoError:
flash("Blogbeitraege konnten nicht geladen werden.", "error")
finally:
if client:
client.close()
return render_template("admin_blog.html", posts=posts)
@app.route('/blog')
def blog():
posts = []
client = None
try:
client, col = _get_collection("posts")
posts = list(col.find({"published": True}).sort("created_at", -1))
for item in posts:
_with_public_id(item)
except PyMongoError:
posts = []
finally:
if client:
client.close()
return render_template("blog.html", posts=posts)
@app.route('/blog/<post_id>')
def blog_post(post_id):
query = _post_query_from_id(post_id)
if not query:
flash("Ungültige Beitrag-ID.", "error")
return redirect(url_for("blog"))
client = None
post = None
try:
client, col = _get_collection("posts")
post = col.find_one(query)
_with_public_id(post)
except PyMongoError:
post = None
finally:
if client:
client.close()
if not post:
flash("Beitrag nicht gefunden.", "error")
return redirect(url_for("blog"))
return render_template("blog_post.html", post=post)
@app.route('/my/invoices')
@login_required
def my_invoices():
invoices = []
client = None
try:
client, col = _get_collection("invoices")
invoices = list(col.find({"username": session.get("username")}, {"_id": 0}).sort("created_at", -1))
# Beträge in Float umwandeln, da das Jinja-Template echte Zahlen erwartet
for inv in invoices:
amt = inv.get("amount_eur", 0)
if isinstance(amt, str):
try:
# Entfernt Tausenderpunkte und wandelt das deutsche Komma in einen Punkt um
amt = float(amt.replace('.', '').replace(',', '.'))
except ValueError:
amt = 0.0
inv["amount_eur"] = amt
except PyMongoError:
flash("Rechnungen konnten nicht geladen werden.", "error")
finally:
if client:
client.close()
# Load instance requests for this user
instance_requests = []
try:
req_client, req_col = _get_collection("instance_requests")
instance_requests = list(req_col.find({"username": session.get("username")}).sort("created_at", -1))
for item in instance_requests:
item["id"] = str(item.get("_id"))
# Auch hier sicherstellen, dass der Betrag ein Float ist
amt = item.get("amount_eur", 0)
if isinstance(amt, str):
try:
amt = float(amt.replace('.', '').replace(',', '.'))
except ValueError:
amt = 0.0
item["amount_eur"] = amt
except PyMongoError:
flash("Instanz-Anfragen konnten nicht geladen werden.", "error")
finally:
if 'req_client' in locals() and req_client:
req_client.close()
return render_template("my_invoices.html", invoices=invoices, instance_requests=instance_requests)
@app.route('/my/instance', methods=['GET', 'POST'])
@login_required
def my_instance_management():
username = _sanitize_text(session.get("username") or "", 80)
display_name = _sanitize_text(session.get("display_name") or username, 120)
user_id = session.get("user_id")
if request.method == 'POST':
flash("Nutzer können Instanzen nicht selbst erstellen oder ändern. Bitte den Administrator kontaktieren.", "error")
return redirect(url_for("my_instance_management"))
suggested_subdomain = _slugify_subdomain(display_name or username)
instance_doc = _get_instance_for_user(username, display_name, user_id)
instance_view = None
pending_request = None
if instance_doc:
instance_view = {
"school_name": _sanitize_text(instance_doc.get("school_name") or display_name, 120),
"owner_username": _sanitize_text(instance_doc.get("owner_username") or "", 80),
"subdomain": _sanitize_text(instance_doc.get("subdomain") or "", 63),
"domain": _sanitize_text(instance_doc.get("domain") or "", 190),
"https_port": int(instance_doc.get("https_port") or 0),
"status": _sanitize_text(instance_doc.get("status") or "Unbekannt", 40),
"nginx_status": _sanitize_text(instance_doc.get("nginx_status") or "unbekannt", 80),
"last_message": _sanitize_text(instance_doc.get("last_message") or "", 500),
"updated_at": instance_doc.get("updated_at") or "",
}
else:
# Wenn noch keine Instanz da ist, prüfen wir ob eine gerade erstellt wird
req_client, req_col = _get_collection("instance_requests")
try:
pending_request = req_col.find_one({
"user_id": user_id,
"provision_status": {"$in": ["pending", "in_progress"]}
})
finally:
req_client.close()
return render_template(
"my_instance.html",
instance=instance_view,
pending_request=pending_request,
suggested_subdomain=suggested_subdomain,
parent_domain=INSTANCE_PARENT_DOMAIN,
default_school_name=display_name,
)
@app.route('/my/tutorials', methods=['GET'])
@login_required
def my_tutorials():
"""Display tutorial videos for the logged-in user."""
origin = request.url_root.rstrip('/')
tutorials = [
{
"_id": "1",
"title": "Erste Schritte mit Invario",
"description": "Lernen Sie die Grundlagen von Invario und wie Sie Ihre erste Instanz einrichten.",
"video_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"thumbnail_url": "/static/images/tutorial-thumb-1.jpg",
"duration": "12:34",
"category": "Anfänger"
},
{
"_id": "2",
"title": "Inventarverwaltung",
"description": "Alles über die Verwaltung Ihres Inventars und der Ausleihvorgänge.",
"video_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"thumbnail_url": "/static/images/tutorial-thumb-2.jpg",
"duration": "18:45",
"category": "Funktionen"
},
{
"_id": "3",
"title": "Admin-Funktionen",
"description": "Erweiterte Administratorfunktionen und Systemverwaltung.",
"video_url": "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
"thumbnail_url": "/static/images/tutorial-thumb-3.jpg",
"duration": "22:15",
"category": "Admin"
}
]
normalized_tutorials = []
for item in tutorials:
normalized = dict(item)
normalized["video_url"] = _normalize_tutorial_video_url(item.get("video_url") or "", origin)
if normalized.get("video_url"):
normalized_tutorials.append(normalized)
tutorials = normalized_tutorials
return render_template("my_tutorials.html", tutorials=tutorials)
@app.route('/chat', methods=['GET', 'POST'])
@login_required
def user_chat():
client = None
if request.method == 'POST':
message = _sanitize_text(request.form.get("message") or "", 3000)
if not message:
flash("Bitte eine Nachricht eingeben.", "error")
return redirect(url_for("user_chat"))
try:
client, col = _get_collection("chat_messages")
col.insert_one(
{
"username": session.get("username"),
"sender": session.get("display_name"),
"sender_role": "user",
"message": message,
"created_at": _utc_now_iso(),
}
)
flash("Nachricht gesendet.", "success")
except PyMongoError:
flash("Nachricht konnte nicht gesendet werden.", "error")
finally:
if client:
client.close()
return redirect(url_for("user_chat"))
messages = []
try:
client, col = _get_collection("chat_messages")
messages = list(col.find({"username": session.get("username")}, {"_id": 0}).sort("created_at", 1))
except PyMongoError:
flash("Chat konnte nicht geladen werden.", "error")
finally:
if client:
client.close()
return render_template("chat.html", messages=messages)
@app.route('/tickets', methods=['GET', 'POST'])
@login_required
def user_tickets():
client = None
if request.method == 'POST':
title = _sanitize_text(request.form.get("title") or "", 200)
description = _sanitize_text(request.form.get("description") or "", 5000)
priority = _sanitize_text(request.form.get("priority") or "Normal", 30)
if not title or not description:
flash("Bitte Titel und Beschreibung ausfüllen.", "error")
return redirect(url_for("user_tickets"))
try:
client, col = _get_collection("support_tickets")
col.insert_one(
{
"username": session.get("username"),
"display_name": session.get("display_name"),
"title": title,
"description": description,
"priority": priority,
"status": "Offen",
"admin_response": "",
"created_at": _utc_now_iso(),
"updated_at": _utc_now_iso(),
}
)
flash("Support-Ticket erstellt.", "success")
except PyMongoError:
flash("Support-Ticket konnte nicht erstellt werden.", "error")
finally:
if client:
client.close()
return redirect(url_for("user_tickets"))
tickets = []
try:
client, col = _get_collection("support_tickets")
tickets = list(col.find({"username": session.get("username")}).sort("created_at", -1))
for ticket in tickets:
ticket["id"] = str(ticket.get("_id"))
except PyMongoError:
flash("Tickets konnten nicht geladen werden.", "error")
finally:
if client:
client.close()
return render_template("tickets.html", tickets=tickets)
@app.route('/admin/users', methods=['GET', 'POST'])
@admin_required
def admin_users():
if request.method == 'POST':
action = _sanitize_text(request.form.get("action") or "", 50)
username = _sanitize_text(request.form.get("username") or "", 80)
if not username:
flash("Bitte Benutzername angeben.", "error")
return redirect(url_for("admin_users"))
if action == "make_admin":
user_store.make_admin(username)
flash("Benutzer zum Admin gemacht.", "success")
elif action == "remove_admin":
user_store.remove_admin(username)
flash("Admin-Rechte entfernt.", "success")
elif action == "delete_user":
if username == session.get("username"):
flash("Sie können Ihr eigenes Konto nicht löschen.", "error")
return redirect(url_for("admin_users"))
user_store.delete_user(username)
flash("Benutzer gelöscht.", "success")
else:
flash("Ungültige Aktion.", "error")
return redirect(url_for("admin_users"))
users = _list_users_for_admin()
instances_by_owner = _list_instances_grouped_by_owner()
return render_template("admin_users.html", users=users, instances_by_owner=instances_by_owner)
@app.route('/admin/chats', methods=['GET', 'POST'])
@admin_required
def admin_chats():
client = None
selected_user = _sanitize_text(request.args.get("username") or request.form.get("username") or "", 80)
if request.method == 'POST':
message = _sanitize_text(request.form.get("message") or "", 3000)
if not selected_user or not message:
flash("Bitte Empfaenger und Nachricht angeben.", "error")
return redirect(url_for("admin_chats"))
try:
client, col = _get_collection("chat_messages")
col.insert_one(
{
"username": selected_user,
"sender": session.get("display_name") or session.get("username"),
"sender_role": "admin",
"message": message,
"created_at": _utc_now_iso(),
}
)
flash("Antwort gesendet.", "success")
except PyMongoError:
flash("Antwort konnte nicht gesendet werden.", "error")
finally:
if client:
client.close()
return redirect(url_for("admin_chats", username=selected_user))
conversations = []
messages = []
try:
client, col = _get_collection("chat_messages")
conversations = sorted(col.distinct("username"))
if selected_user:
messages = list(col.find({"username": selected_user}, {"_id": 0}).sort("created_at", 1))
except PyMongoError:
flash("Admin-Chat konnte nicht geladen werden.", "error")
finally:
if client:
client.close()
return render_template(
"admin_chats.html",
conversations=conversations,
selected_user=selected_user,
messages=messages,
)
@app.route('/admin/tickets', methods=['GET', 'POST'])
@admin_required
def admin_tickets():
client = None
if request.method == 'POST':
ticket_id = _sanitize_text(request.form.get("ticket_id") or "", 64)
status = _sanitize_text(request.form.get("status") or "", 40)
admin_response = _sanitize_text(request.form.get("admin_response") or "", 5000)
if not ticket_id:
flash("Ticket-ID fehlt.", "error")
return redirect(url_for("admin_tickets"))
try:
client, col = _get_collection("support_tickets")
col.update_one(
{"_id": ObjectId(ticket_id)},
{
"$set": {
"status": status or "In Bearbeitung",
"admin_response": admin_response,
"updated_at": _utc_now_iso(),
}
},
)
flash("Ticket aktualisiert.", "success")
except Exception:
flash("Ticket konnte nicht aktualisiert werden.", "error")
finally:
if client:
client.close()
return redirect(url_for("admin_tickets"))
tickets = []
try:
client, col = _get_collection("support_tickets")
tickets = list(col.find().sort("created_at", -1))
for ticket in tickets:
ticket["id"] = str(ticket.get("_id"))
except PyMongoError:
flash("Support-Tickets konnten nicht geladen werden.", "error")
finally:
if client:
client.close()
return render_template("admin_tickets.html", tickets=tickets)
@app.route('/admin/invoices', methods=['GET', 'POST'])
@admin_required
def admin_invoices():
client = None
if request.method == 'POST':
action = _sanitize_text(request.form.get("action") or "", 50)
invoice_id = _sanitize_text(request.form.get("invoice_id") or "", 64)
prov_id = _sanitize_text(request.form.get("prov_id") or "", 64)
try:
client, col = _get_collection("invoices")
if action == "create":
username = _sanitize_text(request.form.get("username") or "", 80)
invoice_number = _sanitize_text(request.form.get("invoice_number") or "", 120)
period = _sanitize_text(request.form.get("period") or "", 20)
due_date = _sanitize_text(request.form.get("due_date") or "", 20)
status = _sanitize_text(request.form.get("status") or "Zu prüfen", 40)
amount_text = _sanitize_text(request.form.get("amount_eur") or "0", 20)
normalized_invoice_number = invoice_number or f"INV-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
pdf_path = _save_invoice_pdf(request.files.get("invoice_pdf"), normalized_invoice_number)
try:
amount = float(amount_text)
except ValueError:
amount = 0.0
if not username or not period:
flash("Bitte Benutzername und Zeitraum angeben.", "error")
return redirect(url_for("admin_invoices"))
col.insert_one(
{
"username": username,
"invoice_number": normalized_invoice_number,
"period": period,
"amount_eur": amount,
"status": status,
"due_date": due_date or "2026-12-31",
"pdf_path": pdf_path or "",
"created_at": _utc_now_iso(),
}
)
flash("Rechnung angelegt.", "success")
elif action == "create_from_provision" and prov_id:
# Create an invoice from an instance request
req_client = None
try:
req_client, req_col = _get_collection("instance_requests")
prov = req_col.find_one({"_id": ObjectId(prov_id)})
if not prov:
flash("Instanz-Anfrage nicht gefunden.", "error")
return redirect(url_for("admin_invoices"))
username = prov.get("username") or ""
invoice_number = f"INV-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"
period = _sanitize_text(request.form.get("period") or prov.get("period") or "Sofortbuchung", 20)
due_date = _sanitize_text(request.form.get("due_date") or date.today().isoformat(), 20)
amount = float(prov.get("amount_eur") or 0.0)
col.insert_one(
{
"username": username,
"invoice_number": invoice_number,
"period": period,
"amount_eur": amount,
"status": "Zu prüfen",
"due_date": due_date,
"pdf_path": "",
"created_at": _utc_now_iso(),
"source": "booking_payment",
"booking_flow": prov.get("booking_flow"),
"booking_package": prov.get("package"),
"booking_package_label": prov.get("package_label"),
"booking_data": prov.get("booking_data"),
"assigned_admin_username": prov.get("assigned_admin_username"),
"assigned_admin_display_name": prov.get("assigned_admin_display_name"),
"assigned_admin_email": prov.get("assigned_admin_email"),
}
)
req_col.update_one({"_id": prov.get("_id")}, {"$set": {"provision_status": "invoice_created", "updated_at": _utc_now_iso()}})
flash("Rechnung aus Instanz-Anfrage erstellt.", "success")
except Exception:
flash("Fehler beim Erstellen der Rechnung aus Instanz-Anfrage.", "error")
finally:
if req_client:
req_client.close()
elif action == "update_provision" and prov_id:
# Update provision request status (accept/reject/complete)
new_status = _sanitize_text(request.form.get("prov_status") or "", 40)
req_client = None
try:
req_client, req_col = _get_collection("instance_requests")
req_col.update_one({"_id": ObjectId(prov_id)}, {"$set": {"provision_status": new_status, "updated_at": _utc_now_iso()}})
flash("Instanz-Anfrage aktualisiert.", "success")
except Exception:
flash("Instanz-Anfrage konnte nicht aktualisiert werden.", "error")
finally:
if req_client:
req_client.close()
elif action in {"run_provision", "retry_provision", "configure_provision"} and prov_id:
# Admin-triggered provisioning actions: start/retry/configure
req_client = None
try:
req_client, req_col = _get_collection("instance_requests")
prov = req_col.find_one({"_id": ObjectId(prov_id)})
if not prov:
flash("Instanz-Anfrage nicht gefunden.", "error")
return redirect(url_for("admin_invoices"))
subdomain = prov.get("subdomain") or _slugify_subdomain(prov.get("booking_data", {}).get("school_name") or prov.get("username") or prov.get("booking_number") or "prov")
provision_status = prov.get("provision_status") or "not_started"
provision_port = prov.get("provision_port") or None
provision_error = ""
provision_config_ok = prov.get("provision_config_ok") or False
if action in {"run_provision", "retry_provision"}:
try:
port = steering.instace.new(subdomain)
except Exception as e:
port = False
provision_error = str(e)
if port and port is not False:
provision_status = "started"
try:
provision_port = int(port) if isinstance(port, (int, str)) and str(port).isdigit() else port
except Exception:
provision_port = port
else:
provision_status = "failed"
if not provision_error:
provision_error = "Provisioning script returned failure"
if action == "configure_provision" or (action in {"run_provision", "retry_provision"} and prov.get("package")):
module = prov.get("package")
if module in {"inventarsystem", "buecherei", "terminverwaltung", "emailversand", "starter", "advanced"}:
try:
edit_ok = steering.instace.edit(subdomain, module)
provision_config_ok = bool(edit_ok)
if not provision_config_ok and not provision_error:
provision_error = f"Module configuration failed for {module}"
except Exception as e:
provision_config_ok = False
provision_error = str(e)
req_col.update_one({"_id": prov.get("_id")}, {"$set": {"provision_status": provision_status, "provision_port": provision_port or "", "provision_config_ok": provision_config_ok, "provision_error": provision_error, "updated_at": _utc_now_iso()}})
flash("Provisionierungsaktion ausgeführt.", "success")
except Exception as e:
flash(f"Provisionierungsaktion fehlgeschlagen: {str(e)}", "error")
finally:
if req_client:
req_client.close()
elif action == "delete_instance_request" and prov_id:
req_client = None
try:
req_client, req_col = _get_collection("instance_requests")
req_col.delete_one({"_id": ObjectId(prov_id)})
flash("Instanz-Anfrage gelöscht.", "success")
except Exception:
flash("Instanz-Anfrage konnte nicht gelöscht werden.", "error")
finally:
if req_client:
req_client.close()
elif action == "update" and invoice_id:
amount_text = _sanitize_text(request.form.get("amount_eur") or "0", 20)
invoice_number = _sanitize_text(request.form.get("invoice_number") or "", 120)
pdf_path = _save_invoice_pdf(request.files.get("invoice_pdf"), invoice_number or "invoice")
try:
amount = float(amount_text)
except ValueError:
amount = 0.0
update_payload = {
"invoice_number": invoice_number,
"period": _sanitize_text(request.form.get("period") or "", 20),
"status": _sanitize_text(request.form.get("status") or "Zu prüfen", 40),
"due_date": _sanitize_text(request.form.get("due_date") or "", 20),
"amount_eur": amount,
}
if pdf_path:
update_payload["pdf_path"] = pdf_path
col.update_one(
{"_id": ObjectId(invoice_id)},
{
"$set": update_payload
},
)
flash("Rechnung aktualisiert.", "success")
elif action == "delete" and invoice_id:
col.delete_one({"_id": ObjectId(invoice_id)})
flash("Rechnung gelöscht.", "success")
else:
flash("Ungültige Aktion.", "error")
except Exception:
flash("Rechnungsverwaltung fehlgeschlagen.", "error")
finally:
if client:
client.close()
return redirect(url_for("admin_invoices"))
invoices = []
try:
client, col = _get_collection("invoices")
invoices = list(col.find().sort("created_at", -1))
for item in invoices:
item["id"] = str(item.get("_id"))
# Fehlerbehebung: Betrag für das Template in Float umwandeln
amt = item.get("amount_eur", 0)
if isinstance(amt, str):
try:
amt = float(amt.replace('.', '').replace(',', '.'))
except ValueError:
amt = 0.0
item["amount_eur"] = amt
except PyMongoError:
flash("Rechnungen konnten nicht geladen werden.", "error")
finally:
if client:
client.close()
# Load instance requests for admin review
instance_requests = []
try:
req_client, req_col = _get_collection("instance_requests")
instance_requests = list(req_col.find().sort("created_at", -1))
for item in instance_requests:
item["id"] = str(item.get("_id"))
# Fehlerbehebung: Betrag auch hier sicher in Float umwandeln
amt = item.get("amount_eur", 0)
if isinstance(amt, str):
try:
amt = float(amt.replace('.', '').replace(',', '.'))
except ValueError:
amt = 0.0
item["amount_eur"] = amt
except PyMongoError:
flash("Instanz-Anfragen konnten nicht geladen werden.", "error")
finally:
if 'req_client' in locals() and req_client:
req_client.close()
users = _list_users_for_admin()
return render_template("admin_invoices.html", invoices=invoices, users=users, provision_requests=instance_requests)
@app.route('/datenschutz')
def datenschutz():
return render_template("datenschutz.html")
@app.route('/nutzungsbedingungen')
def nutzungsbedingungen():
return render_template("nutzungsbedingungen.html")
@app.route('/vertrag')
def vertrag():
return render_template("vertrag.html")
@app.route('/avv')
def avv():
return render_template("avv.html")
@app.route('/instance_request/check_subdomain', methods=['GET'])
def check_subdomain():
slug = request.args.get('slug', '').lower().strip()
if not slug or len(slug) < 3:
return jsonify({"available": False, "message": "Ungültiger oder zu kurzer Name"}), 400
# 1. Prüfen, ob die Subdomain bereits in den aktiven Instanzen existiert
inst_client, inst_col = _get_collection("instances")
try:
existing_instance = inst_col.find_one({"subdomain": slug})
finally:
inst_client.close()
if existing_instance:
return jsonify({"available": False}), 200
# 2. Prüfen, ob bereits eine Anfrage (Instance Request) dafür läuft
req_client, req_col = _get_collection("instance_requests")
try:
existing_request = req_col.find_one({
"tenant_slug": slug,
"provision_status": {"$in": ["pending", "in_progress"]}
})
finally:
req_client.close()
if existing_request:
return jsonify({"available": False}), 200
# 3. Optional: Gegen reservierte System-Subdomains prüfen (z.B. www, admin, api, mail)
reserved_slugs = {"www", "admin", "api", "mail", "auth", "dashboard", "support"}
if slug in reserved_slugs:
return jsonify({"available": False}), 200
# Wenn alles frei ist:
return jsonify({"available": True}), 200
@app.route('/impressum')
def impressum():
return render_template("impressum.html")
@app.route('/downloads/rechtliches')
def download_legal_pdfs_bundle():
bundle = _build_legal_pdf_bundle()
return send_file(
bundle,
mimetype="application/zip",
as_attachment=True,
download_name="invario-rechtliche-dokumente.zip",
)
@app.route('/logout')
def logout():
session.pop('username', None)
session.pop('display_name', None)
session.pop('is_admin', None)
flash('Logged out successfully', 'info')
return redirect(url_for('login'))
def generate_secure_password(length=14):
alphabet = string.ascii_letters + string.digits
return ''.join(secrets.choice(alphabet) for _ in range(length))
def _utc_now_iso():
return datetime.now(timezone.utc).isoformat()
@app.route('/instance_request/status', methods=['GET'])
def tenant_status():
"""
Status endpoint checked by the frontend polling mechanism.
"""
slug = request.args.get('slug')
req_id = request.args.get('request_id')
req_client, req_col = _get_collection("instance_requests")
try:
query = {}
if req_id:
query["_id"] = ObjectId(req_id)
elif slug:
query["tenant_slug"] = slug
else:
return jsonify({"status": "error", "message": "Missing slug or request_id"}), 400
prov_req = req_col.find_one(query)
finally:
req_client.close()
if not prov_req:
return jsonify({"status": "pending"}), 200
current_status = prov_req.get("provision_status")
# Fail immediately if background execution crashed
if current_status == "failed":
return jsonify({
"status": "failed",
"error": prov_req.get("last_error", "Tenant provisioning failed.")
}), 200
if current_status not in ["ready", "completed"]:
return jsonify({"status": "pending"}), 200
inst_client, inst_col = _get_collection("instances")
try:
instance = inst_col.find_one({"subdomain": prov_req.get("tenant_slug")})
finally:
inst_client.close()
if not instance:
return jsonify({"status": "pending"}), 200
# Perform health check on newly provisioned container endpoint
target_url = f"https://{instance['domain']}/health"
try:
res = requests.get(target_url, timeout=3, verify=False)
if res.status_code == 200:
return jsonify({
"status": "ready",
"url": f"https://{instance['domain']}",
"username": instance.get("admin_username", "admin"),
"password": instance.get("admin_password")
}), 200
except requests.RequestException:
pass
return jsonify({"status": "pending"}), 200
@app.route('/admin/test/drop_all_instances', methods=['POST'])
@login_required
def admin_delete_all_instances():
# 1. Sicherheits-Check: Nur Admins dürfen das!
# Passe diesen Check an deine bestehende Rollenverwaltung an
if not session.get("is_admin") and session.get("role") != "admin":
flash("Zugriff verweigert. Diese Aktion ist nur für Administratoren erlaubt.", "error")
return redirect(url_for('my_instance_management')) # oder zum Dashboard
inst_client, inst_col = _get_collection("instances")
req_client, req_col = _get_collection("instance_requests")
try:
# Löscht alle Dokumente in den beiden Collections
deleted_instances = inst_col.delete_many({}).deleted_count
deleted_requests = req_col.delete_many({}).deleted_count
flash(f"Test-Reset erfolgreich: {deleted_instances} Instanzen und {deleted_requests} Anfragen wurden unwiderruflich gelöscht.", "success")
print(f"[ADMIN ACTION] Alle Instanzen ({deleted_instances}) und Requests ({deleted_requests}) wurden von {session.get('username')} gelöscht.")
except Exception as e:
print(f"[ERROR] Fehler beim Löschen der Datenbank: {e}")
flash("Ein Fehler ist aufgetreten. Datenbank konnte nicht bereinigt werden.", "error")
finally:
inst_client.close()
req_client.close()
# Leite den Admin danach zurück zu einer sicheren Seite
return redirect(url_for('my_instance_management'))
@app.route('/test_email')
def test_email():
"""
Test endpoint to send a sample email.
This is for development purposes only.
"""
try:
send_accreditation_email(
recipient="maximilian.gruendinger@invario-software.de",
domain="testdomain.invario-software.de",
username="testuser",
password="TestPassword123",
school_name="Test School",
address="Test Address 123, 12345 Test City",
price="200",
date="Test City, 01.01.2023",
software_name="starter",
user_id="test"
)
return "Test email sent successfully."
except Exception as e:
return f"Failed to send test email: {str(e)}", 500
def main():
app.run(host="0.0.0.0", port=4999, debug=False)
if __name__ == "__main__":
main()