215 lines
7.0 KiB
Python
215 lines
7.0 KiB
Python
"""
|
||
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 <br@runden.de>
|
||
: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", {})
|
||
|
||
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/<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)
|