py app init + zitadel doku

app/routes.py

@@ -0,0 +1,184 @@
+"""
+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:  <Vorname Nachname>
+: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": {"<org-id>": "<org-name>"}, ...}
+    roles_claim = userinfo.get("urn:zitadel:iam:org:project:roles", {})
+
+    logger.info(
+        "OIDC Login – sub: %s | email: %s | roles: %s | full_userinfo: %s",
+        sub, email, list(roles_claim.keys()), dict(userinfo),
+    )
+
+    if not roles_claim:
+        logger.warning("Kein Roles-Claim im Token für sub=%s – Zugriff verweigert.", sub)
+        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()
+
+    session["user"] = {"sub": sub, "email": email, "role": role}
+    session["id_token"] = token.get("id_token")
+
+    return redirect(url_for("main.index"))
+
+
+@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/<int:project_id>")
+@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)