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 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")), ) 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: client, col = _get_collection("school_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 "", 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 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 "", } ) 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: str, display_name: str) -> dict | None: uname = _sanitize_text(username or "", 80) if not uname: return None client = None try: client, col = _get_collection("school_instances") return col.find_one({"owner_username": uname}, sort=[("updated_at", -1), ("created_at", -1)]) except PyMongoError: return None finally: if client: 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.html') def google_verification(token): # Serve files named like google.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 = ( '\n' '\n' ' \n' f' {site_root}\n' ' daily\n' ' 1.0\n' ' \n' '' ) 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)) def background_booking_provisioning(app_instance, prov_id, subdomain, modules_to_provision, user_id, school_name, user_email): """Hintergrund-Worker that actually deploys the instance using the correct API""" with app_instance.app_context(): req_client, req_col = _get_collection("instance_requests") try: # 1. Set Status to In Progress req_col.update_one({"_id": ObjectId(prov_id)}, {"$set": {"provision_status": "in_progress"}}) # 2. Dynamically determine next Port and Password (Proven working logic) existing_tenants = Instance.list() used_ports = [t['port'] for t in existing_tenants if 'port' in t] next_port = max(used_ports) + 1 if used_ports else 10002 admin_password = generate_secure_password() # 3. Create Tenant Shell success = Instance.new(subdomain, next_port, admin_password) if not success: raise Exception("Tenant shell creation failed (Instance.new returned False)") # 4. Enable selected Modules # Always enable "starter" as a baseline if needed, then the booked ones if "starter" not in modules_to_provision: Instance.edit(subdomain, "starter") for mod in modules_to_provision: Instance.edit(subdomain, mod) # 5. Register Instance in Database domain = f"{subdomain}.invario-software.de" instance_doc = { "owner_id": user_id, "school_name": school_name, "subdomain": subdomain, "domain": domain, "https_port": next_port, "status": "ready", "nginx_status": "active", "admin_username": "admin", "admin_password": admin_password } inst_client, inst_col = _get_collection("instances") try: created_inst_id = str(inst_col.insert_one(instance_doc).inserted_id) finally: inst_client.close() # 6. Send Email Notification try: if user_email: send_accreditation_email( recipient=user_email, domain=domain, username="admin", password=admin_password, invoice_pdf_path=f"/var/invoices/{subdomain}_invoice.pdf", contract_pdf_path=f"/var/contracts/{subdomain}_contract.pdf" ) except Exception as mail_err: print(f"[WARNING] Email sending failed: {mail_err}") # 7. Mark Request as Ready for the Frontend 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 completed for {subdomain}") except Exception as e: print(f"[ERROR] Async Provisioning failed for {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" if not packages_str: 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: flash(f"Ungültige Buchungsoption: {pkg}", "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), "contact_email": _sanitize_text(request.values.get("contact_email") or "", 160).lower(), "contact_phone": _sanitize_text(request.values.get("contact_phone") or "", 60), "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), "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 Logic & Validation --- required_fields = { "contact_person": "Ansprechpartner", "school_name": "Schule/Einrichtung", "contact_email": "E-Mail", "billing_street": "Straße", "billing_zip": "PLZ", "billing_city": "Ort", } missing_labels = [label for key, label in required_fields.items() if not form_data.get(key)] if missing_labels: flash(f"Bitte folgende Pflichtfelder ausfüllen: {', '.join(missing_labels)}.", "error") return render_template("schnell_buchung.html", **review_data) if not form_data["agreements_accepted"]: flash("Bitte Nutzungsbedingungen, Lizenzvereinbarung und Datenschutzerklärung bestätigen.", "error") return render_template("schnell_buchung.html", **review_data) if not _validate_email(form_data["contact_email"]): flash("Bitte eine gültige E-Mail-Adresse angeben.", "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 message = ( f"{flow_config['message_prefix']} {selected_package} {flow_config['message_suffix']}\n" f"Ansprechpartner: {form_data['contact_person']}\n" f"Schule: {form_data['school_name']}\n" f"E-Mail: {form_data['contact_email']}\n" f"Telefon: {form_data['contact_phone'] or '-'}\n" f"Rechnungsadresse: {form_data['billing_street']}, {form_data['billing_zip']} {form_data['billing_city']}\n" f"Notizen: {form_data['notes'] or '-'}" ) try: # FLOW 1: Beratung / Gesprächs-Buchung 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": message, "created_at": _utc_now_iso(), "booking_flow": booking_flow, "booking_package": package_raw, "booking_data": form_data, } ) finally: client.close() flash(f"Gesprächstermin für {selected_package} wurde an das Team gesendet.", "success") return redirect(url_for("user_chat")) # FLOW 2: Kostenpflichtige Direkt-Buchung mit Provisioning if booking_flow == "payment": booking_number = f"BOOK-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}" raw_base = form_data.get("school_name") or session.get("username") or booking_number base = _slugify_subdomain(raw_base).lower()[:51] # Subdomain-Prüfung über das korrekte Instance Model try: existing_tenants = Instance.list() or [] existing_subdomains = [ t.get("subdomain") or t.get("tenant_slug") for t in existing_tenants if isinstance(t, dict) ] except Exception as e: print(f"[WARNING] Konnte bestehende Instanzen nicht abfragen: {e}") existing_subdomains = [] if base in existing_subdomains: flash("Die gewünschte Subdomain (aus dem Schulnamen) existiert bereits. Bitte wähle einen anderen Schulnamen.", "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() valid_modules = set(BOOKING_PACKAGE_MAP.keys()) modules_to_provision = {pkg for pkg in package_list if pkg in valid_modules} # Flask App Context für den Thread sicher abrufen app_obj = app user_id = session.get("user_id") user_email = session.get("email") or form_data.get("contact_email") try: thread = threading.Thread( target=background_booking_provisioning, args=(app_obj, prov_id, subdomain, modules_to_provision, user_id, form_data["school_name"], user_email) ) thread.start() print(f"[INFO] Background provisioning gestartet für Subdomain: {subdomain}") except Exception as thread_err: print(f"[ERROR] Fehler beim Starten des Background Threads: {thread_err}") flash(f"Buchung für {selected_package} wurde erfasst. Die Instanz wird jetzt eingerichtet.", "success") return redirect(url_for("my_instance_management")) except PyMongoError as pe: print(f"[ERROR] Database error in booking_payment: {pe}") flash("Buchungsanfrage konnte nicht gesendet werden. Bitte versuchen Sie es erneut.", "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', 'POST']) @admin_required def admin_instances(): version_options = _parse_instance_version_options() available_users = _list_available_instance_users() if request.method == 'POST': action = _sanitize_text(request.form.get("action") or "create", 20).lower() posted_subdomain = _sanitize_text(request.form.get("subdomain") or "", 120) school_name = _sanitize_text(request.form.get("school_name") or "", 120) raw_subdomain = _sanitize_text(request.form.get("subdomain") or "", 120) owner_username = _sanitize_text(request.form.get("owner_username") or "", 80) app_image_tag = _sanitize_text(request.form.get("app_image_tag") or "latest", 80) library_enabled = (request.form.get("library_enabled") or "").strip().lower() in {"1", "on", "true", "yes"} subdomain = _slugify_subdomain(posted_subdomain or raw_subdomain or school_name) if action == "reload_nginx": reload_ok, reload_message = _reload_host_nginx() if not reload_ok: lowered = (reload_message or "").lower() if "nicht gefunden" in lowered or "not found" in lowered or "kein nginx-reload-befehl" in lowered: flash("Host-Nginx kann aus dem Website-Container nicht direkt neu geladen werden.", "error") flash(_host_reload_hint(), "info") flash("Nach erfolgreichem Host-Reload bitte 'Reload auf Host bestätigt' ausführen.", "info") else: flash(reload_message or "Nginx-Reload fehlgeschlagen.", "error") return redirect(url_for("admin_instances")) updated_rows = _promote_manual_nginx_status( "Nginx erfolgreich neu geladen (Admin-Aktion)." ) flash("Nginx wurde erfolgreich neu geladen.", "success") if updated_rows > 0: flash(f"{updated_rows} Instanz(en) von manual_required auf ok gesetzt.", "info") else: flash("Keine Instanz mit nginx_status=manual_required gefunden.", "info") if reload_message: flash(reload_message, "info") return redirect(url_for("admin_instances")) if action == "confirm_nginx_reload": updated_rows = _promote_manual_nginx_status( "Host-Nginx manuell neu geladen und in Admin bestätigt." ) if updated_rows > 0: flash(f"{updated_rows} Instanz(en) von manual_required auf ok gesetzt.", "success") else: flash("Keine Instanz mit nginx_status=manual_required gefunden.", "info") return redirect(url_for("admin_instances")) if action == "delete": if not _is_valid_subdomain(subdomain): flash("Ungültige Subdomain für Löschung.", "error") return redirect(url_for("admin_instances")) existing_record = _get_school_instance_by_subdomain(subdomain) target_dir = _instance_dir_path(subdomain) dir_exists = bool(target_dir and os.path.isdir(target_dir)) if not existing_record and not dir_exists: flash("Instanz nicht gefunden.", "error") return redirect(url_for("admin_instances")) delete_ok, delete_message = _delete_instance_stack(subdomain) if not delete_ok: flash(delete_message or "Instanz konnte nicht gelöscht werden.", "error") return redirect(url_for("admin_instances")) if existing_record and not _delete_school_instance(subdomain): flash( "Instanz wurde technisch gelöscht, aber der Datenbankeintrag konnte nicht entfernt werden.", "error", ) return redirect(url_for("admin_instances")) flash(f"Instanz {subdomain} wurde gelöscht.", "success") if delete_message: flash(delete_message, "info") return redirect(url_for("admin_instances")) if action == "toggle_library": target = _get_school_instance_by_subdomain(subdomain) if not target: flash("Instanz nicht gefunden.", "error") return redirect(url_for("admin_instances")) instance_dir = _resolve_instance_dir(subdomain) if not instance_dir: flash("Instanzverzeichnis nicht gefunden.", "error") return redirect(url_for("admin_instances")) target_enabled = (request.form.get("library_enabled") or "").strip().lower() in {"1", "on", "true", "yes"} ok, message = _set_instance_library_enabled(instance_dir, target_enabled) if not ok: _upsert_school_instance( { "subdomain": subdomain, "status": "Fehler", "nginx_status": "error", "last_message": message, } ) flash(message, "error") return redirect(url_for("admin_instances")) restart_ok, restart_output = _restart_instance_stack(instance_dir) client = None try: client, col = _get_collection("school_instances") col.update_one( {"subdomain": subdomain}, { "$set": { "school_name": target.get("school_name") or "", "owner_username": target.get("owner_username") or "", "subdomain": subdomain, "domain": target.get("domain") or f"{subdomain}.{INSTANCE_PARENT_DOMAIN}", "https_port": int(target.get("https_port") or 0), "instance_dir": instance_dir, "app_image_tag": target.get("app_image_tag") or "latest", "library_enabled": target_enabled, "status": "Läuft" if restart_ok else "Fehler", "nginx_status": "ok" if restart_ok else "error", "last_message": "Bibliothek aktiviert/deaktiviert und Instanz neu gestartet." if restart_ok else _tail_output(restart_output, 18), "updated_at": _utc_now_iso(), }, "$setOnInsert": {"created_at": _utc_now_iso()}, }, upsert=True, ) except PyMongoError: pass finally: if client: client.close() if restart_ok: flash(f"Bibliothek wurde {'aktiviert' if target_enabled else 'deaktiviert'} und die Instanz neu gestartet.", "success") if restart_output: flash(_tail_output(restart_output, 12), "info") else: flash(f"Bibliothek wurde gespeichert, aber der Neustart ist fehlgeschlagen.\n{_tail_output(restart_output, 18)}", "error") return redirect(url_for("admin_instances")) if action not in {"create", "start"}: flash("Ungültige Aktion für Instanzverwaltung.", "error") return redirect(url_for("admin_instances")) if not available_users: flash("Keine freien Nutzer verfügbar. Bitte zuerst einen neuen Nutzer anlegen oder eine bestehende Instanz löschen.", "error") return redirect(url_for("admin_instances")) if not school_name: flash("Bitte einen Schulnamen angeben.", "error") return redirect(url_for("admin_instances")) if not owner_username: flash("Bitte einen Nutzer zuweisen.", "error") return redirect(url_for("admin_instances")) owner_doc = _find_user(owner_username) if not owner_doc: flash("Ausgewählter Nutzer wurde nicht gefunden.", "error") return redirect(url_for("admin_instances")) if not school_name: school_name = _sanitize_text(owner_doc.get("display_name") or owner_username, 120) if app_image_tag not in version_options: flash("Ungültige Version ausgewählt.", "error") return redirect(url_for("admin_instances")) existing_for_owner = _get_instance_for_user(owner_username, "") if existing_for_owner: existing_sub = _sanitize_text(existing_for_owner.get("subdomain") or "", 63) if existing_sub and existing_sub != subdomain: flash( f"Nutzer {owner_username} ist bereits der Instanz {existing_sub} zugewiesen. " "Bitte erst diese Zuweisung ändern.", "error", ) return redirect(url_for("admin_instances")) if not _is_valid_subdomain(subdomain): flash("Ungültige Subdomain. Erlaubt sind a-z, 0-9 und Bindestriche (3-63 Zeichen).", "error") return redirect(url_for("admin_instances")) success, message, details = _run_instance_provision( action, school_name, subdomain, app_image_tag=app_image_tag, library_enabled=library_enabled, ) instance_data = { "school_name": school_name, "owner_username": owner_username, "subdomain": details.get("SUBDOMAIN") or subdomain, "domain": details.get("DOMAIN") or f"{subdomain}.{INSTANCE_PARENT_DOMAIN}", "https_port": int((details.get("HTTPS_PORT") or "0") or 0), "instance_dir": details.get("INSTANCE_DIR") or os.path.join(INSTANCE_BASE_DIR, subdomain), "app_image_tag": details.get("APP_IMAGE_TAG") or app_image_tag, "library_enabled": library_enabled if details.get("LIBRARY_ENABLED") is None else details.get("LIBRARY_ENABLED") == "1", "status": "Läuft" if success else "Fehler", "nginx_status": details.get("NGINX_STATUS") or ("ok" if success else "error"), "last_message": message, } _upsert_school_instance(instance_data) if success: flash(f"Instanz gestartet: {instance_data['domain']}", "success") if message: flash(message, "info") else: flash(message or "Instanz konnte nicht gestartet werden.", "error") return redirect(url_for("admin_instances")) instances = _list_school_instances() instance_dashboard = _build_instance_dashboard(instances) return render_template( "admin_instances.html", instances=instances, users=_list_users_for_admin(), available_users=available_users, instance_dashboard=instance_dashboard, version_options=version_options, instance_repo_url=INSTANCE_REPO_URL, parent_domain=INSTANCE_PARENT_DOMAIN, base_dir=INSTANCE_BASE_DIR, provision_script=INSTANCE_PROVISION_SCRIPT, ) @app.route('/admin/instances/stats') @admin_required def admin_instances_stats(): instances = _list_school_instances() return jsonify(_collect_runtime_stats(instances)) @app.route('/admin/system', methods=['GET', 'POST']) @admin_required def admin_system_tools(): if request.method == 'POST': action = _sanitize_text(request.form.get("action") or "", 40) subdomain = _sanitize_text(request.form.get("subdomain") or "", 63) if action == "restart_core": compose_file = os.path.join(BASE_DIR, "docker-compose.yml") ok, output = _run_command( ["docker", "compose", "-f", compose_file, "restart", "website", "mongodb"], cwd=BASE_DIR, timeout=180, ) flash("Core-Services wurden neugestartet." if ok else f"Core-Restart fehlgeschlagen:\n{_tail_output(output, 10)}", "success" if ok else "error") return redirect(url_for("admin_system_tools")) if action in {"backup_instance", "update_instance", "restart_instance"}: instance_dir = _resolve_instance_dir(subdomain) if not instance_dir: flash("Instanz nicht gefunden oder ungültige Subdomain.", "error") return redirect(url_for("admin_system_tools")) if action == "backup_instance": command = ["bash", "./backup.sh", "--mode", "auto"] timeout = 1800 title = f"Backup für {subdomain}" elif action == "update_instance": command = ["bash", "./update.sh"] timeout = 2400 title = f"Update für {subdomain}" else: title = f"Restart für {subdomain}" ok, output = _restart_instance_stack(instance_dir) if ok: flash(f"{title} erfolgreich.\n{_tail_output(output, 12)}", "success") _upsert_school_instance( { "subdomain": subdomain, "status": "Läuft", "nginx_status": "ok", "last_message": "Instanz über System-Tools neu gestartet.", } ) else: flash(f"{title} fehlgeschlagen.\n{_tail_output(output, 18)}", "error") _upsert_school_instance( { "subdomain": subdomain, "status": "Fehler", "nginx_status": "error", "last_message": _sanitize_text(_tail_output(output, 6), 500), } ) return redirect(url_for("admin_system_tools")) ok, output = _run_command(command, cwd=instance_dir, timeout=timeout) if ok: flash(f"{title} erfolgreich.\n{_tail_output(output, 12)}", "success") else: flash(f"{title} fehlgeschlagen.\n{_tail_output(output, 18)}", "error") return redirect(url_for("admin_system_tools")) if action == "create_instance_admin": username = _sanitize_text(request.form.get("admin_username") or "", 80) password = (request.form.get("admin_password") or "").strip() first_name = _sanitize_text(request.form.get("admin_first_name") or "Admin", 80) last_name = _sanitize_text(request.form.get("admin_last_name") or "User", 80) ok, msg = _create_instance_admin_user(subdomain, username, password, first_name, last_name) flash(msg, "success" if ok else "error") return redirect(url_for("admin_system_tools")) flash("Unbekannte System-Aktion.", "error") return redirect(url_for("admin_system_tools")) instances = _list_school_instances() system_snapshot = _build_server_management_snapshot(instances) return render_template("admin_system.html", instances=instances, system_snapshot=system_snapshot) @app.route('/admin/system/stats') @admin_required def admin_system_stats(): instances = _list_school_instances() return jsonify(_build_server_management_snapshot(instances)) @app.route('/admin/system/logs/live') @admin_required def admin_system_live_logs(): return jsonify(_collect_homepage_service_logs()) @app.route('/admin/system/logs/core') @admin_required def admin_download_core_logs(): ok, output = _collect_core_logs() if not ok: flash(f"Core-Logs konnten nicht geladen werden.\n{_tail_output(output, 14)}", "error") return redirect(url_for("admin_system_tools")) filename = f"core-logs-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}.log" return send_file( BytesIO(output.encode("utf-8")), mimetype="text/plain", as_attachment=True, download_name=filename, ) @app.route('/admin/system/logs/instance/') @admin_required def admin_download_instance_logs(subdomain): ok, output = _collect_instance_logs(subdomain) if not ok: flash(f"Instanz-Logs konnten nicht geladen werden.\n{_tail_output(output, 14)}", "error") return redirect(url_for("admin_system_tools")) safe_name = _slugify_subdomain(subdomain) filename = f"instance-{safe_name}-logs-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}.log" return send_file( BytesIO(output.encode("utf-8")), mimetype="text/plain", as_attachment=True, download_name=filename, ) @app.route('/admin/system/backup/export/') @admin_required def admin_export_instance_backup(subdomain): ok, filename, archive_path = _build_instance_backup_archive(subdomain) if not ok or not archive_path: flash(filename or "Backup-Export fehlgeschlagen.", "error") return redirect(url_for("admin_system_tools")) @after_this_request def _cleanup_archive(response): try: os.remove(archive_path) except OSError: pass return response return send_file( archive_path, mimetype="application/gzip", as_attachment=True, download_name=filename, ) @app.route('/admin/system/backup/import/', methods=['POST']) @admin_required def admin_import_instance_backup(subdomain): instance_dir = _resolve_instance_dir(subdomain) if not instance_dir: flash("Instanzverzeichnis nicht gefunden.", "error") return redirect(url_for("admin_system_tools")) upload = request.files.get("backup_file") if not upload or not upload.filename: flash("Bitte eine Backup-Datei auswählen.", "error") return redirect(url_for("admin_system_tools")) filename = secure_filename(upload.filename) if not (filename.endswith(".tar.gz") or filename.endswith(".tgz")): flash("Nur .tar.gz oder .tgz Backups sind erlaubt.", "error") return redirect(url_for("admin_system_tools")) fd, temp_path = tempfile.mkstemp(prefix="instance-restore-", suffix=".tar.gz") os.close(fd) try: upload.save(temp_path) ok, message = _safe_extract_tar_archive(temp_path, instance_dir) if not ok: flash(message, "error") return redirect(url_for("admin_system_tools")) restart_ok, restart_output = _restart_instance_stack(instance_dir) if restart_ok: flash(f"Backup für {subdomain} erfolgreich eingespielt und Instanz neu gestartet.", "success") if restart_output: flash(_tail_output(restart_output, 10), "info") else: flash(f"Backup eingespielt, aber Neustart fehlgeschlagen:\n{_tail_output(restart_output, 14)}", "error") except Exception as exc: flash(f"Backup konnte nicht eingespielt werden: {exc}", "error") finally: try: os.remove(temp_path) except OSError: pass return redirect(url_for("admin_system_tools")) @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/') 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)) 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")) 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) current_instance = _get_instance_for_user(username, display_name) current_subdomain = _sanitize_text((current_instance or {}).get("subdomain") or "", 63) suggested_subdomain = _slugify_subdomain(display_name or username) 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")) instance_doc = _get_instance_for_user(username, display_name) instance_view = 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 "", } return render_template( "my_instance.html", instance=instance_view, 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")) 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")) 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() def run_tenant_provisioning_async(app, user_id, tenant_slug, school_name, user_email, prov_id): with app.app_context(): # Hilfsfunktion zum Aktualisieren des Status def _update_prov_status(status, extra_fields=None): req_client, req_col = _get_collection("instance_requests") try: update_data = { "provision_status": status, "updated_at": _utc_now_iso() } if extra_fields: update_data.update(extra_fields) req_col.update_one( {"_id": ObjectId(prov_id)}, {"$set": update_data} ) finally: req_client.close() # 1. Status auf "in_progress" setzen _update_prov_status("in_progress") # GESAMTEN PROZESS IN TRY-EXCEPT PACKEN try: # 2. Port dynamisch ermitteln existing_tenants = Instance.list() used_ports = [t['port'] for t in existing_tenants if 'port' in t] next_port = max(used_ports) + 1 if used_ports else 10002 admin_password = generate_secure_password() # 3. Skript für Tenant-Erstellung ausführen success = Instance.new(tenant_slug, next_port, admin_password) if not success: _update_prov_status("failed", {"last_error": "Tenant shell creation failed"}) return # Module aktivieren Instance.edit(tenant_slug, "starter") # 4. Instanz in der Datenbank anlegen domain = f"{tenant_slug}.invario-software.de" instance_doc = { "owner_id": user_id, "school_name": school_name, "subdomain": tenant_slug, "domain": domain, "https_port": next_port, "status": "ready", "nginx_status": "active", "admin_username": "admin", "admin_password": admin_password } inst_client, inst_col = _get_collection("instances") try: inst_result = inst_col.insert_one(instance_doc) created_inst_id = str(inst_result.inserted_id) finally: inst_client.close() # 5. E-Mail versenden (wird vor dem Status-Update gemacht, damit bei Crash 'failed' ausgelöst wird) send_accreditation_email( recipient=user_email, domain=domain, username="admin", password=admin_password, invoice_pdf_path=f"/var/invoices/{tenant_slug}_invoice.pdf", contract_pdf_path=f"/var/contracts/{tenant_slug}_contract.pdf" ) # 6. Request als erfolgreich markieren (WICHTIG: Status "ready" und exakte Felder für Frontend) _update_prov_status("ready", { "instance_id": created_inst_id, "url": f"https://{domain}", "username": "admin", "password": admin_password # Optional: "invoice_url": f"/download/invoice/{tenant_slug}" falls du den PDF-Download im Frontend anbietest }) except Exception as e: # Wenn irgendetwas im Block abstürzt, Frontend informieren! print(f"[ERROR] Provisioning fehlgeschlagen für {tenant_slug}:") traceback.print_exc() _update_prov_status("failed", {"last_error": str(e)}) @app.route('/booking_payment_async', methods=['POST']) @login_required def booking_payment_async(): try: user_id = session.get("user_id") user_email = session.get("email") tenant_slug = _slugify_subdomain(request.form.get('tenant_slug')) school_name = request.form.get('school_name', tenant_slug) if not tenant_slug: return jsonify({"success": False, "error": "Ungültige Subdomain angegeben."}), 400 # Datenbank-Eintrag erstellen... req_client, req_col = _get_collection("instance_requests") try: prov_doc = { "user_id": user_id, "tenant_slug": tenant_slug, "school_name": school_name, "provision_status": "pending", "created_at": _utc_now_iso(), "updated_at": _utc_now_iso() } prov_id = str(req_col.insert_one(prov_doc).inserted_id) finally: req_client.close() # Asynchronen Thread starten... thread = threading.Thread( target=run_tenant_provisioning_async, args=(app, user_id, tenant_slug, school_name, user_email, prov_id) ) thread.start() return jsonify({ "success": True, "tenant_slug": tenant_slug, "request_id": prov_id }), 200 except Exception as e: # WICHTIG: Fängt jegliche Python-Fehler ab und sendet sauberes JSON statt HTML-Crash return jsonify({"success": False, "error": str(e)}), 500 @app.route('/instance_request/status') def tenant_status(): 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() current_status = prov_req.get("provision_status") if prov_req else None if not prov_req or 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 target_url = f"https://{instance['domain']}/health" try: response = requests.get(target_url, timeout=3) if response.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 def main(): app.run(host="0.0.0.0", port=4999, debug=False) if __name__ == "__main__": main()