Merge pull request #23 from ESA-APEx/token_exchange

Token exchange

Author: JanssenBrm

app/config/openeo/settings.py

@@ -0,0 +1,14 @@
+from enum import Enum
+from typing import Optional
+from pydantic import BaseModel
+
+
+class OpenEOBackendConfig(BaseModel):
+    client_credentials: Optional[str] = None
+    token_provider: Optional[str] = None
+    token_prefix: Optional[str] = None
+
+
+class OpenEOAuthMethod(str, Enum):
+    CLIENT_CREDENTIALS = "CLIENT_CREDENTIALS"
+    USER_CREDENTIALS = "USER_CREDENTIALS"

app/config/settings.py

@@ -1,13 +1,18 @@
+import json
+from typing import Dict
+
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
+from app.config.openeo.settings import OpenEOAuthMethod, OpenEOBackendConfig
+
 
 class Settings(BaseSettings):
     app_name: str = Field(
-        default="APEx Disatpcher API", json_schema_extra={"env": "APP_NAME"}
+        default="APEx Dispach API", json_schema_extra={"env": "APP_NAME"}
     )
     app_description: str = Field(
-        default="API description for the APEx Dispatcher",
+        default="",
         json_schema_extra={"env": "APP_DESCRIPTION"},
     )
     env: str = Field(default="development", json_schema_extra={"env": "APP_ENV"})
@@ -16,6 +21,12 @@ class Settings(BaseSettings):
         default="", json_schema_extra={"env": "CORS_ALLOWED_ORIGINS"}
     )
 
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_file_encoding="utf-8",
+        extra="allow",
+    )
+
     # Keycloak / OIDC
     keycloak_host: str = Field(
         default=str("localhost"),
@@ -29,17 +40,47 @@ class Settings(BaseSettings):
         default="", json_schema_extra={"env": "KEYCLOAK_CLIENT_SECRET"}
     )
 
