feat: add OIDC sign-in, roles, sessions and client address rule
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
This commit is contained in:
Claude Sonnet 5.5
@@ -0,0 +1,339 @@
|
||||
"""Sign-in, roles, sign-out and account-state handling through the real views (FR-1..FR-3)."""
|
||||
|
||||
import time
|
||||
from urllib.parse import parse_qs, urlsplit
|
||||
|
||||
import pytest
|
||||
from django.http import HttpResponse
|
||||
from django.test import Client, RequestFactory
|
||||
|
||||
from portal.accounts import identity
|
||||
from portal.accounts.decorators import admin_required, login_required, publisher_required
|
||||
from portal.accounts.models import UserSession
|
||||
from portal.accounts.session_store import LOGIN_FLOW, ROLES
|
||||
from portal.accounts.sessions import SessionUser
|
||||
from portal.adapters.idp import Flow
|
||||
from portal.audit.models import AuditEntry
|
||||
|
||||
pytestmark = pytest.mark.django_db(transaction=True)
|
||||
|
||||
|
||||
def sign_in(adapters, username="alice", next_path=None, client=None):
|
||||
"""Run the whole flow with the fake provider; returns (client, callback response)."""
|
||||
client = client or Client()
|
||||
adapters.idp.sign_in_as(username)
|
||||
query = f"?next={next_path}" if next_path else ""
|
||||
client.get("/auth/login" + query)
|
||||
flow = Flow.from_dict(client.session[LOGIN_FLOW])
|
||||
return client, client.get("/auth/callback", adapters.idp.callback_params(flow))
|
||||
|
||||
|
||||
def age_the_cache(client):
|
||||
session = client.session
|
||||
session[ROLES] = {**session[ROLES], "checked": time.time() - 3600}
|
||||
session.save()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def people(adapters):
|
||||
d = adapters.directory
|
||||
d.add_user("Alice", given_name="Alice", surname="Anders", display_name="Alice Anders")
|
||||
d.add_user("pub", is_publisher=True)
|
||||
d.add_user("boss", is_admin=True, is_publisher=True)
|
||||
d.add_user("gone", enabled=False)
|
||||
d.add_user("locked", locked=True)
|
||||
d.add_user("expired", password_last_set=None)
|
||||
return d
|
||||
|
||||
|
||||
# -- reaching the sign-in -----------------------------------------------------------------
|
||||
|
||||
|
||||
def test_an_unauthenticated_request_redirects_to_sign_in(people):
|
||||
response = Client().get("/")
|
||||
assert response.status_code == 302
|
||||
assert response["Location"] == "/auth/login?next=%2F"
|
||||
|
||||
|
||||
def test_sign_in_redirects_to_the_provider_and_keeps_the_flow_in_the_session(people, adapters):
|
||||
client = Client()
|
||||
response = client.get("/auth/login")
|
||||
assert response.status_code == 302
|
||||
state = parse_qs(urlsplit(response["Location"]).query)["state"][0]
|
||||
assert client.session[LOGIN_FLOW]["state"] == state
|
||||
|
||||
|
||||
def test_the_provider_being_down_shows_an_unavailable_page(people, adapters):
|
||||
adapters.idp.unavailable = True
|
||||
response = Client().get("/auth/login")
|
||||
assert response.status_code == 503
|
||||
|
||||
|
||||
# -- a successful sign-in -------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_a_user_signs_in_and_sees_their_own_name(people, adapters):
|
||||
client, response = sign_in(adapters, "alice")
|
||||
assert response.status_code == 302 and response["Location"] == "/"
|
||||
page = client.get("/")
|
||||
assert page.status_code == 200
|
||||
assert "Alice Anders" in page.content.decode()
|
||||
|
||||
|
||||
def test_the_session_holds_the_exact_directory_name_and_a_lowercase_key(people, adapters):
|
||||
client, _ = sign_in(adapters, "ALICE")
|
||||
row = UserSession.objects.get(session_key=client.session.session_key)
|
||||
assert row.username == "alice"
|
||||
assert client.session["session_user_sam"] == "Alice"
|
||||
|
||||
|
||||
def test_the_requested_page_is_remembered(people, adapters):
|
||||
_, response = sign_in(adapters, "alice", next_path="/auth/signed-out")
|
||||
assert response["Location"] == "/auth/signed-out"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("target", ["https://evil.example.test/", "//evil.example.test", "x"])
|
||||
def test_an_outside_address_cannot_be_used_as_the_destination(people, adapters, target):
|
||||
_, response = sign_in(adapters, "alice", next_path=target)
|
||||
assert response["Location"] == "/"
|
||||
|
||||
|
||||
def test_sign_in_is_audited_with_the_client_address(people, adapters):
|
||||
sign_in(adapters, "alice")
|
||||
entry = AuditEntry.objects.get(action="signin.succeeded")
|
||||
assert entry.actor == "alice" and entry.ip == "127.0.0.1"
|
||||
assert "token" not in str(entry.detail).lower()
|
||||
|
||||
|
||||
def test_signing_in_replaces_a_previous_session_key(people, adapters):
|
||||
client = Client()
|
||||
client.get("/auth/login")
|
||||
before = client.session.session_key
|
||||
sign_in(adapters, "alice", client=client)
|
||||
assert client.session.session_key != before
|
||||
assert not UserSession.objects.filter(session_key=before).exists()
|
||||
|
||||
|
||||
def test_an_already_signed_in_user_is_sent_home(people, adapters):
|
||||
client, _ = sign_in(adapters, "alice")
|
||||
assert client.get("/auth/login")["Location"] == "/"
|
||||
|
||||
|
||||
# -- refusals -----------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_unknown_disabled_and_locked_accounts_are_refused_with_the_same_page(people, adapters):
|
||||
bodies = set()
|
||||
for name in ("nobody", "gone", "locked"):
|
||||
client, response = sign_in(adapters, name)
|
||||
assert response.status_code == 403
|
||||
assert UserSession.objects.exclude(username="").count() == 0
|
||||
assert client.get("/").status_code == 302
|
||||
bodies.add(response.content)
|
||||
assert len(bodies) == 1
|
||||
|
||||
|
||||
def test_refusals_are_audited_with_the_reason(people, adapters):
|
||||
for name in ("nobody", "gone", "locked"):
|
||||
sign_in(adapters, name)
|
||||
refusals = AuditEntry.objects.filter(action="signin.refused")
|
||||
reasons = {e.actor: e.detail["reason"] for e in refusals}
|
||||
assert reasons == {"nobody": "unknown", "gone": "disabled", "locked": "locked"}
|
||||
|
||||
|
||||
def test_the_refusal_page_does_not_name_the_account(people, adapters):
|
||||
_, response = sign_in(adapters, "gone")
|
||||
assert "gone" not in response.content.decode().lower()
|
||||
|
||||
|
||||
def test_a_user_with_an_expired_password_can_sign_in(people, adapters):
|
||||
client, response = sign_in(adapters, "expired")
|
||||
assert response.status_code == 302
|
||||
assert client.get("/").status_code == 200
|
||||
|
||||
|
||||
def test_a_callback_without_a_started_flow_is_refused(people, adapters):
|
||||
response = Client().get("/auth/callback", {"state": "x", "code": "y"})
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
def test_a_callback_works_once(people, adapters):
|
||||
adapters.idp.sign_in_as("alice")
|
||||
client = Client()
|
||||
client.get("/auth/login")
|
||||
params = adapters.idp.callback_params(Flow.from_dict(client.session[LOGIN_FLOW]))
|
||||
assert client.get("/auth/callback", params).status_code == 302
|
||||
assert client.get("/auth/callback", params).status_code == 403
|
||||
|
||||
|
||||
def test_a_wrong_state_is_refused(people, adapters):
|
||||
adapters.idp.sign_in_as("alice")
|
||||
client = Client()
|
||||
client.get("/auth/login")
|
||||
params = adapters.idp.callback_params(Flow.from_dict(client.session[LOGIN_FLOW]))
|
||||
response = client.get("/auth/callback", {**params, "state": "forged"})
|
||||
assert response.status_code == 403
|
||||
assert client.get("/").status_code == 302
|
||||
|
||||
|
||||
def test_a_directory_outage_blocks_new_sessions(people, adapters):
|
||||
adapters.idp.sign_in_as("alice")
|
||||
client = Client()
|
||||
client.get("/auth/login")
|
||||
params = adapters.idp.callback_params(Flow.from_dict(client.session[LOGIN_FLOW]))
|
||||
adapters.directory.unavailable = True
|
||||
response = client.get("/auth/callback", params)
|
||||
assert response.status_code == 503
|
||||
assert UserSession.objects.exclude(username="").count() == 0
|
||||
|
||||
|
||||
# -- roles ----------------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def make_request(portal_user):
|
||||
request = RequestFactory().get("/")
|
||||
request.portal_user = portal_user
|
||||
return request
|
||||
|
||||
|
||||
def portal_user(publisher=False, admin=False):
|
||||
return identity.PortalUser(SessionUser("u", "U"), "U", publisher, admin)
|
||||
|
||||
|
||||
@publisher_required
|
||||
def publisher_page(request):
|
||||
return HttpResponse("publisher")
|
||||
|
||||
|
||||
@admin_required
|
||||
def admin_page(request):
|
||||
return HttpResponse("admin")
|
||||
|
||||
|
||||
@login_required
|
||||
def any_page(request):
|
||||
return HttpResponse("page")
|
||||
|
||||
|
||||
def test_pages_need_a_signed_in_user():
|
||||
for view in (any_page, publisher_page, admin_page):
|
||||
response = view(make_request(None))
|
||||
assert response.status_code == 302 and response["Location"].startswith("/auth/login")
|
||||
|
||||
|
||||
def test_a_user_in_neither_group_is_refused_on_publisher_and_admin_pages():
|
||||
for view in (publisher_page, admin_page):
|
||||
assert view(make_request(portal_user())).status_code == 403
|
||||
|
||||
|
||||
def test_publisher_and_admin_are_separate_roles():
|
||||
assert publisher_page(make_request(portal_user(publisher=True))).status_code == 200
|
||||
assert admin_page(make_request(portal_user(publisher=True))).status_code == 403
|
||||
assert admin_page(make_request(portal_user(admin=True))).status_code == 200
|
||||
assert publisher_page(make_request(portal_user(admin=True))).status_code == 403
|
||||
|
||||
|
||||
def test_roles_come_from_the_directory_at_sign_in(people, adapters):
|
||||
client, _ = sign_in(adapters, "boss")
|
||||
assert client.session[ROLES]["publisher"] is True
|
||||
assert client.session[ROLES]["admin"] is True
|
||||
client, _ = sign_in(adapters, "alice")
|
||||
assert client.session[ROLES]["publisher"] is False
|
||||
|
||||
|
||||
def session_request(client):
|
||||
request = RequestFactory().get("/")
|
||||
request.session = client.session
|
||||
request.client_ip = "127.0.0.1"
|
||||
return request
|
||||
|
||||
|
||||
def test_roles_are_not_rechecked_within_the_cache_time(people, adapters):
|
||||
client, _ = sign_in(adapters, "pub")
|
||||
adapters.directory.unavailable = True # would fail if the directory were asked
|
||||
assert identity.current(session_request(client)).is_publisher is True
|
||||
|
||||
|
||||
def test_a_role_change_is_seen_after_the_cache_expires(people, adapters):
|
||||
client, _ = sign_in(adapters, "pub")
|
||||
adapters.directory.add_user("pub", is_publisher=False)
|
||||
assert identity.current(session_request(client)).is_publisher is True
|
||||
age_the_cache(client)
|
||||
assert identity.current(session_request(client)).is_publisher is False
|
||||
|
||||
|
||||
def test_a_directory_outage_denies_elevated_roles_but_keeps_the_session(people, adapters):
|
||||
client, _ = sign_in(adapters, "boss")
|
||||
age_the_cache(client)
|
||||
adapters.directory.unavailable = True
|
||||
user = identity.current(session_request(client))
|
||||
assert user is not None and not user.is_admin and not user.is_publisher
|
||||
assert client.get("/").status_code == 200
|
||||
|
||||
|
||||
# -- account state after sign-in -----------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.parametrize("change", ["disabled", "locked", "deleted"])
|
||||
def test_a_session_ends_at_the_next_request_when_the_account_becomes_unavailable(
|
||||
people, adapters, change
|
||||
):
|
||||
first, _ = sign_in(adapters, "alice")
|
||||
second, _ = sign_in(adapters, "alice")
|
||||
other, _ = sign_in(adapters, "pub")
|
||||
if change == "deleted":
|
||||
del adapters.directory._users["alice"]
|
||||
else:
|
||||
flag = {"disabled": {"enabled": False}, "locked": {"locked": True}}[change]
|
||||
adapters.directory.add_user("Alice", **flag)
|
||||
age_the_cache(first)
|
||||
|
||||
response = first.get("/")
|
||||
assert response.status_code == 302 and response["Location"].startswith("/auth/login")
|
||||
assert UserSession.objects.filter(username="alice").count() == 0 # every session of the user
|
||||
assert second.get("/").status_code == 302
|
||||
assert other.get("/").status_code == 200 # other users are untouched
|
||||
entry = AuditEntry.objects.get(action="session.ended")
|
||||
assert entry.actor == "alice" and entry.detail == {"reason": "account_unavailable"}
|
||||
|
||||
|
||||
# -- sign-out ---------------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_sign_out_ends_the_session_on_the_server(people, adapters):
|
||||
client, _ = sign_in(adapters, "alice")
|
||||
key = client.session.session_key
|
||||
response = client.post("/auth/logout")
|
||||
assert response["Location"] == "/auth/signed-out"
|
||||
assert not UserSession.objects.filter(session_key=key).exists()
|
||||
assert client.get("/").status_code == 302
|
||||
assert AuditEntry.objects.filter(action="signout", actor="alice").count() == 1
|
||||
|
||||
|
||||
def test_a_replayed_session_cookie_does_not_work_after_sign_out(people, adapters):
|
||||
client, _ = sign_in(adapters, "alice")
|
||||
stolen = client.cookies["sessionid"].value
|
||||
client.post("/auth/logout")
|
||||
replay = Client()
|
||||
replay.cookies["sessionid"] = stolen
|
||||
assert replay.get("/").status_code == 302
|
||||
|
||||
|
||||
def test_sign_out_only_ends_this_session(people, adapters):
|
||||
first, _ = sign_in(adapters, "alice")
|
||||
second, _ = sign_in(adapters, "alice")
|
||||
first.post("/auth/logout")
|
||||
assert second.get("/").status_code == 200
|
||||
|
||||
|
||||
def test_sign_out_needs_post_and_a_csrf_token(people, adapters):
|
||||
client, _ = sign_in(adapters, "alice")
|
||||
assert client.get("/auth/logout").status_code == 405
|
||||
strict = Client(enforce_csrf_checks=True)
|
||||
strict.cookies = client.cookies
|
||||
assert strict.post("/auth/logout").status_code == 403
|
||||
assert client.get("/").status_code == 200
|
||||
|
||||
|
||||
def test_the_signed_out_page_is_public(people):
|
||||
assert Client().get("/auth/signed-out").status_code == 200
|
||||
Reference in New Issue
Block a user