feat: implement authentication and authorization modules with JWT support (e1a1c1f7) · Commits · OOP / code / portal

.pre-commit-config.yaml

+1 −1

Changes for .pre-commit-config.yaml: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -27,7 +27,7 @@ repos:
		"fastapi[standard]>=0.115.0",
		"httpx>=0.28.0",
		"pydantic-settings>=2.9.1",
		"PyJWT[cryptography]>=2.10.1",
		"PyJWT[crypto]>=2.10.1",
		"pytest>=8.0.0",
		"pytest-asyncio>=0.25.0",
		"structlog>=25.0.0",

pyproject.toml

+1 −1

Changes for pyproject.toml: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -11,7 +11,7 @@ dependencies = [
		"fastapi[standard]>=0.115.0",
		"pydantic-settings>=2.9.1",
		"structlog>=25.0.0",
		"PyJWT[cryptography]>=2.10.1",
		"PyJWT[crypto]>=2.10.1",
		"httpx>=0.28.0",
		]

src/portal_service/api/auth/init.py

0 → 100644

+32 −0

Changes for src/portal_service/api/auth/__init__.py: 32 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Authentication & authorization dependencies.

		Organised into focused modules:
		* verification — JWT/JWKS decode
		* dependencies — request-time deps: bearer
		* guards — role/scope authorization

		Public names are re-exported here so callers keep importing from
		`portal_service.api.auth`.
		"""

		from portal_service.api.auth.dependencies import (
		BearerDep,
		CurrentUserDep,
		get_current_user,
		)
		from portal_service.api.auth.guards import (
		require_any_role,
		require_scopes,
		)
		from portal_service.api.auth.verification import _decode_token, _get_jwks_client

		__all__ = [
		"BearerDep",
		"CurrentUserDep",
		"get_current_user",
		"require_any_role",
		"require_scopes",
		"_decode_token",
		"_get_jwks_client",
		]

src/portal_service/api/auth/dependencies.py

0 → 100644

+74 −0

Changes for src/portal_service/api/auth/dependencies.py: 74 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		FastAPI authentication dependencies.

		Bearer-token path for machine-to-machine / API clients:
		* bearer — `Authorization: Bearer <jwt>`

		When AUTH__SECURITY_MODE=none every check is bypassed and the dependency returns
		None, so the app runs without any Keycloak instance.
		"""

		from typing import Annotated

		import structlog
		from fastapi import Depends, HTTPException, status
		from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

		from portal_service.api.auth.verification import _decode_token
		from portal_service.config import AuthSettings, SecurityMode, get_settings
		from portal_service.domain.errors import (
		AuthenticationError,
		IdentityProviderUnavailableError,
		TokenExpiredError,
		)
		from portal_service.domain.models import TokenClaims

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

		_bearer = HTTPBearer(auto_error=False)


		def _auth_settings() -> AuthSettings:
		return get_settings().auth


		BearerDep = Annotated[HTTPAuthorizationCredentials \| None, Depends(_bearer)]


		def get_current_user(
		credentials: BearerDep,
		auth_settings: Annotated[AuthSettings, Depends(_auth_settings)],
		) -> TokenClaims \| None:
		if auth_settings.security_mode == SecurityMode.NONE:
		logger.warning("security disabled — skipping auth")
		return None
		if credentials is None:
		raise HTTPException(
		status_code=status.HTTP_401_UNAUTHORIZED,
		detail="Not authenticated",
		headers={"WWW-Authenticate": "Bearer"},
		)
		try:
		return _decode_token(credentials.credentials)
		except TokenExpiredError:
		raise HTTPException(
		status_code=status.HTTP_401_UNAUTHORIZED,
		detail="Token has expired",
		headers={"WWW-Authenticate": "Bearer"},
		)
		except AuthenticationError as exc:
		logger.warning("bearer token invalid", error=str(exc))
		raise HTTPException(
		status_code=status.HTTP_401_UNAUTHORIZED,
		detail="Could not validate credentials",
		headers={"WWW-Authenticate": "Bearer"},
		)
		except IdentityProviderUnavailableError as exc:
		logger.error("identity provider unreachable", error=str(exc))
		raise HTTPException(
		status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
		detail="Authentication service unavailable",
		)


		CurrentUserDep = Annotated[TokenClaims \| None, Depends(get_current_user)]

src/portal_service/api/auth/guards.py

0 → 100644

+93 −0

Changes for src/portal_service/api/auth/guards.py: 93 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		"""
		Role/scope authorization guards.

		These live in the HTTP layer and raise HTTPException directly (no domain-error
		indirection needed here). Each guard bypasses checks when security is disabled.
		"""

		from typing import Annotated, Any

		import structlog
		from fastapi import Depends, HTTPException, status

		from portal_service.api.auth.dependencies import (
		CurrentUserDep,
		_auth_settings,
		)
		from portal_service.config import AuthSettings, SecurityMode
		from portal_service.domain.models import Role, TokenClaims

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


		def _require(kind: str, required_set: frozenset[str], actual: set[str], user: TokenClaims) -> None:
		"""All-of check: raise 403 if any item from required_set is absent from actual."""
		missing = required_set - actual
		if missing:
		logger.warning(
		f"authorization failed: missing {kind}",
		required=list(required_set),
		present=list(actual),
		sub=user.sub,
		)
		raise HTTPException(
		status_code=status.HTTP_403_FORBIDDEN,
		detail=f"Missing required {kind}: {', '.join(sorted(missing))}",
		)


		def _require_any(kind: str, accepted: frozenset[str], actual: set[str], user: TokenClaims) -> None:
		"""Any-of check: raise 403 unless at least one item from accepted is in actual."""
		if not accepted & actual:
		logger.warning(
		f"authorization failed: no accepted {kind}",
		accepted=sorted(accepted),
		present=sorted(actual),
		sub=user.sub,
		)
		raise HTTPException(
		status_code=status.HTTP_403_FORBIDDEN,
		detail=f"Requires one of {kind}: {', '.join(sorted(accepted))}",
		)


		def require_any_role(*accepted: Role) -> Any:
		"""Allow the request if the user holds at least one of the given roles."""
		if not accepted:
		raise ValueError("require_any_role() needs at least one role")
		accepted_set: frozenset[str] = frozenset(r.value for r in accepted)

		def _check(
		user: CurrentUserDep,
		auth_settings: Annotated[AuthSettings, Depends(_auth_settings)],
		) -> TokenClaims \| None:
		if auth_settings.security_mode == SecurityMode.NONE:
		return None
		if user is None:
		raise HTTPException(
		status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated"
		)
		_require_any("roles", accepted_set, user.roles, user)
		return user

		return Depends(_check)


		def require_scopes(*required: str) -> Any:
		"""Allow the request only if the token carries every one of the given scopes."""
		required_set: frozenset[str] = frozenset(required)

		def _check(
		user: CurrentUserDep,
		auth_settings: Annotated[AuthSettings, Depends(_auth_settings)],
		) -> TokenClaims \| None:
		if auth_settings.security_mode == SecurityMode.NONE:
		return None
		if user is None:
		raise HTTPException(
		status_code=status.HTTP_401_UNAUTHORIZED, detail="Not authenticated"
		)
		_require("scopes", required_set, user.scopes, user)
		return user

		return Depends(_check)