feat: add the login route that redirects to Keycloak (dfd21140) · Commits · OOP / code / portal

src/portal_service/api/auth/routes.py

0 → 100644

+50 −0

Changes for src/portal_service/api/auth/routes.py: 50 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Login routes.

		GET /api/auth/login — start a login: remember the one-time values, send the
		browser to Keycloak.

		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, status
		from fastapi.responses import RedirectResponse

		from portal_service.api.dependencies import SettingsDep, get_login_service
		from portal_service.application.login_service import LoginService

		logger: structlog.BoundLogger = structlog.get_logger(__name__)

		auth_router = APIRouter(prefix="/api/auth", tags=["auth"])


		@auth_router.get("/login")
		async def login(
		service: Annotated[LoginService, Depends(get_login_service)],
		settings: SettingsDep,
		) -> RedirectResponse:
		started = await service.start()
		logger.info("login started")

		response = RedirectResponse(
		started.authorization_url,
		# 303 turns the browser's request into a GET, which is what Keycloak expects.
		status_code=status.HTTP_303_SEE_OTHER,
		)
		response.set_cookie(
		key=settings.session.attempt_cookie_name,
		value=started.attempt_id,
		max_age=settings.session.attempt_max_age,
		# No script needs this, and no other site may trigger its use.
		httponly=True,
		secure=settings.session.https_only,
		# Lax, not Strict: the browser arrives back from Keycloak through a
		# cross-site redirect, and Strict would withhold the cookie exactly then.
		samesite="lax",
		# Only the auth routes ever read it.
		path="/api/auth",
		)
		return response

src/portal_service/api/dependencies.py

+65 −2

