feat: enhance login callback handling with detailed failure reasons and new error class (c7d1ee33) · Commits · OOP / code / portal

src/portal_service/api/auth/routes.py

+34 −11

Changes for src/portal_service/api/auth/routes.py: 34 added lines, 11 removed lines.
Original line number	Diff line number	Diff line
		@@ -13,7 +13,8 @@ Login routes.
		The browser never receives a token; it gets cookies that name server-side rows.
		"""

		from typing import Annotated
		from typing import Annotated, Literal
		from urllib.parse import urlencode

		import structlog
		from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
		@@ -37,6 +38,7 @@ from portal_service.domain.errors import (
		IdentityProviderUnavailableError,
		InvalidSessionError,
		InvalidTokenError,
		LoginAttemptExpiredError,
		)
		from portal_service.domain.ports.identity_provider import IdentityProvider

		@@ -48,7 +50,9 @@ LoginServiceDep = Annotated[LoginService, Depends(get_login_service)]
		SessionServiceDep = Annotated[SessionService, Depends(get_session_service)]
		IdentityProviderDep = Annotated[IdentityProvider, Depends(get_identity_provider)]

		_LOGIN_FAILED = "Login failed, please try again"
		# Why a login failed, as the GUI is told it: coarse on purpose, so a caller learns
		# nothing about which check rejected them. The exact reason is logged instead.
		LoginFailure = Literal["cancelled", "expired", "failed", "unavailable"]


		@auth_router.get("/login")
		@@ -104,15 +108,16 @@ async def callback(
		if error is not None:
		# The user cancelled, or Keycloak refused before any code was issued.
		logger.warning("keycloak did not issue a code", oauth_error=error)
		raise HTTPException(status.HTTP_400_BAD_REQUEST, _LOGIN_FAILED)
		return _login_failed(settings, "cancelled" if error == "access_denied" else "failed")
		if code is None or state is None:
		logger.warning("callback missing code or state")
		raise HTTPException(status.HTTP_400_BAD_REQUEST, _LOGIN_FAILED)
		return _login_failed(settings, "failed")

		attempt_id = request.cookies.get(settings.session.attempt_cookie_name)
		if not attempt_id:
		# The login cookie outlives the trip to Keycloak by minutes, then the browser drops it.
		logger.warning("callback without a login cookie")
		raise HTTPException(status.HTTP_400_BAD_REQUEST, _LOGIN_FAILED)
		return _login_failed(settings, "expired")

		try:
		# Taking the attempt deletes it, so a replayed callback finds nothing.
		@@ -129,24 +134,42 @@ async def callback(
		)

		created = await session_service.create(subject=claims.sub, issued=issued)
		except LoginAttemptExpiredError as exc:
		logger.warning("login could not be completed", error=str(exc))
		return _login_failed(settings, "expired")
		except AuthenticationError as exc:
		logger.warning("login could not be completed", error=str(exc))
		raise HTTPException(status.HTTP_400_BAD_REQUEST, _LOGIN_FAILED) from exc
		return _login_failed(settings, "failed")
		except IdentityProviderUnavailableError as exc:
		logger.error("identity provider unreachable during login", error=str(exc))
		raise HTTPException(
		status.HTTP_503_SERVICE_UNAVAILABLE, "Authentication service unavailable"
		) from exc
		return _login_failed(settings, "unavailable")

		logger.info("login completed", subject=claims.sub)
		return _session_response(settings, created.cookie_value)


		def _session_response(settings: Settings, session_cookie: str) -> RedirectResponse:
		def _frontend_url(settings: Settings) -> str:
		return settings.oidc.post_login_url if settings.oidc and settings.oidc.post_login_url else "/"


		def _login_failed(settings: Settings, reason: LoginFailure) -> RedirectResponse:
		"""
		The callback is a page a person lands on, so a failure sends them back to the GUI
		with a reason it can explain, rather than an error document. The detail is logged.
		"""
		url = _frontend_url(settings)
		separator = "&" if "?" in url else "?"
		response = RedirectResponse(
		settings.oidc.post_login_url if settings.oidc and settings.oidc.post_login_url else "/",
		f"{url}{separator}{urlencode({'error': reason})}",
		status_code=status.HTTP_303_SEE_OTHER,
		)
		# Whatever went wrong, this login is over: the next attempt starts afresh.
		response.delete_cookie(settings.session.attempt_cookie_name, path="/api/auth")
		return response


		def _session_response(settings: Settings, session_cookie: str) -> RedirectResponse:
		response = RedirectResponse(_frontend_url(settings), status_code=status.HTTP_303_SEE_OTHER)
		response.set_cookie(
		key=settings.session.cookie_name,
		value=session_cookie,

src/portal_service/application/login_service.py

+3 −3

Changes for src/portal_service/application/login_service.py: 3 added lines, 3 removed lines.
Original line number	Diff line number	Diff line
		@@ -16,7 +16,7 @@ from portal_service.application.pkce import (
		new_state,
		)
		from portal_service.application.session_crypto import hash_id, new_id
		from portal_service.domain.errors import LoginAttemptError
		from portal_service.domain.errors import LoginAttemptError, LoginAttemptExpiredError
		from portal_service.domain.models import LoginAttempt
		from portal_service.domain.ports.database.login_attempts import LoginAttemptRepository
		from portal_service.domain.ports.identity_provider import IdentityProvider
		@@ -83,9 +83,9 @@ class LoginService:
		"""
		attempt = await self._attempts.take(hash_id(attempt_id))
		if attempt is None:
		raise LoginAttemptError("No login is in progress")
		raise LoginAttemptExpiredError("No login is in progress")
		if attempt.expires_at <= datetime.now(UTC):
		raise LoginAttemptError("The login took too long")
		raise LoginAttemptExpiredError("The login took too long")
		return attempt

		@staticmethod

