1352 lines
63 KiB
Python
1352 lines
63 KiB
Python
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)
|