feat: implement login attempt repository and mapper with integration tests (80ee3c7b) · Commits · OOP / code / portal

pyproject.toml

+1 −0

Changes for pyproject.toml: 1 added line, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -26,6 +26,7 @@ dev = [
		"pytest>=8.0.0",
		"pytest-asyncio>=0.25.0",
		"ruff>=0.11.0",
		"testcontainers>=4.0.0",
		]

		[tool.ruff]

src/portal_service/adapters/database/mappers.py

+26 −2

Changes for src/portal_service/adapters/database/mappers.py: 26 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		from portal_service.adapters.database.sql import LoginSessionRow
		from portal_service.domain.models import LoginSession
		from portal_service.adapters.database.sql import LoginAttemptRow, LoginSessionRow
		from portal_service.domain.models import LoginAttempt, LoginSession


		class LoginSessionMapper:
		@@ -24,3 +24,27 @@ class LoginSessionMapper:
		tokens=domain.encrypted_tokens,
		expires_at=domain.expires_at,
		)


		class LoginAttemptMapper:
		@staticmethod
		def to_domain(row: LoginAttemptRow) -> LoginAttempt:
		return LoginAttempt(
		id=row.id,
		state=row.state,
		nonce=row.nonce,
		code_verifier=row.code_verifier,
		expires_at=row.expires_at,
		created_at=row.created_at,
		)

		@staticmethod
		def to_row(domain: LoginAttempt) -> LoginAttemptRow:
		# created_at is left out: the database sets it on insert.
		return LoginAttemptRow(
		id=domain.id,
		state=domain.state,
		nonce=domain.nonce,
		code_verifier=domain.code_verifier,
		expires_at=domain.expires_at,
		)

src/portal_service/adapters/database/repos/init.py

+2 −0

Changes for src/portal_service/adapters/database/repos/__init__.py: 2 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from portal_service.adapters.database.repos.login_attempts import SqlLoginAttemptRepository
		from portal_service.adapters.database.repos.login_sessions import SqlLoginSessionRepository

		__all__ = [
		"SqlLoginAttemptRepository",
		"SqlLoginSessionRepository",
		]

src/portal_service/adapters/database/repos/login_attempts.py

0 → 100644

+41 −0

Changes for src/portal_service/adapters/database/repos/login_attempts.py: 41 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime
		from typing import Any, cast

		from sqlalchemy import CursorResult, delete, select
		from sqlalchemy.ext.asyncio import AsyncSession

		from portal_service.adapters.database.mappers import LoginAttemptMapper
		from portal_service.adapters.database.sql import LoginAttemptRow
		from portal_service.domain.models import LoginAttempt
		from portal_service.domain.ports.database.login_attempts import LoginAttemptRepository


		class SqlLoginAttemptRepository(LoginAttemptRepository):
		def __init__(self, session: AsyncSession) -> None:
		self._session = session

		async def save(self, attempt: LoginAttempt) -> LoginAttempt:
		merged = await self._session.merge(LoginAttemptMapper.to_row(attempt))
		await self._session.flush()
		stmt = select(LoginAttemptRow).where(LoginAttemptRow.id == merged.id)
		saved = await self._session.scalar(stmt)
		if saved is None:
		raise RuntimeError("Saved login attempt could not be reloaded")
		return LoginAttemptMapper.to_domain(saved)

		async def take(self, attempt_id_hash: str) -> LoginAttempt \| None:
		# Read and delete in one statement: two callbacks racing with the same id
		# cannot both get a row, so an attempt is usable exactly once.
		stmt = (
		delete(LoginAttemptRow)
		.where(LoginAttemptRow.id == attempt_id_hash)
		.returning(LoginAttemptRow)
		.execution_options(synchronize_session=False)
		)
		row: LoginAttemptRow \| None = (await self._session.execute(stmt)).scalar_one_or_none()
		return LoginAttemptMapper.to_domain(row) if row is not None else None

		async def delete_expired(self, now: datetime) -> int:
		stmt = delete(LoginAttemptRow).where(LoginAttemptRow.expires_at <= now)
		result = cast(CursorResult[Any], await self._session.execute(stmt))
		return result.rowcount

src/portal_service/adapters/database/sql.py

+22 −0

Changes for src/portal_service/adapters/database/sql.py: 22 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -45,3 +45,25 @@ class LoginSessionRow(Base):
		created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
		# Hard end of the session, whatever happens to the tokens; expired rows get purged.
		expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))


		class LoginAttemptRow(Base):
		"""
		One login in progress: from the redirect to Keycloak until the callback returns.
		Rows are short-lived and single-use — the callback deletes the row it reads.
		"""

		__tablename__ = "login_attempts"
		__table_args__ = (Index("idx_login_attempts_expires_at", "expires_at"),)

		# SHA-256 hex of the attempt id from the short-lived cookie.
		id: Mapped[str] = mapped_column(String(64), primary_key=True)
		# Returned by Keycloak in the callback; must match to accept it.
		state: Mapped[str] = mapped_column(String(128))
		# Carried inside the ID token; must match to accept it.
		nonce: Mapped[str] = mapped_column(String(128))
		# The PKCE secret, presented at the code exchange (RFC 7636 allows up to 128).
		code_verifier: Mapped[str] = mapped_column(String(128))
		created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
		# A few minutes: long enough to type a password, short enough to limit replay.
		expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))