feat(stac): policy enforcement point proof of concept

common/auth/veda_auth/keycloak_client.py

@@ -7,14 +7,50 @@
 import base64
 import json
 import logging
-from typing import Any, Dict, List, Optional, Tuple
-from urllib.parse import urlencode
+from typing import Any, Dict, List, Optional, Tuple, Union
+from urllib.parse import urlencode, urlparse
 
 import httpx
 
 logger = logging.getLogger(__name__)
 
 
+class TokenError(Exception):
+    """Raised when the access token is expired, revoked, or invalid.
+    Keycloak returns HTTP 401
+    """
+
+    def __init__(self, detail: str = "Access token is expired or invalid"):
+        """To use when there is a token error for RPT call"""
+        self.detail = detail
+        super().__init__(detail)
+
+
+def parse_keycloak_from_openid_url(
+    openid_configuration_url: Union[str, Any]
+) -> Tuple[str, str]:
+    """Extract Keycloak base URL and realm from an OpenID discovery URL such as https://<host>/realms/<realm>/.well-known/openid-configuration"""
+    url_str = str(openid_configuration_url).strip() if openid_configuration_url else ""
+    if not url_str:
+        raise ValueError("Missing or empty OpenID configuration URL")
+
+    parsed = urlparse(url_str)
+    path = (parsed.path or "").rstrip("/")
+
+    if "/realms/" not in path:
+        raise ValueError(
+            "OpenID configuration URL must contain /realms/<realm>/ "
+            "(e.g. .../realms/my-realm/.well-known/openid-configuration)"
+        )
+
+    realm = path.split("/realms/")[-1].split("/")[0]
+    if not realm:
+        raise ValueError("Could not extract realm from OpenID configuration URL")
+
+    keycloak_url = f"{parsed.scheme}://{parsed.netloc}"
+    return keycloak_url, realm
+
+
 def _add_base64_padding(payload: str) -> str:
     """Add padding to base64 string if needed
 
@@ -195,20 +231,21 @@ def check_permission(
 
             # https://www.keycloak.org/docs/latest/authorization_services/#_service_rpt_overview
             for permission in permissions:
-                # Check rsid (RPT token format), resource_id (introspection format), or rsname (resource name)
-                resource_identifier = (
-                    permission.get("rsid")
-                    or permission.get("resource_id")
-                    or permission.get("rsname")
-                )
-                if resource_identifier == resource_id:
+                # rsname is the user defined resource name ("stac:collection:tenant:*") so use it instead
+                rsname = permission.get("rsname") or permission.get("resource_id")
+                if rsname == resource_id:
                     scopes = permission.get("scopes", [])
                     if scope in scopes:
                         return True
 
             return False
         except httpx.HTTPStatusError as e:
-            if e.response.status_code in (401, 403):
+            if e.response.status_code == 401:
+                logger.warning("Token rejected (401): %s", e.response.text)
+                raise TokenError(
+                    "Access token is expired or invalid. Please re-authenticate."
+                ) from e
+            if e.response.status_code == 403:
                 return False
             logger.error(
                 f"Permission check failed: {e.response.status_code} {e.response.text}"

common/auth/veda_auth/pep_middleware.py

@@ -0,0 +1,183 @@
+"""Policy Enforcement Point (PEP) middleware"""
+
+import logging
+import re
+from dataclasses import dataclass
+from typing import Awaitable, Callable, Optional, Sequence
+
+from veda_auth.keycloak_client import KeycloakPDPClient, TokenError
+from veda_auth.resource_extractors import COLLECTIONS_CREATE_PATH_RE
+
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.types import ASGIApp
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ProtectedRoute:
+    """Routes protected by policy decision point
+
+    path_re: regex pattern applied to request path
+    method: HTTP method
+    scope: Keycloak resource scope name to check (e.g. "create", "update", "delete")
+    """
+
+    path_re: str
+    method: str
+    scope: str
+
+
+DEFAULT_PROTECTED_ROUTES: Sequence[ProtectedRoute] = (
+    ProtectedRoute(path_re=COLLECTIONS_CREATE_PATH_RE, method="POST", scope="create"),
+)
+
+
+class PEPMiddleware(BaseHTTPMiddleware):
+    """Middleware that enforces UMA authorization"""
+
+    def __init__(
+        self,
+        app: ASGIApp,
+        *,
+        pdp_client: Callable[[], KeycloakPDPClient],
+        resource_extractor: Callable[[Request], Awaitable[Optional[str]]],
+        protected_routes: Optional[Sequence[ProtectedRoute]] = None,
+    ):
+        """Configure PEP middleware with a PDP client, resource extractor, and protected routes."""
+        super().__init__(app)
+        self._get_pdp_client = pdp_client
+        self._extract_resource_id = resource_extractor
+        routes = (
+            protected_routes
+            if protected_routes is not None
+            else DEFAULT_PROTECTED_ROUTES
+        )
+        self._compiled = [
+            (re.compile(r.path_re), r.method.upper(), r.scope) for r in routes
+        ]
+
+    def _get_matching_scope_and_route(
+        self, request: Request
+    ) -> Optional[tuple[str, str]]:
+        """Return (scope, method) for the route that matches, otherwise return None"""
+        path = request.url.path.rstrip("/") or "/"
+        method = request.method.upper()
+        for pattern, route_method, scope in self._compiled:
+            if route_method == method and pattern.search(path):
+                return (scope, route_method)
+        return None
+
+    def _get_bearer_token(self, request: Request) -> Optional[str]:
+        """Extract the Bearer token from the Authorization header"""
+        auth = request.headers.get("Authorization")
+        if not auth or not auth.startswith("Bearer "):
+            return None
+        return auth[7:].strip()
+
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        """Check UMA authorization for protected routes, pass through otherwise."""
+        matched_request = self._get_matching_scope_and_route(request)
+        if matched_request is None:
+            logger.debug(
+                "PEP: no protected route match for %s %s... continuing",
+                request.method,
+                request.url.path,
+            )
+            return await call_next(request)
+
+        scope, _method = matched_request
+        logger.info(
+            "PEP: matched protected route %s %s and scope=%s",
+            _method,
+            request.url.path,
+            scope,
+        )
+
+        pdp_client = self._get_pdp_client()
+
+        token = self._get_bearer_token(request)
+        if not token:
+            logger.warning(
+                "PEP: missing Bearer token for %s %s", _method, request.url.path
+            )
+            return JSONResponse(
+                status_code=401,
+                content={
+                    "detail": "Missing or invalid Authorization header (Bearer token required)"
+                },
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+
+        resource_id = await self._extract_resource_id(request)
+        if not resource_id:
+            logger.warning("PEP: no resource ID for %s %s", _method, request.url.path)
+            return JSONResponse(
+                status_code=403,
+                content={"detail": "Could not determine resource for authorization"},
+            )
+
+        logger.info(
+            "PEP: checking permission resource_id=%s, scope=%s, path=%s",
+            resource_id,
+            scope,
+            request.url.path,
+        )
+
+        try:
+            authorized = pdp_client.check_permission(
+                access_token=token,
+                resource_id=resource_id,
+                scope=scope,
+            )
+        except TokenError as e:
+            logger.warning(
+                "PEP: token error for %s %s: %s", _method, request.url.path, e.detail
+            )
+            return JSONResponse(
+                status_code=401,
+                content={"detail": e.detail},
+                headers={"WWW-Authenticate": 'Bearer error="invalid_token"'},
+            )
+        except Exception as e:
+            logger.exception(
+                "PEP: Keycloak check failed for resource_id=%s scope=%s: %s",
+                resource_id,
+                scope,
+                e,
+            )
+            return JSONResponse(
+                status_code=502,
+                content={"detail": "Authorization service temporarily unavailable"},
+            )
+
+        logger.info(
+            "PEP: authorization result=%s for resource_id=%s, scope=%s, path=%s",
+            authorized,
+            resource_id,
+            scope,
+            request.url.path,
+        )
+
+        if not authorized:
+            logger.warning(
+                "PEP: denied %s %s  resource_id=%s, scope=%s",
+                _method,
+                request.url.path,
+                resource_id,
+                scope,
+            )
+            return JSONResponse(
+                status_code=403,
+                content={
+                    "detail": (
+                        f"You do not have permission to {scope} this resource "
+                        f"({resource_id}). Verify that your user belongs to "
+                        f"the required tenant and role needed."
+                    )
+                },
+            )
+
+        return await call_next(request)

common/auth/veda_auth/resource_extractors.py

@@ -22,13 +22,17 @@
 STAC_COLLECTION_TEMPLATE = "stac:collection:{}:*"
 STAC_ITEM_TEMPLATE = "stac:item:{}:*"
 
-_COLLECTIONS_CREATE_PATH_PATTERN = re.compile(r".*?/collections$")
-_COLLECTIONS_PATH_PATTERN = re.compile(r".*?/collections/([^/]+)$")
-_COLLECTIONS_ITEM_PATH_PATTERN = re.compile(r".*?/collections/([^/]+)/items/([^/]+)$")
-_COLLECTIONS_ITEMS_PATH_PATTERN = re.compile(r".*?/collections/([^/]+)/items$")
-_COLLECTIONS_BULK_ITEMS_PATH_PATTERN = re.compile(
-    r".*?/collections/([^/]+)/bulk_items$"
-)
+COLLECTIONS_CREATE_PATH_RE = r".*?/collections$"
+COLLECTIONS_PATH_RE = r".*?/collections/([^/]+)$"
+COLLECTIONS_ITEM_PATH_RE = r".*?/collections/([^/]+)/items/([^/]+)$"
+COLLECTIONS_ITEMS_PATH_RE = r".*?/collections/([^/]+)/items$"
+COLLECTIONS_BULK_ITEMS_PATH_RE = r".*?/collections/([^/]+)/bulk_items$"
+
+_COLLECTIONS_CREATE_PATH_PATTERN = re.compile(COLLECTIONS_CREATE_PATH_RE)
+_COLLECTIONS_PATH_PATTERN = re.compile(COLLECTIONS_PATH_RE)
+_COLLECTIONS_ITEM_PATH_PATTERN = re.compile(COLLECTIONS_ITEM_PATH_RE)
+_COLLECTIONS_ITEMS_PATH_PATTERN = re.compile(COLLECTIONS_ITEMS_PATH_RE)
+_COLLECTIONS_BULK_ITEMS_PATH_PATTERN = re.compile(COLLECTIONS_BULK_ITEMS_PATH_RE)
 
 
 def _stac_collection_resource_id(request: Request) -> str:

ingest_api/runtime/src/main.py

@@ -1,3 +1,5 @@
+import logging
+
 import src.dependencies as dependencies
 import src.schemas as schemas
 import src.services as services
@@ -8,13 +10,17 @@
 from src.doc import DESCRIPTION
 from src.monitoring import ObservabilityMiddleware, logger, metrics, tracer
 from src.utils import get_keycloak_client_credentials
-from veda_auth.keycloak_client import KeycloakPDPClient
+from veda_auth.keycloak_client import KeycloakPDPClient, parse_keycloak_from_openid_url
+from veda_auth.pep_middleware import PEPMiddleware
+from veda_auth.resource_extractors import extract_ingest_resource_id
 
 from fastapi import Depends, FastAPI, HTTPException, Security
 from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 from starlette.requests import Request
 
+pep_logger = logging.getLogger(__name__)
+
 app = FastAPI(
     title="VEDA Ingestion API",
     description=DESCRIPTION,
@@ -229,33 +235,11 @@ def _extract_access_token(request: Request) -> str:
 
 
 def _parse_keycloak_config() -> tuple[str, str]:
-    """Extract Keycloak URL and realm from OIDC configuration URL"""
-    oidc_url = (
-        str(auth_settings.openid_configuration_url)
-        if auth_settings.openid_configuration_url
-        else None
-    )
-    if not oidc_url:
-        raise HTTPException(
-            status_code=503,
-            detail="Missing OPENID_CONFIGURATION_URL",
-        )
-
-    if "/realms/" not in oidc_url:
-        raise HTTPException(
-            status_code=503,
-            detail="Invalid OpenID configuration URL format",
-        )
-
-    keycloak_url = oidc_url.split("/realms/")[0]
-    realm_parts = oidc_url.split("/realms/")
-    if len(realm_parts) < 2:
-        raise HTTPException(
-            status_code=503,
-            detail="Could not extract realm from OpenID configuration URL",
-        )
-    realm = realm_parts[1].split("/")[0]
-    return keycloak_url, realm
+    """Extract Keycloak URL and realm from OIDC configuration URL."""
+    try:
+        return parse_keycloak_from_openid_url(auth_settings.openid_configuration_url)
+    except ValueError as e:
+        raise HTTPException(status_code=503, detail=str(e)) from e
 
 
 def _get_keycloak_credentials() -> tuple[str, str]:
@@ -288,6 +272,39 @@ def _get_keycloak_credentials() -> tuple[str, str]:
         ) from e
 
 
+def _get_keycloak_pdp_client() -> KeycloakPDPClient:
+    """Build Keycloak PDP client for PEP middleware from UMA resource server credentials"""
+    keycloak_url, realm = _parse_keycloak_config()
+    client_id, client_secret = _get_keycloak_credentials()
+    return KeycloakPDPClient(
+        keycloak_url=keycloak_url,
+        realm=realm,
+        client_id=client_id,
+        client_secret=client_secret,
+    )
+
+
+if (
+    auth_settings.openid_configuration_url
+    and settings.keycloak_uma_resource_server_client_secret_name
+):
+    pep_logger.info(
+        "PEP middleware enabled for Ingest API, secret_name=%s",
+        settings.keycloak_uma_resource_server_client_secret_name,
+    )
+    app.add_middleware(
+        PEPMiddleware,
+        pdp_client=_get_keycloak_pdp_client,
+        resource_extractor=extract_ingest_resource_id,
+    )
+else:
+    pep_logger.info(
+        "PEP middleware disabled for Ingest API, openid_url=%s, secret_name=%s",
+        bool(auth_settings.openid_configuration_url),
+        bool(settings.keycloak_uma_resource_server_client_secret_name),
+    )
+
+
 @app.get(
     "/auth/tenants/writable", response_model=schemas.TenantAccessResponse, tags=["Auth"]
 )