🔐 OAuth 2.0, OpenID Connect & JWT Reference¶

from neuroglia.dependency_injection import ServiceCollection
from neuroglia.mvc import ControllerBase
from fastapi import HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from typing import Dict, List

class JWTAuthService:
    def __init__(self,
                 issuer: str = "https://keycloak.mariospizzeria.com/auth/realms/pizzeria",
                 audience: str = "pizzeria-api",
                 jwks_url: str = "https://keycloak.mariospizzeria.com/auth/realms/pizzeria/protocol/openid_connect/certs"):
        self.issuer = issuer
        self.audience = audience
        self.jwks_url = jwks_url
        self._public_keys = {}

    async def validate_token(self, token: str) -> Dict:
        """Validate JWT token and return claims"""
        try:
            # Decode without verification first to get kid
            unverified_header = jwt.get_unverified_header(token)
            kid = unverified_header.get('kid')

            # Get public key for signature verification
            public_key = await self._get_public_key(kid)

            # Verify and decode token
            payload = jwt.decode(
                token,
                public_key,
                algorithms=['RS256'],
                issuer=self.issuer,
                audience=self.audience,
                options={"verify_exp": True}
            )

            return payload

        except jwt.ExpiredSignatureError:
            raise HTTPException(status_code=401, detail="Token has expired")
        except jwt.InvalidTokenError as e:
            raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")

    def check_scope(self, required_scope: str, token_scopes: str) -> bool:
        """Check if required scope is present in token scopes"""
        scopes = token_scopes.split(' ') if token_scopes else []
        return required_scope in scopes

    async def _get_public_key(self, kid: str):
        """Fetch and cache public keys from JWKS endpoint"""
        # Implementation would fetch from Keycloak JWKS endpoint
        # and cache the public keys for signature verification
        pass

from functools import wraps
from fastapi import HTTPException

