feat: enhance login flow with state-based attempt cookies and improve error handling (3c61ec8f) · Commits · OOP / code / portal

src/portal_service/api/auth/routes.py

+35 −14

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

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

		@@ -54,6 +55,18 @@ IdentityProviderDep = Annotated[IdentityProvider, Depends(get_identity_provider)
		# nothing about which check rejected them. The exact reason is logged instead.
		LoginFailure = Literal["cancelled", "expired", "failed", "unavailable"]

		# The alphabet of the state we issue; a callback's state is checked against it before
		# it names a cookie, since it comes from the URL.
		_STATE = re.compile(r"[A-Za-z0-9_-]+")


		def _attempt_cookie(settings: Settings, state: str) -> str:
		"""
		One login cookie per login, named by its state, so logins started in two tabs
		do not overwrite each other's cookie.
		"""
		return f"{settings.session.attempt_cookie_name}_{state[:16]}"


		@auth_router.get("/login")
		async def login(service: LoginServiceDep, settings: SettingsDep) -> RedirectResponse:
		@@ -79,7 +92,7 @@ async def _start(service: LoginService, settings: Settings, *, register: bool) -
		status_code=status.HTTP_303_SEE_OTHER,
		)
		response.set_cookie(
		key=settings.session.attempt_cookie_name,
		key=_attempt_cookie(settings, started.state),
		value=started.attempt_id,
		max_age=settings.session.attempt_max_age,
		# No script needs this, and no other site may trigger its use.
		@@ -106,19 +119,22 @@ async def callback(
		state: str \| None = None,
		error: str \| None = None,
		) -> RedirectResponse:
		attempt_cookie = _attempt_cookie(settings, state) if state and _STATE.fullmatch(state) else None
		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)
		return _login_failed(settings, "cancelled" if error == "access_denied" else "failed")
		if code is None or state is None:
		return _login_failed(
		settings, "cancelled" if error == "access_denied" else "failed", attempt_cookie
		)
		if code is None or state is None or attempt_cookie is None:
		logger.warning("callback missing code or state")
		return _login_failed(settings, "failed")
		return _login_failed(settings, "failed", attempt_cookie)

		attempt_id = request.cookies.get(settings.session.attempt_cookie_name)
		attempt_id = request.cookies.get(attempt_cookie)
		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")
		return _login_failed(settings, "expired")
		return _login_failed(settings, "expired", attempt_cookie)

		try:
		# Taking the attempt deletes it, so a replayed callback finds nothing.
		@@ -137,23 +153,25 @@ 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")
		return _login_failed(settings, "expired", attempt_cookie)
		except AuthenticationError as exc:
		logger.warning("login could not be completed", error=str(exc))
		return _login_failed(settings, "failed")
		return _login_failed(settings, "failed", attempt_cookie)
		except IdentityProviderUnavailableError as exc:
		logger.error("identity provider unreachable during login", error=str(exc))
		return _login_failed(settings, "unavailable")
		return _login_failed(settings, "unavailable", attempt_cookie)

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


		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:
		def _login_failed(
		settings: Settings, reason: LoginFailure, attempt_cookie: str \| None
		) -> 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.
		@@ -165,11 +183,14 @@ def _login_failed(settings: Settings, reason: LoginFailure) -> RedirectResponse:
		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")
		if attempt_cookie is not None:
		response.delete_cookie(attempt_cookie, path="/api/auth")
		return response


		def _session_response(settings: Settings, session_cookie: str) -> RedirectResponse:
		def _session_response(
		settings: Settings, session_cookie: str, attempt_cookie: str
		) -> RedirectResponse:
		response = RedirectResponse(_frontend_url(settings), status_code=status.HTTP_303_SEE_OTHER)
		response.set_cookie(
		key=settings.session.cookie_name,
		@@ -182,7 +203,7 @@ def _session_response(settings: Settings, session_cookie: str) -> RedirectRespon
		path="/",
		)
		# The login is over: the attempt cookie has nothing left to name.
		response.delete_cookie(settings.session.attempt_cookie_name, path="/api/auth")
		response.delete_cookie(attempt_cookie, path="/api/auth")
		return response

src/portal_service/application/login_service.py

+2 −0

Changes for src/portal_service/application/login_service.py: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -28,6 +28,7 @@ class StartedLogin:

		authorization_url: str
		attempt_id: str = field(repr=False)
		state: str = field(repr=False)


		class LoginService:
		@@ -74,6 +75,7 @@ class LoginService:
		code_challenge=code_challenge(code_verifier),
		),
		attempt_id=attempt_id,
		state=state,
		)

		async def take_attempt(self, attempt_id: str) -> LoginAttempt:

src/portal_service/domain/models.py

+0 −12

Changes for src/portal_service/domain/models.py: 0 added lines, 12 removed lines.
Original line number	Diff line number	Diff line
		@@ -23,19 +23,12 @@ class TokenClaims(BaseModel):
		aud: str \| list[str] \| None = None
		exp: int
		iat: int
		jti: str \| None = None
		preferred_username: str \| None = None
		email: str \| None = None
		scope: str = ""
		nonce: str \| None = None
		# The tenant OEG scopes data by; an administrator sets it on the user.
		organization: str \| None = None
		realm_access: dict[str, Any] \| None = None
		resource_access: dict[str, Any] \| None = None

		@property
		def scopes(self) -> set[str]:
		return set(self.scope.split())

		@property
		def roles(self) -> set[str]:
		@@ -48,11 +41,6 @@ class TokenClaims(BaseModel):
		return set()
		return set(self.realm_access.get("roles", []))

		@property
		def is_human(self) -> bool:
		"""True when the token belongs to a human user (not a service account)."""
		return any(r in self.roles for r in (Role.APP_PROVIDER, Role.OPERATOR, Role.PLATFORM_ADMIN))


		class SessionTokens(BaseModel):
		"""

src/portal_service/domain/ports/identity_provider.py

+1 −1

Changes for src/portal_service/domain/ports/identity_provider.py: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -7,7 +7,7 @@ from portal_service.domain.models import IssuedTokens

		class IdentityProvider(ABC):
		"""
		The four things the login flow asks of Keycloak. Building URLs needs no
		What the login flow asks of Keycloak. Building URLs needs no
		network; the two exchanges do.
		"""

tests/conftest.py

+2 −2

Changes for tests/conftest.py: 2 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -33,7 +33,7 @@ _BASE_ENV = {
		"OIDC__ISSUER": ISSUER,
		"OIDC__CLIENT_ID": CLIENT_ID,
		"OIDC__CLIENT_SECRET": "test-client-secret",
		"OIDC__REDIRECT_URI": "http://localhost:8000/v1/auth/callback",
		"OIDC__REDIRECT_URI": "http://localhost:8000/api/auth/callback",
		"OIDC__POST_LOGIN_URL": "http://localhost:3000",
		}

		@@ -118,7 +118,7 @@ def oidc_settings() -> OIDCSettings:
		issuer=ISSUER,
		client_id=CLIENT_ID,
		client_secret="test-client-secret",
		redirect_uri="http://localhost:8000/v1/auth/callback",
		redirect_uri="http://localhost:8000/api/auth/callback",
		)