Django scaffold aligned with the business delivery lane: tenant-keyed Member model without passwords, identity app as sole OIDC/session boundary, dev-claims login, authenticated /app/ shell, ADR-0001, and tests. T01/T02/T05/T06 done; OIDC registration, real user-engine HTTP, flex-auth, and packaging remain open.
118 lines
4.3 KiB
Python
118 lines
4.3 KiB
Python
"""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 = IdentityClaims(
|
|
issuer=str(issuer),
|
|
subject=str(sub),
|
|
email=str(raw.get("email") or ""),
|
|
name=str(raw.get("name") or ""),
|
|
preferred_username=str(raw.get("preferred_username") or ""),
|
|
)
|
|
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(),
|
|
)
|
|
establish_session(request, claims)
|
|
return redirect(settings.LOGIN_REDIRECT_URL)
|
|
|
|
return render(request, "identity/dev_login.html")
|
|
|
|
|
|
@require_http_methods(["GET", "POST"])
|
|
def logout_view(request: HttpRequest) -> HttpResponse:
|
|
logout(request)
|
|
return redirect(settings.LOGOUT_REDIRECT_URL)
|