""" Flask-Routen und Routing-Guards der OIDC-Showcase-App. Alle geschützten Routen prüfen (1) den Zitadel-Claim aus der Session und (2) die lokale DB-Zuordnung. Jede Verletzung leitet auf /error um. :author: Björn Runden :date: 2026-06-17 :version: 0.1 """ import logging from urllib.parse import urlencode from flask import Blueprint, render_template, session, redirect, url_for from .auth import oauth, login_required, role_required, get_current_user from .database import db from .models import ( User, Client, Project, ProjectMitarbeiterMapping, ProjectKundenMapping, Note, ) logger = logging.getLogger(__name__) main = Blueprint("main", __name__) # --------------------------------------------------------------------------- # Öffentliche Routen # --------------------------------------------------------------------------- @main.route("/login") def login(): redirect_uri = url_for("main.auth_callback", _external=True) return oauth.zitadel.authorize_redirect(redirect_uri) @main.route("/auth/callback") def auth_callback(): token = oauth.zitadel.authorize_access_token() # Authlib parst den ID-Token und legt die Claims unter token["userinfo"] ab. # Zitadel muss "Assert Roles on Authentication" aktiviert haben (ZITADEL-SETUP.md). userinfo = token.get("userinfo") or {} sub = userinfo.get("sub") email = userinfo.get("email", "") # Zitadel-Roles-Claim: {"Admin": {"": ""}, ...} roles_claim = userinfo.get("urn:zitadel:iam:org:project:roles", {}) if not roles_claim: logger.warning("Kein Roles-Claim im Token für sub=%s – Zugriff verweigert.", sub) _log_login(sub, email, list(roles_claim.keys()), app_role=None, projects=[]) return redirect(url_for("main.error")) role = next(iter(roles_claim.keys())) user = db.session.get(User, sub) if user: user.email = email user.global_role = role else: user = User(zitadel_user_id=sub, email=email, global_role=role) db.session.add(user) db.session.commit() # Projekt-Zuordnungen für den Log ermitteln if role == "Mitarbeiter": mappings = ProjectMitarbeiterMapping.query.filter_by(user_id=sub).all() projects = Project.query.filter( Project.id.in_({m.project_id for m in mappings}) ).all() elif role == "Kunde": mappings = ProjectKundenMapping.query.filter_by(user_id=sub).all() projects = Project.query.filter( Project.id.in_({m.project_id for m in mappings}) ).all() else: projects = Project.query.all() # Admin sieht alles _log_login(sub, email, list(roles_claim.keys()), app_role=role, projects=projects) session["user"] = {"sub": sub, "email": email, "role": role} session["id_token"] = token.get("id_token") return redirect(url_for("main.index")) def _log_login(sub, email, zitadel_roles, app_role, projects): project_list = ", ".join(f"[{p.id}] {p.name}" for p in projects) or "–" logger.info( "\n" "╔══ LOGIN ══════════════════════════════════\n" "║ sub : %s\n" "║ email : %s\n" "║ Zitadel-Rol. : %s\n" "║ App-Rolle : %s\n" "║ Projekte : %s\n" "╚═══════════════════════════════════════════", sub, email, ", ".join(zitadel_roles) or "–", app_role or "KEIN ZUGRIFF", project_list, ) @main.route("/logout") def logout(): id_token = session.pop("id_token", None) session.clear() try: metadata = oauth.zitadel.load_server_metadata() end_session_endpoint = metadata.get("end_session_endpoint") if end_session_endpoint: params = {"post_logout_redirect_uri": url_for("main.index", _external=True)} if id_token: params["id_token_hint"] = id_token return redirect(f"{end_session_endpoint}?{urlencode(params)}") except Exception as exc: logger.error("Logout: Fehler beim Laden der Server-Metadaten: %s", exc) return redirect(url_for("main.index")) @main.route("/error") def error(): return render_template("login_error.html"), 403 # --------------------------------------------------------------------------- # Geschützte Routen # --------------------------------------------------------------------------- @main.route("/") @login_required def index(): user = get_current_user() role = user.get("role") sub = user.get("sub") accessible_projects = [] forbidden_project = None if role == "Mitarbeiter": mappings = ProjectMitarbeiterMapping.query.filter_by(user_id=sub).all() accessible_ids = {m.project_id for m in mappings} accessible_projects = Project.query.filter(Project.id.in_(accessible_ids)).all() forbidden_project = Project.query.filter(Project.id.notin_(accessible_ids)).first() elif role == "Kunde": mappings = ProjectKundenMapping.query.filter_by(user_id=sub).all() accessible_ids = {m.project_id for m in mappings} accessible_projects = Project.query.filter(Project.id.in_(accessible_ids)).all() forbidden_project = Project.query.filter(Project.id.notin_(accessible_ids)).first() return render_template( "index.html", user=user, projects=accessible_projects, forbidden_project=forbidden_project, ) @main.route("/admin") @role_required("Admin") def admin(): user = get_current_user() return render_template( "admin.html", user=user, clients=Client.query.all(), projects=Project.query.all(), users=User.query.all(), ) @main.route("/mitarbeiter") @role_required("Mitarbeiter") def mitarbeiter(): user = get_current_user() sub = user["sub"] mappings = ProjectMitarbeiterMapping.query.filter_by(user_id=sub).all() projects = Project.query.filter( Project.id.in_({m.project_id for m in mappings}) ).all() return render_template("mitarbeiter.html", user=user, projects=projects) @main.route("/projekt/") @login_required def projekt(project_id): user = get_current_user() role = user.get("role") sub = user.get("sub") project = db.session.get(Project, project_id) if not project: return redirect(url_for("main.error")) if role == "Admin": pass elif role == "Mitarbeiter": if not ProjectMitarbeiterMapping.query.filter_by(user_id=sub, project_id=project_id).first(): return redirect(url_for("main.error")) elif role == "Kunde": if not ProjectKundenMapping.query.filter_by(user_id=sub, project_id=project_id).first(): return redirect(url_for("main.error")) else: return redirect(url_for("main.error")) notes = Note.query.filter_by(project_id=project_id).order_by(Note.created_at.desc()).all() return render_template("projekt.html", user=user, project=project, notes=notes)