feat: implement logout route and session termination logic with Keycloak integration (303b45aa) · Commits · OOP / code / portal

src/portal_service/api/auth/routes.py

+31 −2

Changes for src/portal_service/api/auth/routes.py: 31 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -6,6 +6,8 @@ Login routes.
		GET /api/auth/callback — finish it: check the callback belongs to that login,
		trade the code for tokens, and open a session.
		GET /api/auth/me — who the session cookie belongs to.
		POST /api/auth/logout — end the session here, and say where to end it at
		Keycloak. POST, so another site cannot trigger it.

		The browser never receives a token; it gets cookies that name server-side rows.
		"""
		@@ -13,10 +15,10 @@ The browser never receives a token; it gets cookies that name server-side rows.
		from typing import Annotated

		import structlog
		from fastapi import APIRouter, Depends, HTTPException, Request, status
		from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
		from fastapi.responses import RedirectResponse

		from portal_service.api.auth.schemas import CurrentUser
		from portal_service.api.auth.schemas import CurrentUser, LogoutResponse
		from portal_service.api.auth.verification import _decode_id_token, _decode_token
		from portal_service.api.dependencies import (
		CurrentSessionDep,
		@@ -31,6 +33,7 @@ from portal_service.config import Settings
		from portal_service.domain.errors import (
		AuthenticationError,
		IdentityProviderUnavailableError,
		InvalidSessionError,
		InvalidTokenError,
		)
		from portal_service.domain.ports.identity_provider import IdentityProvider
		@@ -156,3 +159,29 @@ async def me(session: CurrentSessionDep) -> CurrentUser:
		logger.info("session token no longer usable", error=str(exc))
		raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Not authenticated") from exc
		return CurrentUser.from_claims(claims)


		@auth_router.post("/logout")
		async def logout(
		request: Request,
		response: Response,
		settings: SettingsDep,
		service: SessionServiceDep,
		) -> LogoutResponse:
		"""
		Always succeeds: logging out of a session that is already gone is not an error,
		and a browser holding a stale cookie must still end up logged out.
		"""
		logout_url: str \| None = None
		cookie = request.cookies.get(settings.session.cookie_name)
		if cookie:
		try:
		# Not refreshed on purpose: logging out must not depend on Keycloak
		# being reachable, and the ID token the logout URL needs must survive.
		logout_url = await service.end(await service.load_without_refreshing(cookie))
		logger.info("session ended")
		except InvalidSessionError:
		logger.info("logout for a session that was already gone")

		response.delete_cookie(settings.session.cookie_name, path="/")
		return LogoutResponse(logout_url=logout_url)

src/portal_service/api/auth/schemas.py

+9 −0

Changes for src/portal_service/api/auth/schemas.py: 9 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -22,3 +22,12 @@ class CurrentUser(BaseModel):
		roles=sorted(claims.roles),
		organization=claims.organization,
		)


		class LogoutResponse(BaseModel):
		"""
		Where to send the browser to end the Keycloak session too. None when we have no
		ID token to prove which session to end; the portal session is gone either way.
		"""

		logout_url: str \| None = None

src/portal_service/api/dependencies.py

+9 −1

Changes for src/portal_service/api/dependencies.py: 9 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -103,9 +103,17 @@ def get_session_service(
		sessions=sessions,
		identity_provider=identity_provider,
		session_ttl=timedelta(seconds=settings.session.max_age),
		# Where Keycloak sends the browser once it has ended its own session.
		post_logout_redirect_uri=_post_login_url(settings),
		)


		def _post_login_url(settings: Settings) -> str:
		if settings.oidc is not None and settings.oidc.post_login_url:
		return settings.oidc.post_login_url
		return "/"


		async def get_current_session(
		request: Request,
		settings: SettingsDep,
		@@ -117,7 +125,7 @@ async def get_current_session(
		raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Not authenticated")
		try:
		# Refreshed here, so every route that needs the user also gets a usable token.
		return await service.ensure_fresh(await service.load(cookie))
		return await service.ensure_fresh(await service.load_without_refreshing(cookie))
		except InvalidSessionError:
		# Same answer for every reason: malformed, unknown, expired, undecryptable, or
		# refused by Keycloak when refreshing.

src/portal_service/application/session_service.py

+22 −2

Changes for src/portal_service/application/session_service.py: 22 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -45,10 +45,12 @@ class SessionService:
		sessions: LoginSessionRepository,
		identity_provider: IdentityProvider,
		session_ttl: timedelta,
		post_logout_redirect_uri: str,
		) -> None:
		self._sessions = sessions
		self._identity_provider = identity_provider
		self._session_ttl = session_ttl
		self._post_logout_redirect_uri = post_logout_redirect_uri

		async def create(self, subject: str, issued: IssuedTokens) -> CreatedSession:
		credentials = SessionCredentials.generate()
		@@ -67,9 +69,13 @@ class SessionService:
		)
		return CreatedSession(cookie_value=credentials.to_cookie(), expires_at=expires_at)

		async def load(self, cookie_value: str) -> ActiveSession:
		async def load_without_refreshing(self, cookie_value: str) -> ActiveSession:
		"""
		Opens the session a cookie names. Every failure is the same to the caller:
		Opens the session a cookie names, as stored: the access token may already be
		expired. Callers that need a usable token pass the result to `ensure_fresh`;
		logout deliberately does not, so it works while Keycloak is unreachable.

		Every failure is the same to the caller:
		the cookie is malformed, names nothing, has expired, or does not decrypt.
		All of them mean "not logged in".
		"""
		@@ -123,3 +129,17 @@ class SessionService:
		)
		)
		return ActiveSession(session=row, tokens=tokens, credentials=active.credentials)

		async def end(self, active: ActiveSession) -> str \| None:
		"""
		Ends the session here and says where to send the browser to end it at
		Keycloak too. Without that second half the user stays logged in there, and
		clicking "sign in" would walk straight back in without a password prompt.
		"""
		await self._sessions.delete_by_id(active.session.id)
		if active.tokens.id_token is None:
		return None
		return self._identity_provider.logout_url(
		id_token=active.tokens.id_token,
		post_logout_redirect_uri=self._post_logout_redirect_uri,
		)

tests/test_logout_route.py

0 → 100644

+157 −0

Changes for tests/test_logout_route.py: 157 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		POST /api/auth/logout — ending the session here and at Keycloak.

		Logging out must always work, including for a session that is already gone: the
		browser has to end up without a cookie either way.
		"""

		from collections.abc import Callable, Iterator
		from datetime import UTC, datetime, timedelta
		from http.cookies import SimpleCookie

		import pytest
		from fastapi import FastAPI
		from fastapi.testclient import TestClient

		from portal_service.api.dependencies import get_identity_provider, get_login_session_repo
		from portal_service.application.session_crypto import SessionCredentials, encrypt_tokens
		from portal_service.domain.models import LoginSession, SessionTokens
		from tests.fakes import FakeIdentityProvider, FakeLoginSessionRepository

		SESSION_COOKIE = "oop_session"


		@pytest.fixture
		def sessions() -> FakeLoginSessionRepository:
		return FakeLoginSessionRepository()


		@pytest.fixture
		def identity_provider() -> FakeIdentityProvider:
		return FakeIdentityProvider()


		@pytest.fixture
		def client(
		app_jwt: FastAPI,
		patch_jwks: None,
		sessions: FakeLoginSessionRepository,
		identity_provider: FakeIdentityProvider,
		) -> Iterator[TestClient]:
		app_jwt.dependency_overrides[get_login_session_repo] = lambda: sessions
		app_jwt.dependency_overrides[get_identity_provider] = lambda: identity_provider
		yield TestClient(app_jwt)
		app_jwt.dependency_overrides.clear()


		def _open_session(
		client: TestClient,
		sessions: FakeLoginSessionRepository,
		access_token: str,
		*,
		id_token: str \| None = "the-id-token",
		access_token_expires_in: timedelta = timedelta(minutes=5),
		) -> SessionCredentials:
		credentials = SessionCredentials.generate()
		now = datetime.now(UTC)
		sessions.rows[credentials.session_id_hash] = LoginSession(
		id=credentials.session_id_hash,
		subject="user-123",
		access_token_expires_at=now + access_token_expires_in,
		encrypted_tokens=encrypt_tokens(
		SessionTokens(access_token=access_token, refresh_token="rt", id_token=id_token),
		credentials,
		),
		expires_at=now + timedelta(hours=1),
		)
		client.cookies.set(SESSION_COOKIE, credentials.to_cookie())
		return credentials


		def test_logout_deletes_the_session_and_clears_the_cookie(
		client: TestClient, sessions: FakeLoginSessionRepository, make_token: Callable[..., str]
		) -> None:
		credentials = _open_session(client, sessions, make_token())

		response = client.post("/api/auth/logout")

		assert response.status_code == 200
		assert credentials.session_id_hash not in sessions.rows
		cookies = SimpleCookie()
		cookies.load(response.headers["set-cookie"])
		assert cookies[SESSION_COOKIE]["max-age"] == "0"


		def test_logout_says_where_to_end_the_keycloak_session(
		client: TestClient, sessions: FakeLoginSessionRepository, make_token: Callable[..., str]
		) -> None:
		_open_session(client, sessions, make_token())

		body = client.post("/api/auth/logout").json()

		# The hint tells Keycloak which session to end, so the user is not asked.
		assert body["logout_url"] == "https://idp.test/logout?id_token_hint=the-id-token"


		def test_the_session_no_longer_answers_after_logout(
		client: TestClient, sessions: FakeLoginSessionRepository, make_token: Callable[..., str]
		) -> None:
		_open_session(client, sessions, make_token())
		client.post("/api/auth/logout")

		# The cookie the test client still holds now names nothing.
		assert client.get("/api/auth/me").status_code == 401


		def test_logout_without_a_session_is_not_an_error(client: TestClient) -> None:
		response = client.post("/api/auth/logout")

		assert response.status_code == 200
		assert response.json()["logout_url"] is None


		def test_logout_with_a_stale_cookie_still_clears_it(client: TestClient) -> None:
		client.cookies.set(SESSION_COOKIE, "not-a-real-cookie")

		response = client.post("/api/auth/logout")

		assert response.status_code == 200
		assert response.json()["logout_url"] is None
		cookies = SimpleCookie()
		cookies.load(response.headers["set-cookie"])
		assert cookies[SESSION_COOKIE]["max-age"] == "0"


		def test_a_session_without_an_id_token_still_logs_out_here(
		client: TestClient, sessions: FakeLoginSessionRepository, make_token: Callable[..., str]
		) -> None:
		credentials = _open_session(client, sessions, make_token(), id_token=None)

		body = client.post("/api/auth/logout").json()

		assert body["logout_url"] is None
		assert credentials.session_id_hash not in sessions.rows


		def test_an_expired_token_does_not_stop_logout(
		client: TestClient,
		sessions: FakeLoginSessionRepository,
		identity_provider: FakeIdentityProvider,
		make_token: Callable[..., str],
		) -> None:
		# No refresh is attempted: a revoked session must still be able to log out.
		credentials = _open_session(
		client,
		sessions,
		make_token(expires_in=-10),
		access_token_expires_in=-timedelta(seconds=10),
		)

		assert client.post("/api/auth/logout").status_code == 200
		assert identity_provider.refreshed == []
		assert credentials.session_id_hash not in sessions.rows


		def test_logout_cannot_be_triggered_by_a_get(client: TestClient) -> None:
		# A GET could be fired from any other site; a cross-site POST carries no cookie.
		assert client.get("/api/auth/logout").status_code == 405