feat: add /me endpoint to retrieve current user information and session validation

                            browser to Keycloak.

  GET /api/auth/callback  — finish it: check the callback belongs to that login,

                            trade the code for tokens, and open a session.

  GET /api/auth/me        — who the session cookie belongs to.

The browser never receives a token; it gets cookies that name server-side rows.

"""

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

from fastapi.responses import RedirectResponse

from portal_service.api.auth.verification import _decode_id_token

from portal_service.api.auth.schemas import CurrentUser

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

from portal_service.api.dependencies import (

    CurrentSessionDep,

    SettingsDep,

    get_identity_provider,

    get_login_service,

    # The login is over: the attempt cookie has nothing left to name.

    response.delete_cookie(settings.session.attempt_cookie_name, path="/api/auth")

    return response

@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 = _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)

"""What the browser is told about the logged-in user. Never tokens."""

from pydantic import BaseModel

from portal_service.domain.models import TokenClaims

class CurrentUser(BaseModel):

    subject: str

    username: str | None = None

    email: str | None = None

    roles: list[str] = []

    organization: str | None = None

    @classmethod

    def from_claims(cls, claims: TokenClaims) -> "CurrentUser":

        return cls(

            subject=claims.sub,

            username=claims.preferred_username,

            email=claims.email,

            # Sorted so the same session always answers the same way.

            roles=sorted(claims.roles),

            organization=claims.organization,

        )

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.application.session_service import SessionService

from portal_service.application.session_service import ActiveSession, SessionService

from portal_service.config import OIDCSettings, Settings, get_settings

from portal_service.domain.errors import InvalidSessionError

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

        sessions=sessions,

        session_ttl=timedelta(seconds=settings.session.max_age),

    )

async def get_current_session(

    request: Request,

    settings: SettingsDep,

    service: Annotated[SessionService, Depends(get_session_service)],

) -> ActiveSession:

    """The session the request's cookie names, or 401."""

    cookie = request.cookies.get(settings.session.cookie_name)

    if not cookie:

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

    try:

        return await service.load(cookie)

    except InvalidSessionError:

        # Same answer for every reason: malformed, unknown, expired or undecryptable.

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

CurrentSessionDep = Annotated[ActiveSession, Depends(get_current_session)]

from dataclasses import dataclass, field

from datetime import UTC, datetime, timedelta

from portal_service.application.session_crypto import SessionCredentials, encrypt_tokens

from portal_service.domain.models import IssuedTokens, LoginSession

from portal_service.application.session_crypto import (

    SessionCredentials,

    decrypt_tokens,

    encrypt_tokens,

)

from portal_service.domain.errors import InvalidSessionError

from portal_service.domain.models import IssuedTokens, LoginSession, SessionTokens

from portal_service.domain.ports.database.login_sessions import LoginSessionRepository

    expires_at: datetime

@dataclass(frozen=True)

class ActiveSession:

    """A session opened by a cookie: the stored row and the tokens it held."""

    session: LoginSession

    tokens: SessionTokens = field(repr=False)

    credentials: SessionCredentials = field(repr=False)

class SessionService:

    def __init__(self, sessions: LoginSessionRepository, session_ttl: timedelta) -> None:

        self._sessions = sessions

            )

        )

        return CreatedSession(cookie_value=credentials.to_cookie(), expires_at=expires_at)

    async def load(self, cookie_value: str) -> ActiveSession:

        """

        Opens the session a cookie names. Every failure is the same to the caller:

        the cookie is malformed, names nothing, has expired, or does not decrypt.

        All of them mean "not logged in".

        """

        credentials = SessionCredentials.from_cookie(cookie_value)

        row = await self._sessions.get_by_id(credentials.session_id_hash)

        if row is None:

            raise InvalidSessionError("No session for this cookie")

        if row.expires_at <= datetime.now(UTC):

            raise InvalidSessionError("Session has expired")

        return ActiveSession(

            session=row,

            tokens=decrypt_tokens(row.encrypted_tokens, credentials),

            credentials=credentials,

        )

    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