Changes for src/portal_service/api/dependencies.py: 65 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		from collections.abc import AsyncGenerator
		from datetime import timedelta
		from typing import Annotated, cast

		from fastapi import Depends, Request
		from fastapi import Depends, HTTPException, Request, status
		from sqlalchemy import text
		from sqlalchemy.ext.asyncio import AsyncEngine
		from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession

		from portal_service.adapters.database.repos import (
		SqlLoginAttemptRepository,
		SqlLoginSessionRepository,
		)
		from portal_service.adapters.keycloak.oidc_client import KeycloakOIDCClient
		from portal_service.app_state import AppState
		from portal_service.application.login_service import LoginService
		from portal_service.config import OIDCSettings, Settings, get_settings
		from portal_service.domain.ports.database.login_attempts import LoginAttemptRepository
		from portal_service.domain.ports.database.login_sessions import LoginSessionRepository
		from portal_service.domain.ports.identity_provider import IdentityProvider


		def get_app_state(request: Request) -> AppState:
		@@ -24,3 +36,54 @@ async def get_database_health(
		return True
		except Exception:
		return False


		async def get_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
		"""One transaction per request: committed if the route succeeds, rolled back if not."""
		async with get_app_state(request=request).session_maker() as session:
		try:
		yield session
		await session.commit()
		except Exception:
		await session.rollback()
		raise


		SessionDep = Annotated[AsyncSession, Depends(get_session)]
		SettingsDep = Annotated[Settings, Depends(get_settings)]


		def get_login_attempt_repo(session: SessionDep) -> LoginAttemptRepository:
		return SqlLoginAttemptRepository(session)


		def get_login_session_repo(session: SessionDep) -> LoginSessionRepository:
		return SqlLoginSessionRepository(session)


		def get_oidc_settings(settings: SettingsDep) -> OIDCSettings:
		if settings.oidc is None:
		# Running with AUTH__SECURITY_MODE=none and no identity provider configured.
		raise HTTPException(
		status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
		detail="Login is not configured",
		)
		return settings.oidc


		def get_identity_provider(
		oidc: Annotated[OIDCSettings, Depends(get_oidc_settings)],
		) -> IdentityProvider:
		return KeycloakOIDCClient(oidc)


		def get_login_service(
		attempts: Annotated[LoginAttemptRepository, Depends(get_login_attempt_repo)],
		identity_provider: Annotated[IdentityProvider, Depends(get_identity_provider)],
		settings: SettingsDep,
		) -> LoginService:
		return LoginService(
		attempts=attempts,
		identity_provider=identity_provider,
		attempt_ttl=timedelta(seconds=settings.session.attempt_max_age),
		)

src/portal_service/application/login_service.py

0 → 100644

+66 −0

Changes for src/portal_service/application/login_service.py: 66 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Starting a login.

		Generates the one-time values, remembers them server-side, and says where to send
		the browser. The route deals with cookies and redirects; this deals with the flow.
		"""

		from dataclasses import dataclass, field
		from datetime import UTC, datetime, timedelta

		from portal_service.application.pkce import (
		code_challenge,
		new_code_verifier,
		new_nonce,
		new_state,
		)
		from portal_service.application.session_crypto import hash_id, new_id
		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


		@dataclass(frozen=True)
		class StartedLogin:
		"""Where to send the browser, and the id the login cookie must carry."""

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


		class LoginService:
		def __init__(
		self,
		attempts: LoginAttemptRepository,
		identity_provider: IdentityProvider,
		attempt_ttl: timedelta,
		) -> None:
		self._attempts = attempts
		self._identity_provider = identity_provider
		self._attempt_ttl = attempt_ttl

		async def start(self) -> StartedLogin:
		code_verifier = new_code_verifier()
		state = new_state()
		nonce = new_nonce()
		attempt_id = new_id()

		# Saved before the redirect: a callback that arrives without a matching
		# attempt is rejected, so the values can never be supplied by the caller.
		await self._attempts.save(
		LoginAttempt(
		id=hash_id(attempt_id),
		state=state,
		nonce=nonce,
		code_verifier=code_verifier,
		expires_at=datetime.now(UTC) + self._attempt_ttl,
		)
		)
		return StartedLogin(
		authorization_url=self._identity_provider.authorization_url(
		state=state,
		nonce=nonce,
		code_challenge=code_challenge(code_verifier),
		),
		attempt_id=attempt_id,
		)

src/portal_service/application/session_crypto.py

+14 −1

Changes for src/portal_service/application/session_crypto.py: 14 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -39,7 +39,7 @@ class SessionCredentials:
		@property
		def session_id_hash(self) -> str:
		"""Primary key of the login_sessions row."""
		return hashlib.sha256(self.session_id.encode()).hexdigest()
		return hash_id(self.session_id)

		def to_cookie(self) -> str:
		return f"{self.session_id}.{_b64encode(self.key)}"
		@@ -56,6 +56,19 @@ class SessionCredentials:
		return cls(session_id=session_id, key=key)


		def new_id() -> str:
		"""A random identifier for a cookie: 32 bytes, URL-safe."""
		return secrets.token_urlsafe(_KEY_BYTES)


		def hash_id(value: str) -> str:
		"""
		What the database stores instead of the identifier itself, so a stolen table
		cannot be turned back into cookies.
		"""
		return hashlib.sha256(value.encode()).hexdigest()


		def encrypt_tokens(tokens: SessionTokens, credentials: SessionCredentials) -> bytes:
		"""Returns nonce + ciphertext. A fresh nonce each call, so re-encrypting after a
		token refresh with the same key is safe."""

src/portal_service/config.py

+5 −2

Changes for src/portal_service/config.py: 5 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -50,9 +50,12 @@ class OIDCSettings(BaseModel):


		class SessionSettings(BaseModel):
		secret_key: str
		cookie_name: str = "oop_session"
		max_age: int = 3600
		# The cookie naming a login in flight: only needs to outlive the trip to Keycloak.
		attempt_cookie_name: str = "oop_login"
		attempt_max_age: int = 300
		# True in the cluster, where the portal is served over HTTPS.
		https_only: bool = False


		@@ -75,7 +78,7 @@ class Settings(BaseSettings):

		auth: AuthSettings = AuthSettings()
		oidc: OIDCSettings \| None = None
		session: SessionSettings
		session: SessionSettings = SessionSettings()
		postgresql_settings: PostgreSQLSettings = PostgreSQLSettings()
		cors: CORSSettings = CORSSettings()