Route rejected company sign-in to central account recovery
All checks were successful
Application acceptance / application-tests (push) Successful in 1m8s
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s
Build and Publish Container Image / build-and-push (push) Successful in 30s

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
This commit is contained in:
tegwick 2026-09-12 10:34:44 +02:00
parent c1727ed14d
commit c067993037
5 changed files with 54 additions and 13 deletions

View file

@ -148,7 +148,10 @@ def test_verified_login_needs_explicit_confirmation_and_stable_mapping(client, s
def test_invalid_tokens_never_create_users(client, signed_flow, changes):
client.post("/accounts/oidc/start/")
response = signed_flow(client, changes)
assert response.status_code == 403
assert response.status_code == 302
assert response.url == "https://users.coulomb.social/access-recovery"
assert response["Referrer-Policy"] == "no-referrer"
assert "oidc_confirm" not in client.session
assert not Mitarbeiter.objects.exists()
assert "_auth_user_id" not in client.session
@ -159,7 +162,7 @@ def test_wrong_algorithm_denied(client, signed_flow):
signed_flow(
client, token_override=jwt.encode({"sub": "forged"}, "x" * 32, algorithm="HS256")
).status_code
== 403
== 302
)
@ -168,20 +171,20 @@ def test_state_is_browser_bound_expired_and_one_use(client, signed_flow):
original = client.session["oidc_pending"]
other = Client()
response = other.get("/accounts/oidc/callback/", {"state": original["state"], "code": "x"})
assert response.status_code == 403
assert response.status_code == 302
assert signed_flow(client).status_code == 302
assert (
client.get(
"/accounts/oidc/callback/", {"state": original["state"], "code": "x"}
).status_code
== 403
== 302
)
assert client.post("/accounts/oidc/confirm/").status_code == 403
assert client.post("/accounts/oidc/confirm/").status_code == 302
client.post("/accounts/oidc/start/")
session = client.session
session["oidc_pending"]["created"] -= 601
session.save()
assert signed_flow(client).status_code == 403
assert signed_flow(client).status_code == 302
def test_confirmation_requires_csrf(client, signed_flow):
@ -205,13 +208,13 @@ def test_no_email_merge_and_inactive_or_staff_accounts_stay_denied(client, signe
assert client.get("/ausschreibungen/").status_code == 302
client.post("/accounts/oidc/start/")
signed_flow(client)
assert client.post("/accounts/oidc/confirm/").status_code == 403
assert client.post("/accounts/oidc/confirm/").status_code == 302
linked.is_active = True
linked.is_staff = True
linked.save()
client.post("/accounts/oidc/start/")
signed_flow(client)
assert client.post("/accounts/oidc/confirm/").status_code == 403
assert client.post("/accounts/oidc/confirm/").status_code == 302
def test_absolute_session_expiry_and_wrong_identity_do_not_reuse_operator(client, signed_flow):
@ -219,7 +222,7 @@ def test_absolute_session_expiry_and_wrong_identity_do_not_reuse_operator(client
client.force_login(operator)
client.post("/accounts/oidc/start/")
assert "_auth_user_id" not in client.session
assert signed_flow(client, {"tenant": "tenant:platform"}).status_code == 403
assert signed_flow(client, {"tenant": "tenant:platform"}).status_code == 302
client.post("/accounts/oidc/start/")
signed_flow(client)
client.post("/accounts/oidc/confirm/")
@ -261,5 +264,14 @@ def test_wrong_signature_denied(client, signed_flow):
client.post("/accounts/oidc/start/")
another = rsa.generate_private_key(public_exponent=65537, key_size=2048)
token = jwt.encode({"sub": "forged"}, another, algorithm="RS256")
assert signed_flow(client, token_override=token).status_code == 403
assert signed_flow(client, token_override=token).status_code == 302
assert not Mitarbeiter.objects.exists()
def test_rejected_callback_uses_fixed_recovery_without_browser_credentials(client, configured):
response = client.get("/accounts/oidc/callback/", {"code":"private-code", "state":"private-state", "next":"https://evil.example"})
assert response.url == "https://users.coulomb.social/access-recovery"
assert "no-store" in response["Cache-Control"]
assert response["Referrer-Policy"] == "no-referrer"
assert not Mitarbeiter.objects.exists()
assert "_auth_user_id" not in client.session

View file

@ -19,6 +19,13 @@ from .forms import AuthenticationForm
from .models import Mitarbeiter, OIDCIdentity
def account_recovery():
response = redirect(settings.NETKINGDOM_ACCOUNT_PORTAL_URL.rstrip("/") + "/access-recovery")
response["Cache-Control"] = "no-store"
response["Referrer-Policy"] = "no-referrer"
return response
def context(**extra):
return {"company_name": settings.COMPANY_DISPLAY_NAME, **extra}
@ -71,7 +78,7 @@ def oidc_callback(request):
raise oidc.LoginRejectedError("Authorization was not completed")
verified = oidc.complete(pending, request.GET.get("state", ""), request.GET.get("code", ""))
except (ValueError, jwt.PyJWTError, URLError, TimeoutError, OSError, ImproperlyConfigured):
return render(request, "accounts/welcome.html", context(login_error=True), status=403)
return account_recovery()
request.session["oidc_confirm"] = verified
# Remove the authorization code from the address bar before displaying identity.
return redirect("accounts:oidc_confirm")
@ -91,7 +98,7 @@ def oidc_confirm(request):
or verified.get("client") != settings.NETKINGDOM_CLIENT_ID
):
request.session.pop("oidc_confirm", None)
return render(request, "accounts/welcome.html", context(login_error=True), status=403)
return account_recovery()
if request.method == "GET":
return render(request, "accounts/confirm.html", context(identity_label=verified["label"]))
if request.method != "POST":
@ -129,7 +136,7 @@ def oidc_confirm(request):
if not user.is_active or user.is_staff or user.is_superuser:
raise oidc.LoginRejectedError("Product account is not admitted")
except (IntegrityError, oidc.LoginRejectedError):
return render(request, "accounts/welcome.html", context(login_error=True), status=403)
return account_recovery()
login(request, user, backend="django.contrib.auth.backends.ModelBackend")
request.session["oidc_expires"] = verified["expires"]
request.session["oidc_binding"] = [verified["issuer"], verified["tenant"], verified["client"]]