151 lines
5.6 KiB
Python
151 lines
5.6 KiB
Python
"""SessionService: state validation and session rotation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from ragcore.application.auth import InvalidStateError, PendingLogin, SessionService
|
|
from ragcore.application.auth.ports import AuthorizationChallenge
|
|
from ragcore.domain.auth import OidcClaims
|
|
from ragcore.domain.control import Role
|
|
|
|
|
|
class FakeOidcProvider:
|
|
def __init__(self, claims: OidcClaims) -> None:
|
|
self.claims = claims
|
|
self.challenge_calls: list[str] = []
|
|
self.login_calls: list[tuple[str, str, str, str]] = []
|
|
|
|
async def build_authorization_challenge(self, *, redirect_uri: str) -> AuthorizationChallenge:
|
|
self.challenge_calls.append(redirect_uri)
|
|
return AuthorizationChallenge(
|
|
redirect_url="https://idp.example.test/authorize?...",
|
|
state=f"state-{len(self.challenge_calls)}",
|
|
nonce=f"nonce-{len(self.challenge_calls)}",
|
|
code_verifier=f"verifier-{len(self.challenge_calls)}",
|
|
)
|
|
|
|
async def complete_login(
|
|
self, *, code: str, code_verifier: str, nonce: str, redirect_uri: str
|
|
) -> OidcClaims:
|
|
self.login_calls.append((code, code_verifier, nonce, redirect_uri))
|
|
return self.claims
|
|
|
|
|
|
def _service(
|
|
provider: FakeOidcProvider, *, group_roles: Mapping[str, Role] | None = None
|
|
) -> SessionService:
|
|
return SessionService(
|
|
provider=provider,
|
|
group_roles=group_roles or {"ragcore-editors": Role.EDITOR},
|
|
workspace_id=uuid4(),
|
|
)
|
|
|
|
|
|
async def test_start_login_returns_a_redirect_and_pending_login() -> None:
|
|
provider = FakeOidcProvider(OidcClaims(subject="alice", email=None, groups=()))
|
|
service = _service(provider)
|
|
|
|
start = await service.start_login(redirect_uri="https://ragcore.example.test/auth/callback")
|
|
|
|
assert start.redirect_url.startswith("https://idp.example.test/")
|
|
assert start.pending.state
|
|
assert start.pending.nonce
|
|
assert start.pending.code_verifier
|
|
assert provider.challenge_calls == ["https://ragcore.example.test/auth/callback"]
|
|
|
|
|
|
async def test_complete_login_mints_a_session_with_mapped_roles() -> None:
|
|
provider = FakeOidcProvider(
|
|
OidcClaims(subject="alice", email="alice@example.test", groups=("ragcore-editors",))
|
|
)
|
|
service = _service(provider)
|
|
start = await service.start_login(redirect_uri="https://ragcore.example.test/auth/callback")
|
|
|
|
session = await service.complete_login(
|
|
code="auth-code",
|
|
state=start.pending.state,
|
|
pending=start.pending,
|
|
redirect_uri="https://ragcore.example.test/auth/callback",
|
|
)
|
|
|
|
assert session.subject == "alice"
|
|
assert session.roles == frozenset({Role.EDITOR})
|
|
assert session.csrf_token
|
|
assert not session.is_expired
|
|
assert provider.login_calls == [
|
|
("auth-code", start.pending.code_verifier, start.pending.nonce, "https://ragcore.example.test/auth/callback")
|
|
]
|
|
|
|
|
|
async def test_mismatched_state_is_rejected() -> None:
|
|
provider = FakeOidcProvider(OidcClaims(subject="alice", email=None, groups=()))
|
|
service = _service(provider)
|
|
start = await service.start_login(redirect_uri="https://ragcore.example.test/auth/callback")
|
|
|
|
with pytest.raises(InvalidStateError):
|
|
await service.complete_login(
|
|
code="auth-code",
|
|
state="a-completely-different-state",
|
|
pending=start.pending,
|
|
redirect_uri="https://ragcore.example.test/auth/callback",
|
|
)
|
|
# The provider is never even asked to exchange the code once state fails.
|
|
assert provider.login_calls == []
|
|
|
|
|
|
async def test_two_logins_never_reuse_the_same_session_id_session_rotation() -> None:
|
|
"""Session rotation's standard security meaning: the post-authentication
|
|
session identifier is never the pending (pre-authentication) one, and a
|
|
fresh login always mints a brand-new session -- never resurrects or
|
|
reuses a previous one, defeating session-fixation attacks.
|
|
"""
|
|
|
|
provider = FakeOidcProvider(OidcClaims(subject="alice", email=None, groups=()))
|
|
service = _service(provider)
|
|
|
|
first_start = await service.start_login(redirect_uri="https://ragcore.example.test/auth/callback")
|
|
first_session = await service.complete_login(
|
|
code="code-1",
|
|
state=first_start.pending.state,
|
|
pending=first_start.pending,
|
|
redirect_uri="https://ragcore.example.test/auth/callback",
|
|
)
|
|
|
|
second_start = await service.start_login(redirect_uri="https://ragcore.example.test/auth/callback")
|
|
second_session = await service.complete_login(
|
|
code="code-2",
|
|
state=second_start.pending.state,
|
|
pending=second_start.pending,
|
|
redirect_uri="https://ragcore.example.test/auth/callback",
|
|
)
|
|
|
|
assert first_session.id != second_session.id
|
|
assert first_session.csrf_token != second_session.csrf_token
|
|
|
|
|
|
async def test_unrecognized_groups_produce_a_roleless_session() -> None:
|
|
provider = FakeOidcProvider(
|
|
OidcClaims(subject="mallory", email=None, groups=("not-a-real-group",))
|
|
)
|
|
service = _service(provider, group_roles={"ragcore-admins": Role.PLATFORM_ADMIN})
|
|
start = await service.start_login(redirect_uri="https://ragcore.example.test/auth/callback")
|
|
|
|
session = await service.complete_login(
|
|
code="auth-code",
|
|
state=start.pending.state,
|
|
pending=start.pending,
|
|
redirect_uri="https://ragcore.example.test/auth/callback",
|
|
)
|
|
|
|
assert session.roles == frozenset()
|
|
|
|
|
|
def test_pending_login_is_an_immutable_value() -> None:
|
|
pending = PendingLogin(state="s", nonce="n", code_verifier="v")
|
|
with pytest.raises(AttributeError):
|
|
pending.state = "other" # type: ignore[misc]
|