ebbe.bassandClaude Sonnet 5.5
3c69c2a156
Authorization-code flow with PKCE through an idp adapter (with a fake), ID token validation, identity and roles resolved from the directory with a short cache, refusal of unknown, disabled and locked accounts with one generic page, sessions that end when the account becomes unavailable, sign-out that ends the server-side session, role decorators and the trusted-proxy client address rule. Adds the OIDC_* and PUBLIC_BASE_URL configuration. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01N3t4WL7bd9bdHHNjEsxzqC
106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
"""Sign-in with OIDC, sign-out and the signed-in landing page (FR-1, FR-3)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from django.db import transaction
|
|
from django.shortcuts import redirect, render
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
from django.views.decorators.http import require_GET, require_POST
|
|
|
|
from portal.adapters.directory import DirectoryUnavailable
|
|
from portal.adapters.idp import Flow, IdpUnavailable, LoginRejected
|
|
from portal.adapters.registry import get_idp
|
|
from portal.audit import services as audit
|
|
|
|
from . import identity
|
|
from .decorators import login_required
|
|
from .session_store import LOGIN_FLOW
|
|
from .sessions import start_session
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _unavailable(request):
|
|
return render(request, "accounts/unavailable.html", status=503)
|
|
|
|
|
|
def _refused(request, reason: str, username: str = ""):
|
|
"""One generic page for every refusal, so it never says whether the name exists."""
|
|
audit.record_failure(
|
|
username,
|
|
"signin.refused",
|
|
ip=request.client_ip,
|
|
detail={"reason": reason},
|
|
)
|
|
return render(request, "accounts/refused.html", status=403)
|
|
|
|
|
|
def _safe_next(value: str) -> str:
|
|
if value.startswith("/") and url_has_allowed_host_and_scheme(value, allowed_hosts=None):
|
|
return value
|
|
return "/"
|
|
|
|
|
|
@require_GET
|
|
def login(request):
|
|
if request.portal_user is not None:
|
|
return redirect("home")
|
|
try:
|
|
url, flow = get_idp().begin()
|
|
except IdpUnavailable:
|
|
return _unavailable(request)
|
|
target = _safe_next(request.GET.get("next", ""))
|
|
request.session[LOGIN_FLOW] = {**flow.as_dict(), "next": target}
|
|
return redirect(url)
|
|
|
|
|
|
@require_GET
|
|
def callback(request):
|
|
data = request.session.pop(LOGIN_FLOW, None) # a flow works once
|
|
try:
|
|
flow = Flow.from_dict(data) if data else None
|
|
except KeyError:
|
|
flow = None
|
|
if flow is None:
|
|
return _refused(request, "no_flow")
|
|
try:
|
|
found = get_idp().complete(request.GET, flow)
|
|
except IdpUnavailable:
|
|
return _unavailable(request)
|
|
except LoginRejected:
|
|
return _refused(request, "provider_response")
|
|
try:
|
|
user = identity.admit(found.username)
|
|
except identity.SignInRefused as refusal:
|
|
return _refused(request, refusal.reason, found.username)
|
|
except DirectoryUnavailable:
|
|
return _unavailable(request)
|
|
|
|
start_session(request, user.sam)
|
|
identity.remember(request, user)
|
|
with transaction.atomic():
|
|
audit.record(user.sam, "signin.succeeded", ip=request.client_ip)
|
|
return redirect(_safe_next(data.get("next", "")))
|
|
|
|
|
|
@login_required
|
|
def home(request):
|
|
return render(request, "accounts/home.html", {"portal_user": request.portal_user})
|
|
|
|
|
|
@require_POST
|
|
def logout(request):
|
|
portal_user = request.portal_user
|
|
if portal_user is not None:
|
|
with transaction.atomic():
|
|
audit.record(portal_user.user.key, "signout", ip=request.client_ip)
|
|
request.session.flush() # deletes the session row on the server
|
|
return redirect("signed-out")
|
|
|
|
|
|
@require_GET
|
|
def signed_out(request):
|
|
return render(request, "accounts/signed_out.html")
|