import os import shutil import uuid import threading import io import logging import traceback import hashlib import hmac import secrets from urllib.parse import quote from datetime import datetime, date from pathlib import Path from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile from fastapi.responses import RedirectResponse, JSONResponse, StreamingResponse, Response from starlette.exceptions import HTTPException as StarletteHTTPException from starlette.middleware.sessions import SessionMiddleware from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from sqlalchemy.orm import Session, joinedload from sqlalchemy.exc import IntegrityError from .database import Base, engine, get_db, SessionLocal from .fields import ASSET_FIELDS, COMPUTER_FIELDS, DEFAULT_FIELDS from .models import Asset, AssetHistory, Category, CategoryField, StatusOption, SyncRun, User from .config import load_config, public_config, save_config from .meshcentral import synchronize from .migrations import apply_lightweight_migrations from openpyxl import Workbook, load_workbook from openpyxl.styles import Font, PatternFill, Alignment BASE_DIR = Path(__file__).resolve().parent UPLOAD_DIR = BASE_DIR / "static" / "uploads" UPLOAD_DIR.mkdir(parents=True, exist_ok=True) LOG_DIR = Path(os.getenv("SYNC_LOG_DIR", "/app/data/logs/sync")) LOG_DIR.mkdir(parents=True, exist_ok=True) APP_LOG_DIR = Path(os.getenv("APP_LOG_DIR", "/app/data/logs")) APP_LOG_DIR.mkdir(parents=True, exist_ok=True) logger = logging.getLogger("assetmanager") logger.setLevel(logging.INFO) if not logger.handlers: file_handler = logging.FileHandler(APP_LOG_DIR / "errors.log", encoding="utf-8") file_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")) logger.addHandler(file_handler) ldap_logger = logging.getLogger("assetmanager.ldap") ldap_logger.setLevel(logging.DEBUG) ldap_logger.propagate = False if not ldap_logger.handlers: ldap_file_handler = logging.FileHandler(APP_LOG_DIR / "ldap.log", encoding="utf-8") ldap_file_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s")) ldap_logger.addHandler(ldap_file_handler) ldap_console_handler = logging.StreamHandler() ldap_console_handler.setFormatter(logging.Formatter("LDAP | %(levelname)s | %(message)s")) ldap_logger.addHandler(ldap_console_handler) app = FastAPI(title=os.getenv("APP_TITLE", "AssetManager")) _session_cfg = load_config().get("authentication", {}) _session_env = _session_cfg.get("session_secret_env", "SESSION_SECRET") _session_secret = os.getenv(_session_env) or os.getenv("SESSION_SECRET") or "assetmanager-change-this-session-secret" app.add_middleware(SessionMiddleware, secret_key=_session_secret, same_site="lax", https_only=False) app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static") templates = Jinja2Templates(directory=BASE_DIR / "templates") templates.env.globals["application_config"] = load_config SYNC_JOBS: dict[str, dict] = {} SYNC_JOBS_LOCK = threading.Lock() def _hash_password(password: str) -> str: salt = secrets.token_bytes(16) iterations = 310_000 digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations) return f"pbkdf2_sha256${iterations}${salt.hex()}${digest.hex()}" def _verify_password(password: str, encoded: str | None) -> bool: if not encoded: return False try: algorithm, iterations_text, salt_hex, digest_hex = encoded.split("$", 3) if algorithm != "pbkdf2_sha256": return False candidate = hashlib.pbkdf2_hmac( "sha256", password.encode("utf-8"), bytes.fromhex(salt_hex), int(iterations_text) ) return hmac.compare_digest(candidate.hex(), digest_hex) except (ValueError, TypeError): return False def _session_user(request: Request) -> dict | None: value = request.session.get("user") return value if isinstance(value, dict) else None def _changed_by(request: Request) -> str | None: user = _session_user(request) return (user.get("display_name") or user.get("username")) if user else None def _login_user(request: Request, user: User) -> None: request.session["user"] = { "id": user.id, "username": user.username, "display_name": user.display_name or user.username, "email": user.email or "", "auth_source": user.auth_source, "is_admin": bool(user.is_admin), "allowed_mesh_groups": user.allowed_mesh_groups or [], } def _is_admin(request: Request) -> bool: user = _session_user(request) return bool(user and user.get("is_admin")) def _require_admin(request: Request) -> None: if not _is_admin(request): raise HTTPException(403, "Diese Funktion ist nur für Administratoren verfügbar.") def _allowed_groups(request: Request) -> list[str] | None: user = _session_user(request) if not user or user.get("is_admin"): return None return [str(x) for x in (user.get("allowed_mesh_groups") or []) if str(x).strip()] def _apply_asset_access(query, request: Request): groups = _allowed_groups(request) return query if groups is None else query.filter(Asset.mesh_group.in_(groups)) def _get_visible_asset(db: Session, request: Request, asset_id: int) -> Asset: query = _apply_asset_access(db.query(Asset), request) asset = query.filter(Asset.id == asset_id).first() if not asset: raise HTTPException(404, "Asset nicht gefunden oder keine Berechtigung") return asset def _action_status(db: Session, action: str) -> StatusOption | None: column = StatusOption.use_for_issue if action == "issue" else StatusOption.use_for_return return db.query(StatusOption).filter(StatusOption.active.is_(True), column.is_(True)).first() def _ldap_authenticate(username: str, password: str, auth_config: dict) -> dict | None: from ldap3 import ALL, Connection, Server, SUBTREE from ldap3.core.exceptions import LDAPException from ldap3.utils.conv import escape_filter_chars ldap = auth_config.get("ldap", {}) server_name = str(ldap.get("server", "")).strip() base_dn = str(ldap.get("user_base_dn", "")).strip() use_ssl = bool(ldap.get("use_ssl", False)) start_tls = bool(ldap.get("start_tls", False)) port = int(ldap.get("port") or (636 if use_ssl else 389)) bind_dn = str(ldap.get("bind_dn", "")).strip() env_name = str(ldap.get("bind_password_env", "LDAP_BIND_PASSWORD")).strip() bind_password = os.getenv(env_name, "") if env_name else "" if not bind_password: bind_password = str(ldap.get("bind_password", "")) ldap_logger.info( "Anmeldeversuch Benutzer=%r Server=%s Port=%s SSL=%s StartTLS=%s BaseDN=%r BindDN=%r Passwortvariable=%r gesetzt=%s", username, server_name, port, use_ssl, start_tls, base_dn, bind_dn, env_name, bool(bind_password), ) if not server_name: ldap_logger.error("LDAP-Server ist nicht konfiguriert") return None if not base_dn: ldap_logger.error("Benutzer-Basis-DN ist nicht konfiguriert") return None if not password: ldap_logger.warning("Leeres Benutzerpasswort für Benutzer=%r", username) return None if bind_dn and not bind_password: ldap_logger.error("Bind-DN ist gesetzt, aber das Bind-Passwort fehlt. Erwartete Umgebungsvariable=%r", env_name) return None server = Server(server_name, port=port, use_ssl=use_ssl, get_info=ALL, connect_timeout=10) search_connection = None user_connection = None try: search_connection = Connection( server, user=bind_dn or None, password=bind_password or None, auto_bind=False, receive_timeout=15, raise_exceptions=False, ) if not search_connection.open(): ldap_logger.error("Verbindung zum LDAP-Server fehlgeschlagen: result=%r last_error=%r", search_connection.result, search_connection.last_error) return None ldap_logger.debug("TCP/LDAP-Verbindung geöffnet") if start_tls and not use_ssl: if not search_connection.start_tls(): ldap_logger.error("StartTLS fehlgeschlagen: result=%r last_error=%r", search_connection.result, search_connection.last_error) return None ldap_logger.debug("StartTLS erfolgreich") if not search_connection.bind(): ldap_logger.error("Dienstkonto-Bind fehlgeschlagen: result=%r last_error=%r", search_connection.result, search_connection.last_error) return None ldap_logger.info("Dienstkonto-Bind erfolgreich") raw_filter = str(ldap.get("user_filter", "(sAMAccountName={username})")) escaped_username = escape_filter_chars(username) user_filter = raw_filter.replace("{username}", escaped_username).replace("{{username}}", escaped_username) display_attr = str(ldap.get("display_name_attribute", "displayName")) email_attr = str(ldap.get("email_attribute", "mail")) attributes = list(dict.fromkeys([display_attr, email_attr, "distinguishedName"])) ldap_logger.info("Benutzersuche BaseDN=%r Filter=%r Attribute=%r", base_dn, user_filter, attributes) search_ok = search_connection.search( base_dn, user_filter, search_scope=SUBTREE, attributes=attributes, ) ldap_logger.info( "Benutzersuche beendet: success=%s Treffer=%s result=%r last_error=%r", search_ok, len(search_connection.entries), search_connection.result, search_connection.last_error, ) if len(search_connection.entries) != 1: if len(search_connection.entries) == 0: ldap_logger.warning("Kein LDAP-Benutzer für Benutzer=%r gefunden", username) else: ldap_logger.warning("Mehrere LDAP-Benutzer für Benutzer=%r gefunden: %s", username, len(search_connection.entries)) return None entry = search_connection.entries[0] user_dn = entry.entry_dn display_name = str(getattr(entry, display_attr, "") or username) email = str(getattr(entry, email_attr, "") or "") ldap_logger.info("Benutzer gefunden: DN=%r Anzeigename=%r E-Mail=%r", user_dn, display_name, email) user_connection = Connection( server, user=user_dn, password=password, auto_bind=False, receive_timeout=15, raise_exceptions=False, ) if not user_connection.open(): ldap_logger.error("Verbindung für Benutzer-Bind fehlgeschlagen: result=%r last_error=%r", user_connection.result, user_connection.last_error) return None if start_tls and not use_ssl: if not user_connection.start_tls(): ldap_logger.error("StartTLS beim Benutzer-Bind fehlgeschlagen: result=%r last_error=%r", user_connection.result, user_connection.last_error) return None if not user_connection.bind(): ldap_logger.warning("Benutzer-Bind fehlgeschlagen für DN=%r: result=%r last_error=%r", user_dn, user_connection.result, user_connection.last_error) return None ldap_logger.info("LDAP-Anmeldung erfolgreich für Benutzer=%r DN=%r", username, user_dn) return {"username": username, "display_name": display_name, "email": email} except LDAPException as exc: ldap_logger.exception("LDAP-Ausnahme für Benutzer=%r: %s", username, exc) return None except Exception as exc: ldap_logger.exception("Unerwarteter LDAP-Fehler für Benutzer=%r: %s", username, exc) return None finally: for connection in (user_connection, search_connection): if connection is not None: try: connection.unbind() except Exception: pass @app.middleware("http") async def authentication_gate(request: Request, call_next): mode = str(load_config().get("authentication", {}).get("mode", "none")) public_paths = {"/login", "/logout"} if mode in {"local", "ldap"} and not request.session.get("user") and request.url.path not in public_paths and not request.url.path.startswith("/static/"): if request.method == "GET": return RedirectResponse("/login?toast_warning=Bitte zuerst anmelden", status_code=303) return JSONResponse({"detail": "Anmeldung erforderlich"}, status_code=401) return await call_next(request) def _is_browser_request(request: Request) -> bool: accept = request.headers.get("accept", "") return "text/html" in accept or request.method != "GET" def _toast_redirect(request: Request, message: str, level: str = "error") -> RedirectResponse: referer = request.headers.get("referer") target = referer if referer and referer.startswith(str(request.base_url).rstrip("/")) else "/" separator = "&" if "?" in target else "?" return RedirectResponse(f"{target}{separator}toast_{level}={quote(message)}", status_code=303) @app.exception_handler(StarletteHTTPException) async def http_exception_handler(request: Request, exc: StarletteHTTPException): message = str(exc.detail) if exc.detail else f"HTTP-Fehler {exc.status_code}" logger.error("HTTP %s | %s %s | %s", exc.status_code, request.method, request.url.path, message) if _is_browser_request(request): return _toast_redirect(request, message) return JSONResponse({"detail": message}, status_code=exc.status_code) @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, exc: Exception): logger.error( "Unerwarteter Fehler | %s %s | %s\n%s", request.method, request.url.path, exc, traceback.format_exc(), ) if _is_browser_request(request): return _toast_redirect(request, "Ein unerwarteter Fehler ist aufgetreten. Details wurden protokolliert.") return JSONResponse({"detail": "Interner Serverfehler"}, status_code=500) def _job_log(job_id: str, level: str, message: str) -> None: entry = { "time": datetime.now().strftime("%H:%M:%S"), "level": level, "message": str(message), } with SYNC_JOBS_LOCK: job = SYNC_JOBS.get(job_id) if job is not None: job["logs"].append(entry) job["logs"] = job["logs"][-10000:] try: with (LOG_DIR / f"{job_id}.jsonl").open("a", encoding="utf-8") as handle: import json handle.write(json.dumps(entry, ensure_ascii=False) + "\n") except OSError: pass def _run_sync_job(job_id: str) -> None: db = SessionLocal() try: result = synchronize(db, lambda level, message: _job_log(job_id, level, message)) with SYNC_JOBS_LOCK: job = SYNC_JOBS[job_id] job["status"] = result.run.status job["run_id"] = result.run.id job["finished"] = True except Exception as exc: _job_log(job_id, "error", f"Unerwarteter Fehler: {exc}") with SYNC_JOBS_LOCK: job = SYNC_JOBS[job_id] job["status"] = "error" job["finished"] = True finally: db.close() @app.on_event("startup") def startup(): Base.metadata.create_all(bind=engine) apply_lightweight_migrations() from .database import SessionLocal db = SessionLocal() try: default_statuses = ["Aktiv", "Defekt", "im Lager", "Ausgegeben", "Verschrottet"] if db.query(StatusOption).count() == 0: for order, status_name in enumerate(default_statuses): db.add(StatusOption(name=status_name, sort_order=order, active=True)) db.commit() if not db.query(StatusOption).filter(StatusOption.use_for_issue.is_(True)).first(): item = db.query(StatusOption).filter(StatusOption.name == "Ausgegeben").first() if item: item.use_for_issue = True if not db.query(StatusOption).filter(StatusOption.use_for_return.is_(True)).first(): item = db.query(StatusOption).filter(StatusOption.name == "im Lager").first() if item: item.use_for_return = True db.commit() if db.query(Category).count() == 0: computer = Category(name="Computer", description="Computer und Notebooks aus MeshCentral oder manueller Erfassung") db.add(computer) db.flush() for order, field_name in enumerate(COMPUTER_FIELDS): db.add(CategoryField( category_id=computer.id, field_name=field_name, label=ASSET_FIELDS[field_name], active=True, required=field_name == "name", show_in_list=field_name in {"asset_tag", "name", "hostname", "operating_system", "os_patch_level", "mac_address", "status"}, sort_order=order, )) db.commit() else: computer = db.query(Category).options(joinedload(Category.visible_fields)).filter(Category.name == "Computer").first() if computer: existing = {field.field_name for field in computer.visible_fields} next_order = max((field.sort_order for field in computer.visible_fields), default=-1) + 1 important_list_fields = {"hostname", "manufacturer", "model", "serial_number", "operating_system", "ip_address", "mac_address", "storage_details", "gpu_name", "tpm_version", "antivirus", "mesh_group"} changed = False for field_name in COMPUTER_FIELDS: if field_name not in existing: db.add(CategoryField( category_id=computer.id, field_name=field_name, label=ASSET_FIELDS[field_name], active=True, required=False, show_in_list=field_name in important_list_fields, sort_order=next_order, )) next_order += 1 changed = True if changed: db.commit() finally: db.close() def delete_asset_image(image_path: str | None) -> None: """Löscht nur individuell hochgeladene Asset-Bilder aus dem Upload-Ordner.""" if not image_path or not image_path.startswith("/static/uploads/"): return filename = Path(image_path).name target = UPLOAD_DIR / filename try: target.unlink(missing_ok=True) except OSError: pass def save_upload(upload: UploadFile | None) -> str | None: if not upload or not upload.filename: return None suffix = Path(upload.filename).suffix.lower() if suffix not in {".png", ".jpg", ".jpeg", ".webp", ".gif"}: raise HTTPException(400, "Nur PNG, JPG, WEBP oder GIF sind erlaubt.") filename = f"{uuid.uuid4().hex}{suffix}" target = UPLOAD_DIR / filename with target.open("wb") as buffer: shutil.copyfileobj(upload.file, buffer) return f"/static/uploads/{filename}" def _display_value(value) -> str: if value is None: return "" if isinstance(value, (dict, list)): import json return json.dumps(value, ensure_ascii=False, sort_keys=True) return str(value) def _record_asset_history(db: Session, asset: Asset, changes: dict, source: str = "manual", changed_by: str | None = None) -> None: if not changes: return db.add(AssetHistory(asset_id=asset.id, source=source, changed_by=changed_by, changes=changes)) def _xlsx_response(filename: str, headers: list[str], rows: list[list]) -> Response: """Erzeugt eine echte XLSX-Datei und liefert vollständige Bytes statt eines Streams. Das verhindert beschädigte Downloads bei einzelnen Reverse-Proxies und Browsern. """ workbook = Workbook() sheet = workbook.active sheet.title = "Export" sheet.freeze_panes = "A2" sheet.auto_filter.ref = f"A1:{sheet.cell(row=1, column=max(1, len(headers))).column_letter}1" sheet.append(headers) header_fill = PatternFill("solid", fgColor="D9EAF7") for cell in sheet[1]: cell.font = Font(bold=True) cell.fill = header_fill cell.alignment = Alignment(vertical="top") for row in rows: sheet.append([_display_value(value) for value in row]) for column in sheet.columns: max_length = min(max((len(str(cell.value or "")) for cell in column), default=10) + 2, 60) sheet.column_dimensions[column[0].column_letter].width = max_length output = io.BytesIO() workbook.save(output) payload = output.getvalue() ascii_name = "".join(ch if ch.isascii() and (ch.isalnum() or ch in "._-") else "_" for ch in filename) disposition = f"attachment; filename={ascii_name}; filename*=UTF-8''{quote(filename)}" return Response( content=payload, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={ "Content-Disposition": disposition, "Content-Length": str(len(payload)), "X-Content-Type-Options": "nosniff", "Cache-Control": "no-store", }, ) def _import_field_map(category: Category) -> dict[str, str]: result: dict[str, str] = {"bezeichnung": "name", "name": "name", "kategorie": "category"} for field in category.visible_fields: if field.active and hasattr(Asset, field.field_name): result[field.label.strip().casefold()] = field.field_name result[field.field_name.strip().casefold()] = field.field_name return result def _excel_cell_value(value) -> str | None: if value is None: return None if isinstance(value, datetime): return value.strftime("%Y-%m-%d %H:%M:%S") if isinstance(value, date): return value.isoformat() text = str(value).strip() return text or None def _active_status_options(db: Session) -> list[StatusOption]: return db.query(StatusOption).filter(StatusOption.active.is_(True)).order_by(StatusOption.sort_order, StatusOption.name).all() def _asset_action_changes(asset: Asset, action: str, status_name: str, assigned_to: str | None = None, assigned_on: str | None = None) -> dict[str, dict[str, str]]: changes: dict[str, dict[str, str]] = {} if _display_value(asset.status) != status_name: changes[ASSET_FIELDS["status"]] = {"old": _display_value(asset.status), "new": status_name} asset.status = status_name if action == "issue": new_name = (assigned_to or "").strip() or None new_date = (assigned_on or date.today().isoformat()).strip() if _display_value(asset.assigned_to) != _display_value(new_name): changes[ASSET_FIELDS["assigned_to"]] = {"old": _display_value(asset.assigned_to), "new": _display_value(new_name)} asset.assigned_to = new_name if _display_value(asset.assigned_on) != new_date: changes[ASSET_FIELDS["assigned_on"]] = {"old": _display_value(asset.assigned_on), "new": new_date} asset.assigned_on = new_date else: if asset.assigned_to: changes[ASSET_FIELDS["assigned_to"]] = {"old": _display_value(asset.assigned_to), "new": ""} asset.assigned_to = None if asset.assigned_on: changes[ASSET_FIELDS["assigned_on"]] = {"old": _display_value(asset.assigned_on), "new": ""} asset.assigned_on = None return changes def _list_fields_for_categories(categories: list[Category]) -> list[CategoryField]: """Bildet eine eindeutige, sortierte Vereinigungsmenge der konfigurierten Listenspalten.""" result: list[CategoryField] = [] seen: set[str] = set() for category in categories: for field in sorted(category.visible_fields, key=lambda item: item.sort_order): if not field.active or not field.show_in_list or field.field_name in seen: continue result.append(field) seen.add(field.field_name) return result @app.get("/") def dashboard(request: Request, db: Session = Depends(get_db)): categories = db.query(Category).options(joinedload(Category.visible_fields)).all() assets = ( _apply_asset_access(db.query(Asset), request) .options(joinedload(Asset.category).joinedload(Category.visible_fields)) .order_by(Asset.updated_at.desc()) .limit(10) .all() ) recent_categories = [] seen_category_ids = set() for asset in assets: if asset.category_id not in seen_category_ids: recent_categories.append(asset.category) seen_category_ids.add(asset.category_id) fields = _list_fields_for_categories(recent_categories) return templates.TemplateResponse( "dashboard.html", {"request": request, "categories": categories, "assets": assets, "fields": fields, "is_admin": _is_admin(request), "issue_status_name": (_action_status(db, "issue").name if _action_status(db, "issue") else None)}, ) @app.get("/assets") def assets_list(request: Request, category_id: int | None = None, db: Session = Depends(get_db)): categories = db.query(Category).options(joinedload(Category.visible_fields)).all() query = _apply_asset_access(db.query(Asset), request).options(joinedload(Asset.category).joinedload(Category.visible_fields)) selected = None if category_id: selected = next((category for category in categories if category.id == category_id), None) if not selected: raise HTTPException(404, "Kategorie nicht gefunden") query = query.filter(Asset.category_id == category_id) assets = query.order_by(Asset.name).all() fields = ( [field for field in sorted(selected.visible_fields, key=lambda item: item.sort_order) if field.active and field.show_in_list] if selected else _list_fields_for_categories(categories) ) return templates.TemplateResponse( "assets.html", {"request": request, "assets": assets, "categories": categories, "selected": selected, "fields": fields, "is_admin": _is_admin(request), "issue_status_name": (_action_status(db, "issue").name if _action_status(db, "issue") else None)}, ) @app.get("/assets/new") def asset_new(request: Request, category_id: int | None = None, db: Session = Depends(get_db)): _require_admin(request) categories = db.query(Category).options(joinedload(Category.visible_fields)).all() selected = next((c for c in categories if c.id == category_id), categories[0] if categories else None) return templates.TemplateResponse("asset_form.html", {"request": request, "asset": None, "categories": categories, "selected": selected, "fields": ASSET_FIELDS, "status_options": _active_status_options(db), "duplicate_mode": False}) @app.get("/assets/{asset_id}/duplicate") def asset_duplicate(asset_id: int, request: Request, db: Session = Depends(get_db)): _require_admin(request) original = db.query(Asset).options(joinedload(Asset.category).joinedload(Category.visible_fields)).filter(Asset.id == asset_id).first() if not original: raise HTTPException(404, "Asset nicht gefunden") categories = db.query(Category).options(joinedload(Category.visible_fields)).all() class DuplicateValues: pass duplicate = DuplicateValues() for field_name in ASSET_FIELDS: setattr(duplicate, field_name, getattr(original, field_name, None)) duplicate.asset_tag = None duplicate.mesh_node_id = None duplicate.name = f"{original.name} – Kopie" return templates.TemplateResponse("asset_form.html", { "request": request, "asset": duplicate, "categories": categories, "selected": original.category, "fields": ASSET_FIELDS, "status_options": _active_status_options(db), "duplicate_mode": True, }) @app.post("/assets/new") async def asset_create(request: Request, category_id: int = Form(...), image: UploadFile | None = File(None), db: Session = Depends(get_db)): _require_admin(request) form = await request.form() category = db.query(Category).options(joinedload(Category.visible_fields)).filter(Category.id == category_id).first() if not category: raise HTTPException(404, "Kategorie nicht gefunden") data = {f.field_name: form.get(f.field_name) or None for f in category.visible_fields if f.active and hasattr(Asset, f.field_name)} if not data.get("name"): raise HTTPException(400, "Bezeichnung fehlt") asset = Asset(category_id=category_id, image_path=save_upload(image), **data) db.add(asset) try: db.flush() created_changes = {field.label: {"old": "", "new": _display_value(getattr(asset, field.field_name, None))} for field in category.visible_fields if field.active and getattr(asset, field.field_name, None) not in (None, "")} _record_asset_history(db, asset, created_changes, source="manual-create", changed_by=_changed_by(request)) db.commit() except IntegrityError as exc: db.rollback() raise HTTPException(400, "Asset konnte nicht gespeichert werden. Inventarnummer, Seriennummer oder MeshCentral Node-ID ist möglicherweise bereits vergeben.") from exc return RedirectResponse(f"/assets/{asset.id}", status_code=303) @app.get("/assets/import") def assets_import_form(request: Request, category_id: int | None = None, db: Session = Depends(get_db)): categories = db.query(Category).options(joinedload(Category.visible_fields)).order_by(Category.name).all() selected = next((category for category in categories if category.id == category_id), categories[0] if categories else None) return templates.TemplateResponse( "assets_import.html", {"request": request, "categories": categories, "selected": selected, "result": None}, ) @app.get("/assets/import/template.xlsx") def assets_import_template(request: Request, category_id: int, db: Session = Depends(get_db)): _require_admin(request) category = db.query(Category).options(joinedload(Category.visible_fields)).filter(Category.id == category_id).first() if not category: raise HTTPException(404, "Kategorie nicht gefunden") fields = [field for field in sorted(category.visible_fields, key=lambda item: item.sort_order) if field.active and hasattr(Asset, field.field_name)] if not any(field.field_name == "name" for field in fields): headers = ["Bezeichnung"] + [field.label for field in fields] else: headers = [field.label for field in fields] return _xlsx_response(f"asset-import-{category.name}.xlsx", headers, []) @app.post("/assets/import") async def assets_import( request: Request, category_id: int = Form(...), file: UploadFile = File(...), duplicate_action: str = Form("skip"), db: Session = Depends(get_db), ): _require_admin(request) category = db.query(Category).options(joinedload(Category.visible_fields)).filter(Category.id == category_id).first() categories = db.query(Category).options(joinedload(Category.visible_fields)).order_by(Category.name).all() if not category: raise HTTPException(404, "Kategorie nicht gefunden") if not file.filename or Path(file.filename).suffix.lower() != ".xlsx": raise HTTPException(400, "Bitte eine XLSX-Datei auswählen.") try: raw = await file.read() workbook = load_workbook(io.BytesIO(raw), read_only=True, data_only=True) sheet = workbook.active except Exception as exc: logger.exception("Excel-Import konnte nicht geöffnet werden") raise HTTPException(400, "Die Excel-Datei ist ungültig oder beschädigt.") from exc rows_iter = sheet.iter_rows(values_only=True) header_row = next(rows_iter, None) if not header_row: raise HTTPException(400, "Die Excel-Datei enthält keine Kopfzeile.") field_map = _import_field_map(category) columns: list[str | None] = [] unknown_headers: list[str] = [] for header in header_row: label = _excel_cell_value(header) or "" field_name = field_map.get(label.casefold()) columns.append(field_name) if label and not field_name: unknown_headers.append(label) if "name" not in columns: raise HTTPException(400, "Die Spalte ‚Bezeichnung‘ bzw. ‚Name‘ fehlt.") created = 0 updated = 0 skipped = 0 errors: list[str] = [] changed_by = _changed_by(request) allowed_fields = {field.field_name for field in category.visible_fields if field.active and hasattr(Asset, field.field_name)} | {"name"} for excel_row_number, values in enumerate(rows_iter, start=2): data: dict[str, str | None] = {} for index, value in enumerate(values): if index >= len(columns): break field_name = columns[index] if field_name and field_name in allowed_fields: data[field_name] = _excel_cell_value(value) if not any(value not in (None, "") for value in data.values()): continue if not data.get("name"): errors.append(f"Zeile {excel_row_number}: Bezeichnung fehlt") continue existing = None if data.get("asset_tag"): existing = db.query(Asset).filter(Asset.asset_tag == data["asset_tag"]).first() if existing is None and data.get("serial_number"): existing = db.query(Asset).filter(Asset.serial_number == data["serial_number"]).first() if existing: if duplicate_action != "update": skipped += 1 continue changes: dict[str, dict[str, str]] = {} existing.category_id = category.id for field_name, value in data.items(): old_value = getattr(existing, field_name, None) if _display_value(old_value) != _display_value(value): changes[ASSET_FIELDS.get(field_name, field_name)] = {"old": _display_value(old_value), "new": _display_value(value)} setattr(existing, field_name, value) _record_asset_history(db, existing, changes, source="excel-import-update", changed_by=changed_by) updated += 1 else: try: with db.begin_nested(): asset = Asset(category_id=category.id, **data) db.add(asset) db.flush() created_changes = {ASSET_FIELDS.get(name, name): {"old": "", "new": _display_value(value)} for name, value in data.items() if value not in (None, "")} _record_asset_history(db, asset, created_changes, source="excel-import-create", changed_by=changed_by) created += 1 except IntegrityError: errors.append(f"Zeile {excel_row_number}: Inventarnummer, Seriennummer oder MeshCentral Node-ID ist bereits vergeben") continue try: db.commit() except IntegrityError as exc: db.rollback() raise HTTPException(400, "Der Import konnte wegen doppelter eindeutiger Werte nicht gespeichert werden.") from exc result = {"created": created, "updated": updated, "skipped": skipped, "errors": errors, "unknown_headers": unknown_headers} return templates.TemplateResponse( "assets_import.html", {"request": request, "categories": categories, "selected": category, "result": result}, ) @app.get("/assets/{asset_id}") def asset_detail(asset_id: int, request: Request, db: Session = Depends(get_db)): asset = _apply_asset_access(db.query(Asset), request).options(joinedload(Asset.category).joinedload(Category.visible_fields)).filter(Asset.id == asset_id).first() if not asset: raise HTTPException(404, "Asset nicht gefunden") active_fields = [f for f in asset.category.visible_fields if f.active] history_count = db.query(AssetHistory).filter(AssetHistory.asset_id == asset.id).count() return templates.TemplateResponse("asset_detail.html", {"request": request, "asset": asset, "active_fields": active_fields, "history_count": history_count, "is_admin": _is_admin(request), "issue_status_name": (_action_status(db, "issue").name if _action_status(db, "issue") else None), "today": date.today().isoformat()}) @app.get("/assets/{asset_id}/edit") def asset_edit(asset_id: int, request: Request, category_id: int | None = None, db: Session = Depends(get_db)): _require_admin(request) asset = db.query(Asset).options(joinedload(Asset.category).joinedload(Category.visible_fields)).filter(Asset.id == asset_id).first() categories = db.query(Category).options(joinedload(Category.visible_fields)).all() if not asset: raise HTTPException(404, "Asset nicht gefunden") selected = next((category for category in categories if category.id == category_id), None) if category_id else asset.category if selected is None: selected = asset.category return templates.TemplateResponse("asset_form.html", {"request": request, "asset": asset, "categories": categories, "selected": selected, "fields": ASSET_FIELDS, "status_options": _active_status_options(db), "duplicate_mode": False}) @app.post("/assets/{asset_id}/edit") async def asset_update(asset_id: int, request: Request, category_id: int = Form(...), image: UploadFile | None = File(None), db: Session = Depends(get_db)): asset = db.query(Asset).filter(Asset.id == asset_id).first() category = db.query(Category).options(joinedload(Category.visible_fields)).filter(Category.id == category_id).first() if not asset or not category: raise HTTPException(404, "Asset oder Kategorie nicht gefunden") form = await request.form() changes: dict[str, dict[str, str]] = {} old_category_id = asset.category_id if old_category_id != category_id: old_category = db.query(Category).filter(Category.id == old_category_id).first() changes["Kategorie"] = {"old": old_category.name if old_category else str(old_category_id), "new": category.name} asset.category_id = category_id for field in category.visible_fields: if field.active and hasattr(asset, field.field_name): old_value = getattr(asset, field.field_name) new_value = form.get(field.field_name) or None if _display_value(old_value) != _display_value(new_value): changes[field.label] = {"old": _display_value(old_value), "new": _display_value(new_value)} setattr(asset, field.field_name, new_value) new_image = save_upload(image) if new_image: if asset.image_path != new_image: changes["Bild"] = {"old": asset.image_path or "", "new": new_image} asset.image_path = new_image try: _record_asset_history(db, asset, changes, source="manual-update", changed_by=_changed_by(request)) db.commit() except IntegrityError as exc: db.rollback() raise HTTPException(400, "Asset konnte nicht gespeichert werden. Inventarnummer, Seriennummer oder MeshCentral Node-ID ist möglicherweise bereits vergeben.") from exc return RedirectResponse(f"/assets/{asset.id}", status_code=303) @app.get("/assets/{asset_id}/history") def asset_history(asset_id: int, request: Request, db: Session = Depends(get_db)): asset = _get_visible_asset(db, request, asset_id) entries = db.query(AssetHistory).filter(AssetHistory.asset_id == asset_id).order_by(AssetHistory.changed_at.desc()).all() return templates.TemplateResponse("asset_history.html", {"request": request, "asset": asset, "entries": entries}) @app.get("/assets/{asset_id}/history/export.xlsx") def asset_history_export(asset_id: int, request: Request, db: Session = Depends(get_db)): asset = _get_visible_asset(db, request, asset_id) if not asset: raise HTTPException(404, "Asset nicht gefunden") entries = db.query(AssetHistory).filter(AssetHistory.asset_id == asset_id).order_by(AssetHistory.changed_at.desc()).all() rows = [] for entry in entries: for label, change in (entry.changes or {}).items(): rows.append([entry.changed_at, entry.source, entry.changed_by or "", label, change.get("old", ""), change.get("new", "")]) return _xlsx_response(f"asset-{asset.id}-historie.xlsx", ["Zeitpunkt", "Quelle", "Benutzer", "Feld", "Alter Wert", "Neuer Wert"], rows) @app.get("/assets/export.xlsx") def assets_export(request: Request, category_id: int | None = None, db: Session = Depends(get_db)): categories = db.query(Category).options(joinedload(Category.visible_fields)).all() selected = next((category for category in categories if category.id == category_id), None) if category_id else None query = _apply_asset_access(db.query(Asset), request) if selected: query = query.filter(Asset.category_id == selected.id) assets = query.order_by(Asset.name).all() fields = ([field for field in sorted(selected.visible_fields, key=lambda item: item.sort_order) if field.active and field.show_in_list] if selected else _list_fields_for_categories(categories)) headers = (["Kategorie"] if not selected else []) + [field.label for field in fields] if not any(field.field_name == "name" for field in fields): headers.insert(0, "Bezeichnung") rows = [] for asset in assets: row = ([asset.category.name] if not selected else []) + [getattr(asset, field.field_name, "") for field in fields] if not any(field.field_name == "name" for field in fields): row.insert(0, asset.name) rows.append(row) suffix = selected.name if selected else "alle" return _xlsx_response(f"assets-{suffix}.xlsx", headers, rows) @app.get("/export/assets.xlsx") def assets_export_static(request: Request, category_id: int | None = None, db: Session = Depends(get_db)): return assets_export(request=request, category_id=category_id, db=db) @app.post("/assets/{asset_id}/issue") def asset_issue(asset_id: int, request: Request, return_to: str = Form("detail"), assigned_to: str = Form(""), assigned_on: str = Form(""), db: Session = Depends(get_db)): _require_admin(request) asset = _get_visible_asset(db, request, asset_id) status = _action_status(db, "issue") if not status: raise HTTPException(400, "Kein Status für die Aktion Ausgeben konfiguriert") changes = _asset_action_changes(asset, "issue", status.name, assigned_to, assigned_on) _record_asset_history(db, asset, changes, source="manual-issue", changed_by=_changed_by(request)) db.commit() return RedirectResponse("/" if return_to == "dashboard" else ("/assets" if return_to == "list" else f"/assets/{asset.id}"), status_code=303) @app.post("/assets/{asset_id}/return") def asset_return(asset_id: int, request: Request, return_to: str = Form("detail"), db: Session = Depends(get_db)): _require_admin(request) asset = _get_visible_asset(db, request, asset_id) status = _action_status(db, "return") if not status: raise HTTPException(400, "Kein Status für die Aktion Zurücknehmen konfiguriert") changes = _asset_action_changes(asset, "return", status.name) _record_asset_history(db, asset, changes, source="manual-return", changed_by=_changed_by(request)) db.commit() return RedirectResponse("/" if return_to == "dashboard" else ("/assets" if return_to == "list" else f"/assets/{asset.id}"), status_code=303) @app.post("/assets/{asset_id}/delete") def asset_delete(asset_id: int, request: Request, db: Session = Depends(get_db)): _require_admin(request) asset = db.query(Asset).filter(Asset.id == asset_id).first() if not asset: raise HTTPException(404, "Asset nicht gefunden") image_path = asset.image_path db.delete(asset) db.commit() delete_asset_image(image_path) return RedirectResponse("/assets", status_code=303) @app.post("/assets/delete-all") def assets_delete_all(request: Request, category_id: int | None = Form(None), db: Session = Depends(get_db)): _require_admin(request) query = db.query(Asset) redirect_url = "/assets" if category_id is not None: query = query.filter(Asset.category_id == category_id) redirect_url = f"/assets?category_id={category_id}" assets = query.all() image_paths = [asset.image_path for asset in assets if asset.image_path] for asset in assets: db.delete(asset) db.commit() for image_path in image_paths: delete_asset_image(image_path) return RedirectResponse(redirect_url, status_code=303) @app.get("/status-options") def status_options_page(request: Request, db: Session = Depends(get_db)): _require_admin(request) options = db.query(StatusOption).order_by(StatusOption.sort_order, StatusOption.name).all() return templates.TemplateResponse("status_options.html", {"request": request, "options": options}) @app.post("/status-options") async def status_options_save(request: Request, db: Session = Depends(get_db)): _require_admin(request) form = await request.form() existing = db.query(StatusOption).all() for option in existing: name = (form.get(f"name_{option.id}") or "").strip() if not name: continue option.name = name option.active = form.get(f"active_{option.id}") == "on" try: option.sort_order = int(form.get(f"sort_{option.id}") or option.sort_order) except ValueError: pass issue_id = int(form.get("issue_status_id") or 0) return_id = int(form.get("return_status_id") or 0) for option in existing: option.use_for_issue = option.id == issue_id option.use_for_return = option.id == return_id new_name = (form.get("new_name") or "").strip() if new_name: max_order = max((item.sort_order for item in existing), default=-1) db.add(StatusOption(name=new_name, sort_order=max_order + 1, active=True)) try: db.commit() except IntegrityError as exc: db.rollback() raise HTTPException(400, "Statuswerte müssen eindeutig sein.") from exc return RedirectResponse("/status-options", status_code=303) @app.post("/status-options/{option_id}/delete") def status_option_delete(option_id: int, request: Request, db: Session = Depends(get_db)): _require_admin(request) option = db.query(StatusOption).filter(StatusOption.id == option_id).first() if not option: raise HTTPException(404, "Statuswert nicht gefunden") db.delete(option) db.commit() return RedirectResponse("/status-options", status_code=303) @app.get("/categories") def categories_list(request: Request, db: Session = Depends(get_db)): _require_admin(request) categories = db.query(Category).options(joinedload(Category.visible_fields)).order_by(Category.name).all() return templates.TemplateResponse("categories.html", {"request": request, "categories": categories}) @app.get("/categories/new") def category_new(request: Request): _require_admin(request) return templates.TemplateResponse("category_form.html", {"request": request, "category": None, "field_defs": ASSET_FIELDS, "selected_fields": DEFAULT_FIELDS}) @app.post("/categories/new") async def category_create(request: Request, name: str = Form(...), description: str = Form(""), image: UploadFile | None = File(None), db: Session = Depends(get_db)): _require_admin(request) form = await request.form() category = Category(name=name.strip(), description=description or None, image_path=save_upload(image)) db.add(category) db.flush() order = 0 for field_name, default_label in ASSET_FIELDS.items(): active = form.get(f"active_{field_name}") == "on" if active: db.add(CategoryField(category_id=category.id, field_name=field_name, label=form.get(f"label_{field_name}") or default_label, active=True, required=form.get(f"required_{field_name}") == "on", show_in_list=form.get(f"list_{field_name}") == "on", sort_order=order)) order += 1 if form.get("active_name") != "on": db.add(CategoryField(category_id=category.id, field_name="name", label="Bezeichnung", active=True, required=True, show_in_list=True, sort_order=0)) try: db.commit() except IntegrityError as exc: db.rollback() raise HTTPException(400, "Kategorie konnte nicht gespeichert werden. Der Kategoriename ist möglicherweise bereits vorhanden.") from exc return RedirectResponse("/categories", status_code=303) @app.get("/categories/{category_id}/edit") def category_edit(category_id: int, request: Request, db: Session = Depends(get_db)): _require_admin(request) category = db.query(Category).options(joinedload(Category.visible_fields)).filter(Category.id == category_id).first() if not category: raise HTTPException(404, "Kategorie nicht gefunden") selected_fields = {f.field_name: f for f in category.visible_fields} return templates.TemplateResponse("category_form.html", {"request": request, "category": category, "field_defs": ASSET_FIELDS, "selected_fields": selected_fields}) @app.post("/categories/{category_id}/edit") async def category_update(category_id: int, request: Request, name: str = Form(...), description: str = Form(""), image: UploadFile | None = File(None), db: Session = Depends(get_db)): _require_admin(request) category = db.query(Category).options(joinedload(Category.visible_fields)).filter(Category.id == category_id).first() if not category: raise HTTPException(404, "Kategorie nicht gefunden") form = await request.form() category.name = name.strip() category.description = description or None new_image = save_upload(image) if new_image: category.image_path = new_image # Vorhandene Definitionen explizit löschen und flushen. Das verhindert # UniqueConstraint-Verletzungen bei anschließend identischen Feldnamen. db.query(CategoryField).filter(CategoryField.category_id == category.id).delete(synchronize_session=False) db.flush() selected_names = [field_name for field_name in ASSET_FIELDS if form.get(f"active_{field_name}") == "on"] if "name" not in selected_names: selected_names.insert(0, "name") for order, field_name in enumerate(selected_names): default_label = ASSET_FIELDS[field_name] db.add(CategoryField( category_id=category.id, field_name=field_name, label=form.get(f"label_{field_name}") or default_label, active=True, required=True if field_name == "name" else form.get(f"required_{field_name}") == "on", show_in_list=True if field_name == "name" else form.get(f"list_{field_name}") == "on", sort_order=order, )) try: db.commit() except IntegrityError as exc: db.rollback() raise HTTPException(400, "Kategorie konnte nicht gespeichert werden. Der Kategoriename ist möglicherweise bereits vorhanden.") from exc return RedirectResponse("/categories", status_code=303) @app.get("/login") def login_page(request: Request): if _session_user(request): return RedirectResponse("/profile", status_code=303) mode = load_config().get("authentication", {}).get("mode", "none") return templates.TemplateResponse("login.html", {"request": request, "mode": mode}) @app.post("/login") def login_submit(request: Request, username: str = Form(...), password: str = Form(...), db: Session = Depends(get_db)): username = username.strip() auth_config = load_config().get("authentication", {}) mode = str(auth_config.get("mode", "none")).lower() user = db.query(User).filter(User.username == username).first() valid = False if mode == "local": valid = bool(user and user.is_active and user.auth_source == "local" and _verify_password(password, user.password_hash)) elif mode == "ldap": ldap_user = _ldap_authenticate(username, password, auth_config) if ldap_user: if not user: user = User(username=username, auth_source="ldap", is_active=True) db.add(user) if user.is_active: user.auth_source = "ldap" user.display_name = ldap_user.get("display_name") or username user.email = ldap_user.get("email") or None valid = True elif mode == "none": return RedirectResponse("/?toast_warning=Die Anmeldung ist derzeit deaktiviert", status_code=303) if not valid or not user: logger.warning("Fehlgeschlagene Anmeldung für Benutzer %s im Modus %s", username, mode) return RedirectResponse("/login?toast_error=Benutzername oder Passwort ist ungültig", status_code=303) user.last_login = datetime.utcnow() db.commit() db.refresh(user) _login_user(request, user) return RedirectResponse("/?toast_success=Anmeldung erfolgreich", status_code=303) @app.get("/logout") def logout(request: Request): request.session.clear() return RedirectResponse("/?toast_success=Sie wurden abgemeldet", status_code=303) @app.get("/profile") def profile(request: Request, db: Session = Depends(get_db)): session_user = _session_user(request) if not session_user: return RedirectResponse("/login?toast_warning=Bitte zuerst anmelden", status_code=303) user = db.query(User).filter(User.id == session_user.get("id")).first() if not user: request.session.clear() return RedirectResponse("/login?toast_error=Benutzerkonto wurde nicht gefunden", status_code=303) return templates.TemplateResponse("profile.html", {"request": request, "user": user}) @app.post("/profile/password") def profile_password(request: Request, current_password: str = Form(...), new_password: str = Form(...), db: Session = Depends(get_db)): session_user = _session_user(request) if not session_user: return RedirectResponse("/login", status_code=303) user = db.query(User).filter(User.id == session_user.get("id")).first() if not user or user.auth_source != "local": return RedirectResponse("/profile?toast_warning=Passwortänderungen sind nur für lokale Konten möglich", status_code=303) if not _verify_password(current_password, user.password_hash): return RedirectResponse("/profile?toast_error=Das aktuelle Passwort ist falsch", status_code=303) if len(new_password) < 8: return RedirectResponse("/profile?toast_error=Das neue Passwort muss mindestens 8 Zeichen lang sein", status_code=303) user.password_hash = _hash_password(new_password) db.commit() return RedirectResponse("/profile?toast_success=Passwort geändert", status_code=303) @app.get("/users") def users_page(request: Request, db: Session = Depends(get_db)): _require_admin(request) users = db.query(User).order_by(User.username).all() mesh_groups = [row[0] for row in db.query(Asset.mesh_group).filter(Asset.mesh_group.isnot(None), Asset.mesh_group != "").distinct().order_by(Asset.mesh_group).all()] return templates.TemplateResponse("users.html", {"request": request, "users": users, "auth": public_config().get("authentication", {}), "mesh_groups": mesh_groups}) @app.post("/users/new") def user_create(request: Request, username: str = Form(...), display_name: str = Form(""), email: str = Form(""), password: str = Form(...), is_admin: str | None = Form(None), db: Session = Depends(get_db)): _require_admin(request) username = username.strip() if not username: raise HTTPException(400, "Benutzername fehlt") if len(password) < 8: raise HTTPException(400, "Das Passwort muss mindestens 8 Zeichen lang sein") user = User(username=username, display_name=display_name.strip() or username, email=email.strip() or None, password_hash=_hash_password(password), auth_source="local", is_active=True, is_admin=is_admin == "on") db.add(user) try: db.commit() except IntegrityError as exc: db.rollback() raise HTTPException(400, "Der Benutzername ist bereits vorhanden") from exc return RedirectResponse("/users?toast_success=Benutzer angelegt", status_code=303) @app.post("/users/{user_id}/update") def user_update(user_id: int, request: Request, display_name: str = Form(""), email: str = Form(""), is_active: str | None = Form(None), is_admin: str | None = Form(None), new_password: str = Form(""), mesh_groups: list[str] = Form([]), db: Session = Depends(get_db)): _require_admin(request) user = db.query(User).filter(User.id == user_id).first() if not user: raise HTTPException(404, "Benutzer nicht gefunden") user.display_name = display_name.strip() or user.username user.email = email.strip() or None user.is_active = is_active == "on" user.is_admin = is_admin == "on" user.allowed_mesh_groups = sorted(set(mesh_groups)) if new_password: if user.auth_source != "local": raise HTTPException(400, "LDAP-Passwörter können hier nicht geändert werden") if len(new_password) < 8: raise HTTPException(400, "Das neue Passwort muss mindestens 8 Zeichen lang sein") user.password_hash = _hash_password(new_password) db.commit() return RedirectResponse("/users?toast_success=Benutzer gespeichert", status_code=303) def _ldap_search_users(query_text: str, auth_config: dict) -> list[dict]: from ldap3 import ALL, Connection, Server, SUBTREE from ldap3.utils.conv import escape_filter_chars ldap = auth_config.get("ldap", {}) server_name = str(ldap.get("server", "")).strip() use_ssl = bool(ldap.get("use_ssl", False)); start_tls = bool(ldap.get("start_tls", False)) port = int(ldap.get("port") or (636 if use_ssl else 389)) bind_dn = str(ldap.get("bind_dn", "")).strip() env_name = str(ldap.get("bind_password_env", "LDAP_BIND_PASSWORD")).strip() bind_password = os.getenv(env_name, "") or str(ldap.get("bind_password", "")) base_dn = str(ldap.get("user_base_dn", "")).strip() q = escape_filter_chars(query_text.strip()) search_filter = f"(&(objectCategory=person)(objectClass=user)(|(sAMAccountName=*{q}*)(displayName=*{q}*)(mail=*{q}*)))" server = Server(server_name, port=port, use_ssl=use_ssl, get_info=ALL, connect_timeout=10) conn = Connection(server, user=bind_dn or None, password=bind_password or None, auto_bind=False, receive_timeout=15) if not conn.open(): raise HTTPException(400, "LDAP-Verbindung fehlgeschlagen") if start_tls and not use_ssl and not conn.start_tls(): raise HTTPException(400, "LDAP StartTLS fehlgeschlagen") if not conn.bind(): raise HTTPException(400, f"LDAP Bind fehlgeschlagen: {conn.last_error}") conn.search(base_dn, search_filter, search_scope=SUBTREE, attributes=["sAMAccountName", "displayName", "mail", "distinguishedName"], size_limit=100) result=[] for e in conn.entries: result.append({"username": str(getattr(e,"sAMAccountName","") or ""), "display_name": str(getattr(e,"displayName","") or ""), "email": str(getattr(e,"mail","") or ""), "dn": e.entry_dn}) conn.unbind(); return result @app.get("/settings/ldap-search") def ldap_search_form(request: Request): _require_admin(request) return templates.TemplateResponse("ldap_search.html", {"request": request, "query": "", "results": []}) @app.post("/settings/ldap-search") def ldap_search_page(request: Request, ldap_query: str = Form("")): _require_admin(request) cfg = public_config() results = _ldap_search_users(ldap_query, cfg.get("authentication", {})) if ldap_query.strip() else [] return templates.TemplateResponse("ldap_search.html", {"request": request, "query": ldap_query, "results": results}) @app.get("/settings") def settings_page(request: Request): _require_admin(request) return templates.TemplateResponse("settings.html", {"request": request, "settings": public_config()}) @app.post("/settings") async def settings_save( request: Request, title: str = Form("AssetManager"), company_name: str = Form(""), auth_mode: str = Form("none"), ldap_server: str = Form(""), ldap_port: int = Form(389), ldap_use_ssl: str | None = Form(None), ldap_start_tls: str | None = Form(None), ldap_bind_dn: str = Form(""), ldap_bind_password_env: str = Form("LDAP_BIND_PASSWORD"), ldap_user_base_dn: str = Form(""), ldap_user_filter: str = Form("(sAMAccountName={username})"), ldap_display_name_attribute: str = Form("displayName"), ldap_email_attribute: str = Form("mail"), remove_logo: str | None = Form(None), logo: UploadFile | None = File(None), ): _require_admin(request) current = load_config() general = dict(current.get("general", {})) general["title"] = title.strip() or "AssetManager" general["company_name"] = company_name.strip() if remove_logo == "on": old_logo = general.get("logo") if old_logo: delete_asset_image(old_logo) general["logo"] = "" if logo and logo.filename: old_logo = general.get("logo") new_logo = save_upload(logo) if old_logo and old_logo != new_logo: delete_asset_image(old_logo) general["logo"] = new_logo or "" if auth_mode not in {"none", "local", "ldap"}: auth_mode = "none" authentication = dict(current.get("authentication", {})) authentication["mode"] = auth_mode ldap = dict(authentication.get("ldap", {})) ldap.update({ "server": ldap_server.strip(), "port": ldap_port, "use_ssl": ldap_use_ssl == "on", "start_tls": ldap_start_tls == "on", "bind_dn": ldap_bind_dn.strip(), "bind_password_env": ldap_bind_password_env.strip() or "LDAP_BIND_PASSWORD", "user_base_dn": ldap_user_base_dn.strip(), "user_filter": ldap_user_filter.strip() or "(sAMAccountName={username})", "display_name_attribute": ldap_display_name_attribute.strip() or "displayName", "email_attribute": ldap_email_attribute.strip() or "mail", }) authentication["ldap"] = ldap save_config({"general": general, "authentication": authentication}) return RedirectResponse("/settings?toast_success=Einstellungen gespeichert", status_code=303) @app.get("/sync/meshcentral") def meshcentral_sync_page(request: Request, db: Session = Depends(get_db)): _require_admin(request) runs = db.query(SyncRun).order_by(SyncRun.started_at.desc()).limit(20).all() linked = db.query(Asset).filter(Asset.mesh_node_id.isnot(None)).count() conflicts = db.query(Asset).filter(Asset.mesh_sync_status == "conflict").count() missing = db.query(Asset).filter(Asset.mesh_sync_status == "missing").count() return templates.TemplateResponse("meshcentral_sync.html", { "request": request, "runs": runs, "config": public_config(), "linked": linked, "conflicts": conflicts, "missing": missing }) @app.post("/sync/meshcentral/start") def meshcentral_sync_start(request: Request): _require_admin(request) with SYNC_JOBS_LOCK: if any(not job.get("finished") for job in SYNC_JOBS.values()): return JSONResponse({"error": "Es läuft bereits eine Synchronisierung."}, status_code=409) job_id = uuid.uuid4().hex SYNC_JOBS[job_id] = {"status": "running", "finished": False, "run_id": None, "logs": [], "created_at": datetime.now().isoformat()} _job_log(job_id, "info", "Synchronisierungsauftrag angelegt.") threading.Thread(target=_run_sync_job, args=(job_id,), daemon=True).start() return {"job_id": job_id} @app.get("/sync/meshcentral/jobs/{job_id}") def meshcentral_sync_job(job_id: str, offset: int = 0): with SYNC_JOBS_LOCK: job = SYNC_JOBS.get(job_id) if not job: raise HTTPException(404, "Synchronisierungsauftrag nicht gefunden") logs = list(job["logs"][max(0, offset):]) return { "status": job["status"], "finished": job["finished"], "run_id": job["run_id"], "logs": logs, "next_offset": max(0, offset) + len(logs), } @app.post("/sync/meshcentral") def meshcentral_sync_run(): return RedirectResponse("/sync/meshcentral", status_code=303)