src/portal_service/domain/errors.py

+5 −0

Changes for src/portal_service/domain/errors.py: 5 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -25,6 +25,11 @@ class LoginAttemptError(AuthenticationError):
		pass


		# The login this callback belongs to is gone: never started, already finished, or too old.
		class LoginAttemptExpiredError(LoginAttemptError):
		pass


		# Keycloak refused a login code or refresh token: expired, already used, or not ours.
		class InvalidGrantError(AuthenticationError):
		pass

tests/test_callback_route.py

+27 −8

Changes for tests/test_callback_route.py: 27 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		@@ -9,6 +9,7 @@ ID token verification and the session encryption all run for real.
		from collections.abc import Callable, Iterator
		from http.cookies import SimpleCookie
		from typing import Any
		from urllib.parse import parse_qs, urlparse

		import pytest
		from fastapi import FastAPI
		@@ -87,6 +88,23 @@ def _callback(client: TestClient, attempt: LoginAttempt, **overrides: Any) -> An
		return client.get("/api/auth/callback", params=params, follow_redirects=False)


		def _failure(response: Any) -> str:
		"""
		A failed callback sends the person back to the GUI with a reason, never an error
		document, and never with a session cookie. Returns the reason.
		"""
		assert response.status_code == 303
		location = urlparse(response.headers["location"])
		assert f"{location.scheme}://{location.netloc}{location.path}" == "http://localhost:3000"
		cookies = SimpleCookie()
		for header in response.headers.get_list("set-cookie"):
		cookies.load(header)
		assert SESSION_COOKIE not in cookies
		assert cookies[ATTEMPT_COOKIE]["max-age"] == "0"
		reason: str = parse_qs(location.query)["error"][0]
		return reason


		# ------------------------------------------------------------------- the happy path


		@@ -181,7 +199,7 @@ def test_a_callback_for_a_different_login_is_refused(

		response = _callback(client, attempt, state="someone-elses-state")

		assert response.status_code == 400
		assert _failure(response) == "failed"
		assert sessions.rows == {}
		assert identity_provider.exchanged == []

		@@ -199,7 +217,8 @@ def test_a_replayed_callback_is_refused(
		assert _callback(client, attempt).status_code == 303
		second = _callback(client, attempt)

		assert second.status_code == 400
		# The attempt was used up by the first callback.
		assert _failure(second) == "expired"
		assert len(sessions.rows) == 1


		@@ -209,7 +228,7 @@ def test_a_callback_without_the_login_cookie_is_refused(
		attempt = _start_login(client, attempts)
		client.cookies.delete(ATTEMPT_COOKIE)

		assert _callback(client, attempt).status_code == 400
		assert _failure(_callback(client, attempt)) == "expired"


		def test_an_id_token_from_another_login_is_refused(
		@@ -224,7 +243,7 @@ def test_an_id_token_from_another_login_is_refused(

		response = _callback(client, attempt)

		assert response.status_code == 400
		assert _failure(response) == "failed"
		assert sessions.rows == {}


		@@ -239,7 +258,7 @@ def test_a_response_without_an_id_token_is_refused(

		response = _callback(client, attempt)

		assert response.status_code == 400
		assert _failure(response) == "failed"
		assert sessions.rows == {}


		@@ -251,7 +270,7 @@ def test_keycloak_refusing_the_code_is_a_failed_login(
		attempt = _start_login(client, attempts)
		identity_provider.error = InvalidGrantError("code already used")

		assert _callback(client, attempt).status_code == 400
		assert _failure(_callback(client, attempt)) == "failed"


		def test_keycloak_being_down_is_not_the_users_fault(
		@@ -262,7 +281,7 @@ def test_keycloak_being_down_is_not_the_users_fault(
		attempt = _start_login(client, attempts)
		identity_provider.error = IdentityProviderUnavailableError("token endpoint unreachable")

		assert _callback(client, attempt).status_code == 503
		assert _failure(_callback(client, attempt)) == "unavailable"


		def test_a_cancelled_login_is_refused(
		@@ -276,4 +295,4 @@ def test_a_cancelled_login_is_refused(
		follow_redirects=False,
		)

		assert response.status_code == 400
		assert _failure(response) == "cancelled"