Commit cefbe873 authored by George Papathanail's avatar George Papathanail
Browse files

feat: add a watch-only enforcement flag for token validation and role/consent checks 

keycloak_settings.enforcement_enabled (default True) lets get_caller_context,
require_role, and require_three_legged log what they would have rejected
and let the request through, instead of the only choice being every
parent 6bc11465

src/open_exposure_gateway/core/authorization.py

+17 −0
Original line number Diff line number Diff line 
from dataclasses import dataclass

from typing import Annotated



import structlog

from fastapi import Depends, Request



from open_exposure_gateway.core.auth.tokens import TokenFlow

from open_exposure_gateway.core.config import Settings, get_settings

from open_exposure_gateway.core.exceptions import ForbiddenException

from open_exposure_gateway.dependencies import CallerContext, get_caller_context



logger = structlog.get_logger(__name__)





@dataclass(frozen=True)

class OperationRequirements:
@@ -96,9 +100,18 @@ def _requirements_for(request: Request) -> OperationRequirements: 
def require_role(

    request: Request,

    caller: Annotated[CallerContext, Depends(get_caller_context)],

    settings: Annotated[Settings, Depends(get_settings)],

) -> CallerContext:

    requirements = _requirements_for(request)

    if requirements.role not in caller.roles:

        if not settings.keycloak_settings.enforcement_enabled:

            logger.warning(

                "authz_would_reject",

                reason="missing_role",

                required_role=requirements.role,

                path=request.url.path,

            )

            return caller

        raise ForbiddenException(

            message=f"Caller is missing the required role: {requirements.role}"

        )
@@ -108,12 +121,16 @@ def require_role( 
def require_three_legged(

    request: Request,

    caller: Annotated[CallerContext, Depends(get_caller_context)],

    settings: Annotated[Settings, Depends(get_settings)],

) -> CallerContext:

    requirements = _requirements_for(request)

    # Independent of require_role: a legitimate app-provider service-account token

    # (TWO_LEGGED) must still be rejected here, since role correctness says nothing

    # about whether a real end-user actually consented.

    if requirements.requires_three_legged and caller.flow != TokenFlow.THREE_LEGGED:

        if not settings.keycloak_settings.enforcement_enabled:

            logger.warning("authz_would_reject", reason="not_three_legged", path=request.url.path)

            return caller

        raise ForbiddenException(

            message="This operation requires a three-legged (user-consented) access token"

        )

src/open_exposure_gateway/core/config.py

+7 −0
Original line number Diff line number Diff line
@@ -56,6 +56,13 @@ class KeycloakSettings(BaseModel): 
    issuer: str = "https://keycloak.invalid/realms/CHANGE_ME"

    jwks_url: str = "https://keycloak.invalid/realms/CHANGE_ME/protocol/openid-connect/certs"

    audience: str = "oeg"

    # Secure by default: missing/invalid tokens and role/flow mismatches are

    # rejected. Set to False only for a supervised rollout window -- rejections

    # are logged instead of enforced, so nothing blocks while callers still

    # without real tokens (e.g. an e2e harness) are migrated. Not meant to be

    # left False permanently; there is no separate switch to remove it later,

    # flipping this back to True *is* "removing the bypass".

    enforcement_enabled: bool = True





class CallbackSettings(BaseModel):

src/open_exposure_gateway/dependencies.py

+42 −7
Original line number Diff line number Diff line
@@ -2,6 +2,7 @@ from collections.abc import AsyncGenerator 
from dataclasses import dataclass

from typing import Annotated, Optional, cast



import structlog

from fastapi import Depends, Request

from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer

from sqlalchemy import text
@@ -40,7 +41,7 @@ from open_exposure_gateway.application.services.quality_on_demand_service import 
    QualityOnDemandService,

)

from open_exposure_gateway.core.auth.tokens import TokenFlow

from open_exposure_gateway.core.config import get_settings

from open_exposure_gateway.core.config import Settings, get_settings

from open_exposure_gateway.core.exceptions import UnauthorizedException

from open_exposure_gateway.core.state import AppState

from open_exposure_gateway.ports.database.callbacks import (
@@ -89,28 +90,62 @@ def get_token_verifier(request: Request) -> TokenVerifierPort: 


_bearer_scheme = HTTPBearer(auto_error=False)



logger = structlog.get_logger(__name__)





def _unverified_caller_context(x_correlator: Optional[str]) -> CallerContext:

    """Placeholder used only while `keycloak_settings.enforcement_enabled` is

    False. Its values are never checked for real: require_role/require_three_legged

    (core/authorization.py) each bypass their own check under the same flag, so this

    exists purely to give downstream code a CallerContext to work with during a

    watch-only rollout window, not to grant any implicit access."""

    return CallerContext(

        x_correlator=x_correlator,

        tenant_id="unverified",

        app_provider_id="unverified",

        roles=[],

        flow=TokenFlow.TWO_LEGGED,

        subject_id=None,

    )





async def get_caller_context(

    request: Request,

    x_correlator: XCorrelatorHeader = None,

    credentials: Optional[HTTPAuthorizationCredentials] = Depends(_bearer_scheme),

    token_verifier: TokenVerifierPort = Depends(get_token_verifier),

    settings: Settings = Depends(get_settings),

) -> CallerContext:

    enforcement_enabled = settings.keycloak_settings.enforcement_enabled



    if credentials is None:

        if enforcement_enabled:

            raise UnauthorizedException(message="Missing bearer token")

        logger.warning("auth_would_reject", reason="missing_bearer_token", path=request.url.path)

        return _unverified_caller_context(x_correlator_header(request))



    try:

        decoded = await token_verifier.verify(credentials.credentials)

    claims = decoded.claims

    except UnauthorizedException:

        if enforcement_enabled:

            raise

        logger.warning("auth_would_reject", reason="invalid_token", path=request.url.path)

        return _unverified_caller_context(x_correlator_header(request))



    claims = decoded.claims

    tenant_id = claims.get("organization")

    if not tenant_id:

    roles = claims.get("realm_access", {}).get("roles")

    if not tenant_id or not roles:

        if enforcement_enabled:

            raise UnauthorizedException(message="Access token missing required claims")

        logger.warning("auth_would_reject", reason="missing_claims", path=request.url.path)

        return _unverified_caller_context(x_correlator_header(request))



    return CallerContext(

        x_correlator=x_correlator_header(request),

        tenant_id=tenant_id,

        app_provider_id=claims["sub"],

        roles=claims["realm_access"]["roles"],

        roles=roles,

        flow=decoded.flow,

        subject_id=claims["sub"] if decoded.flow == TokenFlow.THREE_LEGGED else None,

    )