feat: added support for token-based auth and token exchange in openeo

Author: JanssenBrm

app/auth.py

@@ -1,3 +1,5 @@
+from typing import Any, Dict
+import httpx
 import jwt
 from fastapi import Depends, HTTPException, WebSocket, status
 from fastapi.security import OAuth2AuthorizationCodeBearer
@@ -49,7 +51,7 @@ def get_current_user_id(token: str = Depends(oauth2_scheme)):
 async def websocket_authenticate(websocket: WebSocket) -> str | None:
     """
     Authenticate a WebSocket connection using a JWT token from query params.
-    Returns the ID of the authenticated user payload if valid, otherwise closes the connection.
+    Returns the token of the authenticated user payload if valid, otherwise closes the connection.
     """
     logger.debug("Authenticating websocket")
     token = websocket.query_params.get("token")
@@ -59,10 +61,82 @@ async def websocket_authenticate(websocket: WebSocket) -> str | None:
         return None
 
     try:
-        user_id = get_current_user_id(token)
         await websocket.accept()
-        return user_id
+        return token
     except Exception as e:
         logger.error(f"Invalid token in websocket authentication: {e}")
         await websocket.close(code=1008, reason="Invalid token")
         return None
+
+
+async def exchange_token_for_provider(
+    initial_token: str, provider: str
+) -> Dict[str, Any]:
+    """
+    Exchange a Keycloak access token for a token/audience targeted at `provider`
+    using the Keycloak Token Exchange (grant_type=urn:ietf:params:oauth:grant-type:token-exchange).
+
+    :param initial_token: token obtained from the client (Bearer token)
+    :param provider: target provider name or client_id.
+
+    :return: The token response (dict) on success.
+
+    :raise: Raises HTTPException with an appropriate status and message on error.
+    """
+    token_url = f"{KEYCLOAK_BASE_URL}/protocol/openid-connect/token"
+
+    # Check if the necessary settings are in place
+    if not settings.keycloak_client_id or not settings.keycloak_client_secret:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="Token exchange not configured on the server (missing client credentials).",
+        )
+
+    payload = {
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "client_id": settings.keycloak_client_id,
+        "client_secret": settings.keycloak_client_secret,
+        "subject_token": initial_token,
+        "requested_issuer": provider,
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.post(token_url, data=payload)
+    except httpx.RequestError as exc:
+        logger.error(f"Token exchange network error for provider={provider}: {exc}")
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="Failed to contact the identity provider for token exchange.",
+        )
+
+    # Parse response
+    try:
+        body = resp.json()
+    except ValueError:
+        logger.error(
+            f"Token exchange invalid JSON response (status={resp.status_code})"
+        )
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail="Invalid response from identity provider during token exchange.",
+        )
+
+    if resp.status_code != 200:
+        # Keycloak returns error and error_description fields for token errors
+        err = body.get("error_description") or body.get("error") or resp.text
+        logger.error(
+            "Token exchange failed",
+            extra={"provider": provider, "status": resp.status_code, "error": err},
+        )
+        # Map common upstream statuses to meaningful client statuses
+        client_status = (
+            status.HTTP_401_UNAUTHORIZED
+            if resp.status_code in (400, 401, 403)
+            else status.HTTP_502_BAD_GATEWAY
+        )
+
+        raise HTTPException(client_status, detail=body)
+
+    # Successful exchange, return token response (access_token, expires_in, etc.)
+    return body

app/config/settings.py

@@ -35,5 +35,11 @@ class Settings(BaseSettings):
         extra="allow",
     )
 
+    # openEO
+    openeo_enable_user_credentials: bool = Field(
+        default=False,
+        json_schema_extra={"env": "OPENEO_ENABLE_USER_CREDENTIALS"},
+    )
+
 
 settings = Settings()

app/platforms/base.py

@@ -13,12 +13,18 @@ class BaseProcessingPlatform(ABC):
     """
 
     @abstractmethod
-    def execute_job(
-        self, title: str, details: ServiceDetails, parameters: dict, format: OutputFormatEnum
+    async def execute_job(
+        self,
+        user_token: str,
+        title: str,
+        details: ServiceDetails,
+        parameters: dict,
+        format: OutputFormatEnum,
     ) -> str:
         """
         Execute a processing job on the platform with the given service ID and parameters.
 
+        :param user_token: The access token of the user executing the job.
         :param title: The title of the job to be executed.
         :param details: The service details containing the service ID and application.
         :param parameters: The parameters required for the job execution.
@@ -28,23 +34,27 @@ def execute_job(
         pass
 
     @abstractmethod
-    def get_job_status(
-        self, job_id: str, details: ServiceDetails
+    async def get_job_status(
+        self, user_token: str, job_id: str, details: ServiceDetails
     ) -> ProcessingStatusEnum:
         """
         Retrieve the job status of a processing job that is running on the platform.
 
+        :param user_token: The access token of the user executing the job.
         :param job_id: The ID of the job on the platform
         :param details: The service details containing the service ID and application.
         :return: Return the processing status
         """
         pass
 
     @abstractmethod
-    def get_job_results(self, job_id: str, details: ServiceDetails) -> Collection:
+    async def get_job_results(
+        self, user_token: str, job_id: str, details: ServiceDetails
+    ) -> Collection:
         """
         Retrieve the job results of a processing job that is running on the platform.
 
+        :param user_token: The access token of the user executing the job.
         :param job_id: The ID of the job on the platform
         :param details: The service details containing the service ID and application.
         :return: STAC collection representing the results.

app/platforms/implementations/ogc_api_process.py

@@ -12,19 +12,26 @@ class OGCAPIProcessPlatform(BaseProcessingPlatform):
     This class handles the execution of processing jobs on the OGC API Process platform.
     """
 
-    def execute_job(
-        self, title: str, details: ServiceDetails, parameters: dict, format: OutputFormatEnum
+    async def execute_job(
+        self,
+        user_token: str,
+        title: str,
+        details: ServiceDetails,
+        parameters: dict,
+        format: OutputFormatEnum,
     ) -> str:
         raise NotImplementedError("OGC API Process job execution not implemented yet.")
 
-    def get_job_status(
-        self, job_id: str, details: ServiceDetails
+    async def get_job_status(
+        self, user_token: str, job_id: str, details: ServiceDetails
     ) -> ProcessingStatusEnum:
         raise NotImplementedError(
             "OGC API Process job status retrieval not implemented yet."
         )
 
-    def get_job_results(self, job_id: str, details: ServiceDetails) -> Collection:
+    async def get_job_results(
+        self, user_token: str, job_id: str, details: ServiceDetails
+    ) -> Collection:
         raise NotImplementedError(
             "OGC API Process job result retrieval not implemented yet."
         )