"""OIDC login/callback/logout + DEBUG dev-login. Sole auth entrypoints.""" from __future__ import annotations import logging from django.conf import settings from django.contrib import messages from django.contrib.auth import logout from django.http import HttpRequest, HttpResponse, HttpResponseBadRequest from django.shortcuts import redirect, render from django.urls import reverse from django.views.decorators.http import require_GET, require_http_methods from . import oidc from .services import establish_session from .user_engine import IdentityClaims logger = logging.getLogger(__name__) SESSION_OIDC_STATE = "oidc_state" SESSION_OIDC_VERIFIER = "oidc_code_verifier" @require_GET def login_start(request: HttpRequest) -> HttpResponse: if request.user.is_authenticated: return redirect(settings.LOGIN_REDIRECT_URL) if oidc.oidc_configured(): state, verifier = oidc.new_pkce_pair() request.session[SESSION_OIDC_STATE] = state request.session[SESSION_OIDC_VERIFIER] = verifier try: url = oidc.build_authorization_url(state=state, code_verifier=verifier) except Exception: logger.exception("OIDC authorization URL build failed") messages.error(request, "Identity provider is unavailable. Try again later.") return redirect("core:landing") return redirect(url) if settings.DEBUG: return redirect("identity:dev_login") messages.error(request, "Sign-in is not configured (OIDC_ENABLED=false).") return redirect("core:landing") @require_GET def oidc_callback(request: HttpRequest) -> HttpResponse: if not oidc.oidc_configured(): return HttpResponseBadRequest("OIDC is not enabled") error = request.GET.get("error") if error: messages.error(request, f"Sign-in failed: {error}") return redirect("core:landing") code = request.GET.get("code") state = request.GET.get("state") expected_state = request.session.pop(SESSION_OIDC_STATE, None) verifier = request.session.pop(SESSION_OIDC_VERIFIER, None) if not code or not state or state != expected_state or not verifier: return HttpResponseBadRequest("Invalid OIDC callback state") try: token = oidc.exchange_code(code, code_verifier=verifier) userinfo = oidc.fetch_userinfo(token.get("access_token", "")) raw = oidc.claims_from_token_response(token, userinfo) except Exception: logger.exception("OIDC token exchange failed") messages.error(request, "Could not complete sign-in with the identity provider.") return redirect("core:landing") sub = raw.get("sub") if not sub: return HttpResponseBadRequest("Token missing subject") issuer = raw.get("iss") or settings.OIDC_ISSUER claims = _claims_from_oidc_payload(raw, issuer=str(issuer), subject=str(sub)) establish_session(request, claims) return redirect(settings.LOGIN_REDIRECT_URL) @require_http_methods(["GET", "POST"]) def dev_login(request: HttpRequest) -> HttpResponse: """Local-only claims form when OIDC is off. Never enable outside DEBUG.""" if not settings.DEBUG or settings.OIDC_ENABLED: return HttpResponseBadRequest("Dev login only when DEBUG and OIDC disabled") if request.method == "POST": subject = (request.POST.get("subject") or "").strip() if not subject: messages.error(request, "Subject is required") return render(request, "identity/dev_login.html") claims = IdentityClaims( issuer=request.POST.get("issuer") or "https://local.dev/issuer", subject=subject, email=(request.POST.get("email") or "").strip(), name=(request.POST.get("name") or "").strip(), preferred_username=(request.POST.get("preferred_username") or "").strip(), tenant=(request.POST.get("tenant") or settings.DEFAULT_TENANT_ID).strip(), principal_type="human", roles=("user",), authorized_party=settings.USER_ENGINE_APPLICATION_ID, ) establish_session(request, claims) return redirect(settings.LOGIN_REDIRECT_URL) return render( request, "identity/dev_login.html", {"default_tenant": settings.DEFAULT_TENANT_ID}, ) def _claims_from_oidc_payload( raw: dict, *, issuer: str, subject: str ) -> IdentityClaims: aud = raw.get("aud", ()) if isinstance(aud, str): audience = (aud,) else: audience = tuple(str(a) for a in (aud or ())) groups = raw.get("groups") or () if isinstance(groups, str): groups = (groups,) roles = raw.get("roles") or raw.get("tenant_roles") or () if isinstance(roles, str): roles = (roles,) assurance = raw.get("assurance") if isinstance(raw.get("assurance"), dict) else {} return IdentityClaims( issuer=issuer, subject=subject, email=str(raw.get("email") or ""), name=str(raw.get("name") or ""), preferred_username=str(raw.get("preferred_username") or ""), tenant=str(raw.get("tenant") or settings.DEFAULT_TENANT_ID), principal_type=str(raw.get("principal_type") or "human"), groups=tuple(str(g) for g in groups), roles=tuple(str(r) for r in roles) or ("user",), assurance=dict(assurance), audience=audience, authorized_party=str( raw.get("azp") or raw.get("client_id") or settings.OIDC_CLIENT_ID or "" ), ) @require_http_methods(["GET", "POST"]) def logout_view(request: HttpRequest) -> HttpResponse: logout(request) return redirect(settings.LOGOUT_REDIRECT_URL)