def require_scope(required_scope: str):
    """Decorator to require specific OAuth scope"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Get current user token from dependency injection
            auth_service = kwargs.get('auth_service')  # Injected
            token_data = kwargs.get('current_user')    # From JWT validation

            if not auth_service.check_scope(required_scope, token_data.get('scope', '')):
                raise HTTPException(
                    status_code=403,
                    detail=f"Insufficient permissions. Required scope: {required_scope}"
                )

            return await func(*args, **kwargs)
        return wrapper
    return decorator

from neuroglia.mvc import ControllerBase
from classy_fastapi.decorators import get, post, put, delete
from fastapi import Depends

class OrdersController(ControllerBase):
    def __init__(self,
                 service_provider: ServiceProviderBase,
                 mapper: Mapper,
                 mediator: Mediator,
                 auth_service: JWTAuthService):
        super().__init__(service_provider, mapper, mediator)
        self.auth_service = auth_service

    @get("/", response_model=List[OrderDto])
    @require_scope("orders:read")
    async def get_orders(self,
                        current_user: dict = Depends(get_current_user)) -> List[OrderDto]:
        """Get orders - requires orders:read scope"""
        # Customers see only their orders, staff see all
        if "customer" in current_user.get("roles", []):
            query = GetOrdersByCustomerQuery(customer_id=current_user["sub"])
        else:
            query = GetAllOrdersQuery()

        result = await self.mediator.execute_async(query)
        return self.process(result)

    @post("/", response_model=OrderDto, status_code=201)
    @require_scope("orders:write")
    async def create_order(self,
                          create_order_dto: CreateOrderDto,
                          current_user: dict = Depends(get_current_user)) -> OrderDto:
        """Create new order - requires orders:write scope"""
        command = self.mapper.map(create_order_dto, PlaceOrderCommand)
        command.customer_id = current_user["sub"]  # From JWT

        result = await self.mediator.execute_async(command)
        return self.process(result)

class KitchenController(ControllerBase):

    @get("/status", response_model=KitchenStatusDto)
    @require_scope("kitchen:read")
    async def get_kitchen_status(self,
                                current_user: dict = Depends(get_current_user)) -> KitchenStatusDto:
        """Get kitchen status - requires kitchen:read scope"""
        query = GetKitchenStatusQuery()
        result = await self.mediator.execute_async(query)
        return self.process(result)

    @post("/orders/{order_id}/start")
    @require_scope("kitchen:manage")
    async def start_cooking_order(self,
                                 order_id: str,
                                 current_user: dict = Depends(get_current_user)) -> OrderDto:
        """Start cooking order - requires kitchen:manage scope"""
        command = StartCookingCommand(
            order_id=order_id,
            kitchen_staff_id=current_user["sub"]
        )
        result = await self.mediator.execute_async(command)
        return self.process(result)

from fastapi import Depends
from fastapi.security import HTTPBearer

security = HTTPBearer()

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    auth_service: JWTAuthService = Depends()
) -> dict:
    """Extract and validate user from JWT token"""
    token = credentials.credentials
    user_data = await auth_service.validate_token(token)
    return user_data

def configure_auth_services(services: ServiceCollection):
    """Configure authentication services"""
    services.add_singleton(JWTAuthService)
    services.add_scoped(lambda sp: get_current_user)

ROLE_SCOPES = {
    "customer": [
        "orders:read",      # View own orders
        "orders:write",     # Place new orders
        "menu:read"         # Browse menu
    ],
    "kitchen_staff": [
        "orders:read",      # View all orders
        "kitchen:read",     # View kitchen status
        "kitchen:manage",   # Manage cooking queue
        "menu:read"         # View menu
    ],
    "manager": [
        "orders:read",      # View all orders
        "orders:write",     # Create orders for customers
        "kitchen:read",     # Monitor kitchen
        "kitchen:manage",   # Manage kitchen operations
        "menu:read",        # View menu
        "menu:write",       # Update menu items
        "reports:read"      # View analytics
    ],
    "admin": [
        "admin"             # Full access to everything
    ]
}

import pytest
from unittest.mock import Mock, patch

class TestAuthenticatedEndpoints:
    def setup_method(self):
        self.auth_service = Mock(spec=JWTAuthService)
        self.test_user = {
            "sub": "customer_123",
            "name": "Mario Rossi",
            "email": "mario@example.com",
            "roles": ["customer"],
            "scope": "orders:read orders:write menu:read"
        }

    async def test_get_orders_with_valid_token(self, test_client):
        """Test getting orders with valid customer token"""
        self.auth_service.validate_token.return_value = self.test_user
        self.auth_service.check_scope.return_value = True

        headers = {"Authorization": "Bearer valid_token"}
        response = await test_client.get("/orders", headers=headers)

        assert response.status_code == 200
        # Should only return customer's own orders

    async def test_get_orders_insufficient_scope(self, test_client):
        """Test getting orders without required scope"""
        user_without_scope = {**self.test_user, "scope": "menu:read"}
        self.auth_service.validate_token.return_value = user_without_scope
        self.auth_service.check_scope.return_value = False

        headers = {"Authorization": "Bearer limited_token"}
        response = await test_client.get("/orders", headers=headers)

        assert response.status_code == 403
        assert "Insufficient permissions" in response.json()["detail"]

    async def test_kitchen_access_staff_only(self, test_client):
        """Test kitchen endpoints require staff role"""
        staff_user = {
            "sub": "staff_456",
            "roles": ["kitchen_staff"],
            "scope": "kitchen:read kitchen:manage"
        }
        self.auth_service.validate_token.return_value = staff_user
        self.auth_service.check_scope.return_value = True

        headers = {"Authorization": "Bearer staff_token"}
        response = await test_client.get("/kitchen/status", headers=headers)

        assert response.status_code == 200

    async def test_expired_token(self, test_client):
        """Test expired token handling"""
        from jwt import ExpiredSignatureError
        self.auth_service.validate_token.side_effect = ExpiredSignatureError()

        headers = {"Authorization": "Bearer expired_token"}
        response = await test_client.get("/orders", headers=headers)

        assert response.status_code == 401
        assert "expired" in response.json()["detail"].lower()