-    model_config = SettingsConfigDict(
-        env_file=".env",
-        env_file_encoding="utf-8",
-        extra="allow",
+    # openEO Settings
+    openeo_auth_method: OpenEOAuthMethod = Field(
+        default=OpenEOAuthMethod.USER_CREDENTIALS,
+        json_schema_extra={"env": "OPENEO_AUTH_METHOD"},
     )
 
-    # openEO
-    openeo_enable_user_credentials: bool = Field(
-        default=False,
-        json_schema_extra={"env": "OPENEO_ENABLE_USER_CREDENTIALS"},
+    openeo_backends: str | None = Field(
+        default="", json_schema_extra={"env": "OPENEO_BACKENDS"}
     )
 
+    openeo_backend_config: Dict[str, OpenEOBackendConfig] = Field(default_factory=dict)
+
+    def load_openeo_backends_from_env(self):
+        """
+        Populate self.backends from BACKENDS_JSON if provided, otherwise keep defaults.
+        BACKENDS_JSON should be a JSON object keyed by hostname with BackendConfig-like values.
+        """
+        required_fields = []
+        if self.openeo_backends:
+
+            if self.openeo_auth_method == OpenEOAuthMethod.CLIENT_CREDENTIALS:
+                required_fields = ["client_credentials"]
+            elif self.openeo_auth_method == OpenEOAuthMethod.USER_CREDENTIALS:
+                required_fields = ["token_provider"]
+
+            try:
+                raw = json.loads(self.openeo_backends)
+                for host, cfg in raw.items():
+                    backend = OpenEOBackendConfig(**cfg)
+
+                    for field in required_fields:
+                        if not getattr(backend, field, None):
+                            raise ValueError(
+                                f"Backend '{host}' must define '{field}' when "
+                                f"OPENEO_AUTH_METHOD={self.openeo_auth_method}"
+                            )
+                    self.openeo_backend_config[host] = OpenEOBackendConfig(**cfg)
+            except Exception:
+                # Fall back or raise as appropriate
+                raise
+
 
 settings = Settings()
+settings.load_openeo_backends_from_env()

app/platforms/implementations/openeo.py

@@ -1,38 +1,21 @@
 import datetime
-import os
-import re
-from typing import Any
-import urllib
-import jwt
 
-from loguru import logger
+import jwt
 import openeo
 import requests
 from dotenv import load_dotenv
+from loguru import logger
+from stac_pydantic import Collection
 
 from app.auth import exchange_token_for_provider
-from app.config.settings import settings
+from app.config.settings import OpenEOAuthMethod, settings
 from app.platforms.base import BaseProcessingPlatform
 from app.platforms.dispatcher import register_platform
-from app.schemas.enum import OutputFormatEnum, ProcessTypeEnum, ProcessingStatusEnum
+from app.schemas.enum import OutputFormatEnum, ProcessingStatusEnum, ProcessTypeEnum
 from app.schemas.unit_job import ServiceDetails
-from stac_pydantic import Collection
 
 load_dotenv()
 
-# Constants
-BACKEND_AUTH_ENV_MAP = {
-    "openeo.dataspace.copernicus.eu": "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED",
-    "openeofed.dataspace.copernicus.eu": "OPENEO_AUTH_CLIENT_CREDENTIALS_CDSEFED",
-    "openeo.vito.be": "OPENEO_AUTH_CLIENT_CREDENTIALS_OPENEO_VITO",
-}
-
-BACKEND_PROVIDER_ID_MAP = {
-    "openeo.dataspace.copernicus.eu": "esa-eoiam",
-    "openeofed.dataspace.copernicus.eu": "esa-eoiam",
-    "openeo.vito.be": "egi",
-}
-
 
 @register_platform(ProcessTypeEnum.OPENEO)
 class OpenEOPlatform(BaseProcessingPlatform):
@@ -72,23 +55,44 @@ def _connection_expired(self, connection: openeo.Connection) -> bool:
             logger.warning("No JWT bearer token found in connection.")
             return True
 
+    async def _get_bearer_token(self, user_token: str, url: str) -> str:
+        """
+        Retrieve the bearer token for the OpenEO backend. This is done  by exchanging the user's
+        token for a platform-specific token using the configured token provider.
+
+        :param url: The URL of the OpenEO backend.
+        :return: The bearer token as a string.
+        """
+
+        provider = settings.openeo_backend_config[url].token_provider
+        token_prefix = settings.openeo_backend_config[url].token_prefix
+
+        if not provider or not token_prefix:
+            raise ValueError(
+                f"Backend '{url}' must define 'token_provider' and 'token_prefix'"
+            )
+
+        platform_token = await exchange_token_for_provider(
+            initial_token=user_token, provider=provider
+        )
+        return f"{token_prefix}/{platform_token['access_token']}"
+
     async def _authenticate_user(
         self, user_token: str, url: str, connection: openeo.Connection
     ) -> openeo.Connection:
         """
         Authenticate the connection using the user's token.
         This method can be used to set the user's token for the connection.
         """
-        if settings.openeo_enable_user_credentials:
+
+        if url not in settings.openeo_backend_config:
+            raise ValueError(f"No OpenEO backend configuration found for URL: {url}")
+
+        if settings.openeo_auth_method == OpenEOAuthMethod.USER_CREDENTIALS:
             logger.debug("Using user credentials for OpenEO connection authentication")
-            provider = self._get_backend_config(BACKEND_PROVIDER_ID_MAP, url)
-            platform_token = await exchange_token_for_provider(
-                initial_token=user_token, provider=provider
-            )
-            connection.authenticate_bearer_token(
-                bearer_token=platform_token["access_token"]
-            )
-        else:
+            bearer_token = await self._get_bearer_token(user_token, url)
+            connection.authenticate_bearer_token(bearer_token=bearer_token)
+        elif settings.openeo_auth_method == OpenEOAuthMethod.CLIENT_CREDENTIALS:
             logger.debug(
                 "Using client credentials for OpenEO connection authentication"
             )
@@ -99,6 +103,10 @@ async def _authenticate_user(
                 client_id=client_id,
                 client_secret=client_secret,
             )
+        else:
+            raise ValueError(
+                f"Unsupported OpenEO authentication method: {settings.openeo_auth_method}"
+            )
 
         return connection
 
@@ -119,22 +127,6 @@ async def _setup_connection(self, user_token: str, url: str) -> openeo.Connectio
         self._connection_cache[url] = connection
         return connection
 
-    def _get_backend_config(self, map: Any, url: str) -> str:
-        """
-        Get the authentication provider for the OpenEO backend.
-
-        :param url: The URL of the OpenEO backend.
-        :return: The name of the authentication provider.
-        """
-        if not re.match(r"https?://", url):
-            url = f"https://{url}"
-
-        hostname = urllib.parse.urlparse(url).hostname
-        if not hostname or hostname not in map:
-            raise ValueError(f"Unsupported backend: {url} (hostname={hostname})")
-
-        return map[hostname]
-
     def _get_client_credentials(self, url: str) -> tuple[str, str, str]:
         """
         Get client credentials for the OpenEO backend.
@@ -143,16 +135,17 @@ def _get_client_credentials(self, url: str) -> tuple[str, str, str]:
         :param url: The URL of the OpenEO backend.
         :return: A tuple containing provider ID, client ID, and client secret.
         """
-        env_var = self._get_backend_config(BACKEND_AUTH_ENV_MAP, url)
-        credentials_str = os.getenv(env_var)
+        credentials_str = settings.openeo_backend_config[url].client_credentials
 
         if not credentials_str:
-            raise ValueError(f"Environment variable {env_var} not set.")
+            raise ValueError(
+                f"Client credentials not configured for OpenEO backend at {url}"
+            )
 
         parts = credentials_str.split("/", 2)
         if len(parts) != 3:
             raise ValueError(
-                f"Invalid client credentials format in {env_var},"
+                f"Invalid client credentials format for {url},"
                 "expected 'provider_id/client_id/client_secret'."
             )
         provider_id, client_id, client_secret = parts

app/services/processing.py

@@ -39,26 +39,48 @@ async def create_processing_job(
     user = get_current_user_id(token)
     logger.info(f"Creating processing job for {user} with summary: {request}")
 
-    platform = get_processing_platform(request.label)
+    try:
+        platform = get_processing_platform(request.label)
+
+        job_id = await platform.execute_job(
+            user_token=token,
+            title=request.title,
+            details=request.service,
+            parameters=request.parameters,
+            format=request.format,
+        )
 
-    job_id = await platform.execute_job(
-        user_token=token,
-        title=request.title,
-        details=request.service,
-        parameters=request.parameters,
-        format=request.format,
-    )
+        record = ProcessingJobRecord(
+            title=request.title,
+            label=request.label,
+            status=(
+                ProcessingStatusEnum.CREATED if job_id else ProcessingStatusEnum.FAILED
+            ),
+            user_id=user,
+            platform_job_id=job_id,
+            parameters=json.dumps(request.parameters),
+            service=request.service.model_dump_json(),
+            upscaling_task_id=upscaling_task_id,
+        )
+
+    except Exception as e:
+        logger.error(f"Error creating processing job: {e}")
+
+        if upscaling_task_id:
+            # Do create the record in case of upscaling task to mark it as failed
+            record = ProcessingJobRecord(
+                title=request.title,
+                label=request.label,
+                status=ProcessingStatusEnum.FAILED,
+                user_id=user,
+                platform_job_id=None,
+                parameters=json.dumps(request.parameters),
+                service=request.service.model_dump_json(),
+                upscaling_task_id=upscaling_task_id,
+            )
+        else:
+            raise e
 
-    record = ProcessingJobRecord(
-        title=request.title,
-        label=request.label,
-        status=ProcessingStatusEnum.CREATED if job_id else ProcessingStatusEnum.FAILED,
-        user_id=user,
-        platform_job_id=job_id,
-        parameters=json.dumps(request.parameters),
-        service=request.service.model_dump_json(),
-        upscaling_task_id=upscaling_task_id,
-    )
     record = save_job_to_db(database, record)
     return ProcessingJobSummary(
         id=record.id,

docs/environment.md

@@ -0,0 +1,68 @@
+# Configuring the Dispatcher
+The Dispatcher can be configured using environment variables. These variables can be set directly in your shell or defined in a `.env` file for convenience.
+Below are the key settings that can be adjusted to tailor the Dispatcher's behavior to your needs. 
+
+| Environment Variable     | Description                                                 | Values                                    | Default Value      |
+| ------------------------ | ----------------------------------------------------------- | ----------------------------------------- | ------------------ |
+| **General Settings**     |                                                             |                                           |                    |
+| `APP_NAME`               | The name of the application.                                | Text                                      | APEx Dispatch API  |
+| `APP_DESCRIPTION`        | A brief description of the application.                     | Text                                      | ""                 |
+| `APP_ENV`                | The environment in which the application is running         | `development` /  `production`                    | development        |
+| `CORS_ALLOWED_ORIGINS`   | Comma-separated list of allowed origins for CORS.           | Text                                      | ""                 |
+| **Database Settings**|||
+| `DATABASE_URL`           | The database connection URL.                                | Text                                      | "" |
+| **Keycloak Settings**    |                                                             |                                           |                    |
+| `KEYCLOAK_HOST`          | The hostname of the Keycloak server.                        | Text                                      | localhost          |
+| `KEYCLOAK_REALM`         | The Keycloak realm to use for authentication.               | Text                                      | ""                 |
+| `KEYCLOAK_CLIENT_ID`     | The client ID registered in Keycloak.                       | Text                                      | ""                 |
+| `KEYCLOAK_CLIENT_SECRET` | The client secret for the Keycloak client.                  | Text                                      | ""                 |
+| **openEO Settings**      |                                                             |                                           |                    |
+| `OPENEO_AUTH_METHOD`     | The authentication method to use for openEO backends.       | `USER_CREDENTIALS` / `CLIENT_CREDENTIALS` | `USER_CREDENTIALS` |
+| `OPENEO_BACKEND_CONFIG`  | JSON string defining the configuration for openEO backends. | JSON                                      | `{}`               |
+
+
+## openEO Backend Configuration
+The `OPENEO_BACKEND_CONFIG` environment variable allows you to specify the configuration for multiple openEO backends in JSON format.
+Here is an example of how to structure this configuration:  
+
+```json
+{
+  "https://openeo.backend1.com": {
+    "client_credentials": "oidc_provider/client_id/secret_secret",
+    "token_provider": "backend",
+    "token_prefix": "oidc/backend"
+  },
+  ...
+}
+```
+Each backend configuration can include the following fields:
+
+- `client_credentials`: The client credentials for authenticating with the openEO backend. This is required if the `OPENEO_AUTH_METHOD` is set to `CLIENT_CREDENTIALS`. It is a single string in the format `oidc_provider/client_id/client_secret` that should be split into its components when used.
+- `token_provider`: The provider refers to the OIDC IDP alias that needs to be used to exchange the incoming token to an external token. This is required if the `OPENEO_AUTH_METHOD` is set to `USER_CREDENTIALS`. For example, if you have a Keycloak setup with an IDP alias `openeo-idp`, you would set this field to `openeo-idp`. This means that when a user authenticates with their token, the Dispatcher will use the `openeo-idp` to exchange the user's token for a token that is valid for the openEO backend.
+- `token_prefix`: An optional prefix to be added to the token when authenticating (e.g., "CDSE"). The prefix is required by some backends to identify the token type. This will be prepended to the exchanged token when authenticating with the openEO backend.
+
+## Example Configuration
+Here is an example of setting the environment variables in a `.env` file:
+
+```env
+# General Settings
+APP_NAME="APEx Dispatch API"
+APP_DESCRIPTION="APEx Dispatch Service API to run jobs and upscaling tasks"
+APP_ENV=development
+
+CORS_ALLOWED_ORIGINS=http://localhost:5173
+
+# Database Settings
+DATABASE_URL=sqlite:///:memory:
+
+# Keycloak Settings
+KEYCLOAK_HOST=localhost
+KEYCLOAK_REALM=apex
+KEYCLOAK_CLIENT_ID=apex-client-id
+KEYCLOAK_CLIENT_SECRET=apex-client-secret
+
+
+# openEO Settings
+OPENEO_AUTH_METHOD=USER_CREDENTIALS
+OPENEO_BACKENDS='{"https://openeo.backend1.com" {"client_credentials": "oidc_provider/client_id/secret_secret", "token_provider": "backend", "token_prefix": "oidc/backend"}}'
+```