refactor: authentication flow and introduce development user mode

"""

Authentication & authorization dependencies.

Authentication for the portal's own login flow.

Organised into focused modules:

  * verification — JWT/JWKS decode

  * dependencies — request-time deps: bearer

  * guards       — role/scope authorization

  * routes       — login, callback, /me and logout

  * schemas      — what the browser is told about the user

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_id_token,

    _decode_token,

)

__all__ = [

    "BearerDep",

    "CurrentUserDep",

    "get_current_user",

    "require_any_role",

    "require_scopes",

    "_decode_id_token",

    "_decode_token",

    "_get_jwks_client",

from urllib.parse import urlencode

import structlog

from fastapi import APIRouter, Depends, HTTPException, Request, Response, status

from fastapi import APIRouter, Depends, Request, Response, status

from fastapi.responses import RedirectResponse

from starlette.concurrency import run_in_threadpool

from portal_service.api.auth.schemas import CurrentUser, LogoutResponse

from portal_service.api.auth.verification import _decode_id_token, _decode_token

from portal_service.api.auth.verification import _decode_id_token

from portal_service.api.dependencies import (

    CurrentSessionDep,

    CurrentUserDep,

    SettingsDep,

    get_identity_provider,

    get_login_service,

@auth_router.get("/me")

async def me(session: CurrentSessionDep) -> CurrentUser:

    """

    Reads the claims of the stored access token rather than asking Keycloak, so a

    page load costs no network call. The token was verified when it was issued and

    is verified again here, including its expiry.

    """

    try:

        claims = await run_in_threadpool(_decode_token, session.tokens.access_token)

    except AuthenticationError as exc:

        # Includes an expired access token; refreshing it comes next.

        logger.info("session token no longer usable", error=str(exc))

        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Not authenticated") from exc

    return CurrentUser.from_claims(claims)

async def me(user: CurrentUserDep) -> CurrentUser:

    return CurrentUser.from_claims(user)

@auth_router.post("/logout")

JWT verification: JWKS retrieval and token decoding.

Kept free of FastAPI/HTTP concerns — it raises domain errors only, so it can be

reused outside the request path. Callers (the dependencies module) translate

these into HTTP responses.

reused outside the request path. Callers (the auth routes) translate these into

HTTP responses.

"""

from hmac import compare_digest

from fastapi import Depends, HTTPException, Request, status

from sqlalchemy import text

from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession

from starlette.concurrency import run_in_threadpool

from portal_service.adapters.database.repos import (

    SqlLoginAttemptRepository,

    SqlLoginSessionRepository,

)

from portal_service.adapters.keycloak.oidc_client import KeycloakOIDCClient

from portal_service.api.auth.verification import _decode_token

from portal_service.app_state import AppState

from portal_service.application.login_service import LoginService

from portal_service.application.session_service import ActiveSession, SessionService

from portal_service.config import OIDCSettings, Settings, get_settings

from portal_service.domain.errors import IdentityProviderUnavailableError, InvalidSessionError

from portal_service.domain.errors import (

    AuthenticationError,

    IdentityProviderUnavailableError,

    InvalidSessionError,

)

from portal_service.domain.models import Role, TokenClaims

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

CurrentSessionDep = Annotated[ActiveSession, Depends(get_current_session)]

async def get_current_user(session: CurrentSessionDep) -> TokenClaims:

    """

    Who the request acts for, read from the session's access token rather than

    asking Keycloak, so it costs no network call. The token was verified when it was

    issued and is verified again here, including its expiry.

    """

    try:

        return await run_in_threadpool(_decode_token, session.tokens.access_token)

    except AuthenticationError as exc:

        logger.info("session token no longer usable", error=str(exc))

        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "Not authenticated") from exc

CurrentUserDep = Annotated[TokenClaims, Depends(get_current_user)]

def get_dev_user() -> TokenClaims:

    """

    Stands in for `get_current_user` when AUTH__SECURITY_MODE=none: every request

    acts as this user, with every role, and no login or Keycloak is needed.

    """

    return TokenClaims(

        sub="dev-user",

        iss="local",

        exp=0,

        iat=0,

        preferred_username="dev",

        realm_access={"roles": [role.value for role in Role]},

    )

class SecurityMode(StrEnum):

    # Users log in through Keycloak.

    JWT = "jwt"

    # Local development only: no login, every request acts as a dev user with every

    # role, and Keycloak need not be running.

    NONE = "none"