Alternative token exchange

alembic/versions/833e4a41c2ad_generation_of_tables.py

@@ -27,7 +27,7 @@ def upgrade() -> None:
     sa.Column('label', sa.Enum('OPENEO', 'OGC_API_PROCESS', name='processtypeenum'), nullable=False),
     sa.Column('status', sa.Enum('CREATED', 'QUEUED', 'RUNNING', 'FINISHED', 'CANCELED', 'FAILED', 'UNKNOWN', name='processingstatusenum'), nullable=False),
     sa.Column('user_id', sa.String(length=255), nullable=False),
-    sa.Column('service', mysql.LONGTEXT(), nullable=False),
+    sa.Column('service', sa.Text, nullable=False),
     sa.Column('created', sa.DateTime(), nullable=False),
     sa.Column('updated', sa.DateTime(), nullable=False),
     sa.PrimaryKeyConstraint('id')
@@ -46,8 +46,8 @@ def upgrade() -> None:
     sa.Column('status', sa.Enum('CREATED', 'QUEUED', 'RUNNING', 'FINISHED', 'CANCELED', 'FAILED', 'UNKNOWN', name='processingstatusenum'), nullable=False),
     sa.Column('user_id', sa.String(length=255), nullable=False),
     sa.Column('platform_job_id', sa.String(length=255), nullable=True),
-    sa.Column('parameters', mysql.LONGTEXT(), nullable=False),
-    sa.Column('service', mysql.LONGTEXT(), nullable=False),
+    sa.Column('parameters', sa.Text, nullable=False),
+    sa.Column('service', sa.Text, nullable=False),
     sa.Column('created', sa.DateTime(), nullable=False),
     sa.Column('updated', sa.DateTime(), nullable=False),
     sa.Column('upscaling_task_id', sa.Integer(), nullable=True),

app/auth.py

@@ -11,6 +11,7 @@
 
 from app.error import AuthException, DispatcherException
 from app.schemas.websockets import WSStatusMessage
+from app.config.schemas import BackendAuthConfig, TokenExchangeMethod
 
 from .config.settings import settings
 
@@ -109,16 +110,95 @@ async def exchange_token(user_token: str, url: str) -> str:
     :return: The bearer token as a string, prefixed with `token_prefix` when configured.
     """
 
-    provider = settings.backend_auth_config[url].token_provider
-    token_prefix = settings.backend_auth_config[url].token_prefix
+    backend_idp = settings.backend_auth_config[url]
 
-    if not provider:
-        raise ValueError(
-            f"Backend '{url}' must define 'token_provider'"
+    if backend_idp.exchange_method == TokenExchangeMethod.IDENTITY_BROKERING:
+        if not backend_idp.token_provider:
+            raise ValueError(f"Backend '{url}' must define 'token_provider'")
+        external_token = await _retrieve_broker_token(
+            user_token=user_token, provider=backend_idp.token_provider
         )
+    elif backend_idp.exchange_method == TokenExchangeMethod.TRUSTED_TOKEN_DELEGATION:
+        external_token = await _exchange_token_for_provider(
+            initial_token=user_token, backend_idp=backend_idp
+        )
+    else:
+        raise ValueError(f"Unsupported token exchange method: {backend_idp.exchange_method}")
+
+    return (
+        f"{backend_idp.token_prefix}/{external_token}"
+        if backend_idp.token_prefix
+        else external_token
+    )
+
+
+async def _exchange_token_for_provider(
+    initial_token: str, backend_idp: BackendAuthConfig
+) -> str:
+    """Exchange an APEx access token using trusted token delegation."""
+    if not backend_idp.token_url:
+        raise AuthException(
+            http_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            message="Trusted Token Delegation not configured (missing token URL).",
+        )
+    if not backend_idp.client_id:
+        raise AuthException(
+            http_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            message="Trusted Token Delegation not configured (missing client ID).",
+        )
+
+    provider = backend_idp.token_provider or backend_idp.token_url
+    payload = {
+        "client_id": backend_idp.client_id,
+        "client_secret": backend_idp.client_secret,
+        "grant_type": "urn:ietf:params:oauth:grant-type:token-exchange",
+        "subject_token": initial_token,
+        "subject_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "subject_issuer": backend_idp.subject_issuer,
+        "audience": backend_idp.audience,
+        "requested_token_type": "urn:ietf:params:oauth:token-type:access_token",
+        "scope": "openid profile email",
+    }
 
-    external_token = await _retrieve_broker_token(user_token=user_token, provider=provider)
-    return f"{token_prefix}/{external_token}" if token_prefix else external_token
+    try:
+        async with httpx.AsyncClient(timeout=BROKER_TOKEN_TIMEOUT_SECONDS) as client:
+            resp = await client.post(backend_idp.token_url, data=payload)
+    except httpx.TimeoutException:
+        logger.error(f"Trusted token delegation timed out for provider={provider}")
+        raise AuthException(
+            http_status=status.HTTP_504_GATEWAY_TIMEOUT,
+            message=_support_message(provider),
+        )
+    except httpx.RequestError as exc:
+        logger.error(
+            f"Trusted token delegation network error for provider={provider}: "
+            f"{type(exc).__name__}"
+        )
+        raise AuthException(
+            http_status=status.HTTP_502_BAD_GATEWAY,
+            message=_support_message(provider),
+        )
+
+    try:
+        body = resp.json()
+    except ValueError:
+        body = None
+
+    access_token = body.get("access_token") if isinstance(body, dict) else None
+    if resp.status_code != 200 or not isinstance(access_token, str) or not access_token.strip():
+        logger.error(
+            f"Trusted token delegation failed for provider={provider}, "
+            f"status={resp.status_code}"
+        )
+        client_status = (
+            status.HTTP_401_UNAUTHORIZED
+            if resp.status_code in (400, 401, 403)
+            else status.HTTP_502_BAD_GATEWAY
+        )
+        raise AuthException(http_status=client_status, message=_support_message(provider))
+
+    _validate_external_token(access_token, initial_token, body, provider)
+    return access_token
 
 
 def _account_url() -> str:

app/config/schemas.py

@@ -1,5 +1,5 @@
 from enum import Enum
-from typing import Optional
+from typing import Optional, Literal
 from pydantic import BaseModel
 
 
@@ -8,8 +8,25 @@ class AuthMethod(str, Enum):
     USER_CREDENTIALS = "USER_CREDENTIALS"
 
 
+class TokenExchangeMethod(str, Enum):
+    IDENTITY_BROKERING = "identity_brokering"
+    TRUSTED_TOKEN_DELEGATION = "trusted_token_delegation"
+
+
 class BackendAuthConfig(BaseModel):
     auth_method: AuthMethod = AuthMethod.USER_CREDENTIALS
     client_credentials: Optional[str] = None
     token_provider: Optional[str] = None
     token_prefix: Optional[str] = None
+
+    # Values to be set in case of
+    # Cross-Domain Federation/Trusted Token Delegation
+    exchange_method: Literal[
+        TokenExchangeMethod.IDENTITY_BROKERING,
+        TokenExchangeMethod.TRUSTED_TOKEN_DELEGATION,
+    ] = TokenExchangeMethod.IDENTITY_BROKERING
+    token_url: Optional[str] = None
+    client_id: Optional[str] = None
+    client_secret: Optional[str] = None
+    subject_issuer: Optional[str] = None
+    audience: Optional[str] = None

app/config/settings.py

@@ -4,7 +4,7 @@
 from pydantic import Field
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-from app.config.schemas import AuthMethod, BackendAuthConfig
+from app.config.schemas import AuthMethod, BackendAuthConfig, TokenExchangeMethod
 
 
 class Settings(BaseSettings):
@@ -66,7 +66,10 @@ def load_backends_auth_config(self):
                     if backend.auth_method == AuthMethod.CLIENT_CREDENTIALS:
                         required_fields = ["client_credentials"]
                     elif backend.auth_method == AuthMethod.USER_CREDENTIALS:
-                        required_fields = ["token_provider"]
+                        if backend.exchange_method == TokenExchangeMethod.IDENTITY_BROKERING:
+                            required_fields = ["token_provider"]
+                        else:
+                            required_fields = ["token_url", "client_id"]
 
                     for field in required_fields:
                         if not getattr(backend, field, None):

app/database/models/processing_job.py

@@ -3,8 +3,7 @@
 from typing import List, Optional
 
 from loguru import logger
-from sqlalchemy import DateTime, Enum, ForeignKey, Integer, String
-from sqlalchemy.dialects.mysql import LONGTEXT
+from sqlalchemy import DateTime, Enum, ForeignKey, Integer, String, Text
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.database.db import Base
@@ -26,8 +25,8 @@ class ProcessingJobRecord(Base):
     )
     user_id: Mapped[str] = mapped_column(String(255), index=True)
     platform_job_id: Mapped[Optional[str]] = mapped_column(String(255), index=True)
-    parameters: Mapped[str] = mapped_column(LONGTEXT())
-    service: Mapped[str] = mapped_column(LONGTEXT())
+    parameters: Mapped[str] = mapped_column(Text())
+    service: Mapped[str] = mapped_column(Text())
     created: Mapped[datetime.datetime] = mapped_column(
         DateTime, default=datetime.datetime.utcnow, index=True
     )

app/database/models/upscaling_task.py

@@ -2,8 +2,7 @@
 from typing import List, Optional
 
 from loguru import logger
-from sqlalchemy import DateTime, Enum, Integer, String
-from sqlalchemy.dialects.mysql import LONGTEXT
+from sqlalchemy import DateTime, Enum, Integer, String, Text
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.database.db import Base
@@ -22,7 +21,7 @@ class UpscalingTaskRecord(Base):
         Enum(ProcessingStatusEnum), index=True
     )
     user_id: Mapped[str] = mapped_column(String(255), index=True)
-    service: Mapped[str] = mapped_column(LONGTEXT)
+    service: Mapped[str] = mapped_column(Text())
     created: Mapped[datetime.datetime] = mapped_column(
         DateTime, default=datetime.datetime.utcnow, index=True
     )

app/platforms/implementations/ogc_api_process.py

@@ -6,6 +6,8 @@
 from app.auth import exchange_token, get_current_user_claims
 from fastapi import Response
 from loguru import logger
+import jwt
+from urllib.parse import urlparse
 
 from app.platforms.base import BaseProcessingPlatform
 from app.platforms.dispatcher import register_platform
@@ -165,6 +167,22 @@ async def _create_api_client_instance(
 
         return ApiClientWrapper(configuration, **additional_args)
 
+    async def _get_token_for_api(self, user_token: str, url: str) -> str:
+        payload = jwt.decode(user_token, options={"verify_signature": False})
+
+        # Extract the 'iss' (issuer) claim safely
+        issuer = payload.get("iss")
+        parsed_uri = urlparse(issuer)
+
+        # Return token if it is a Terradue/Geohazards-TEP token
+        # (issued by iam.terradue.com)
+        if parsed_uri.netloc == "iam.terradue.com":
+            logger.debug("Skipping token exchange for iam.terradue.com issuer")
+            return user_token
+
+        # Otherwise perform token exchange (using APEx token as input)
+        return await exchange_token(user_token, url)
+
     async def execute_job(
         self,
         user_token: str,
@@ -179,7 +197,7 @@ async def execute_job(
 
         # Exchanging token
         logger.debug("Exchanging user token for OGC API Process execution...")
-        exchanged_token = await exchange_token(
+        exchanged_token = await self._get_token_for_api(
             user_token=user_token, url=details.endpoint
         )
 
@@ -529,12 +547,7 @@ def _extract_assets_from_feature_collection(
                     return collection
 
         return self._build_collection_from_features(
-            features,
-            assets,
-            result_name,
-            user_token,
-            details,
-            internal_job_id
+            features, assets, result_name, user_token, details, internal_job_id
         )
 
     async def get_job_status(
@@ -543,7 +556,7 @@ async def get_job_status(
         logger.debug(f"Fetching job status for OGC API job with ID {job_id}")
 
         logger.debug("Exchanging user token for OGC API Process execution...")
-        exchanged_token = await exchange_token(
+        exchanged_token = await self._get_token_for_api(
             user_token=user_token, url=details.endpoint
         )
 
@@ -562,7 +575,7 @@ async def get_job_results(
         logger.debug(f"Fetching job result for opfenEO job with ID {job_id}")
 
         logger.debug("Exchanging user token for OGC API Process execution...")
-        exchanged_token = await exchange_token(
+        exchanged_token = await self._get_token_for_api(
             user_token=user_token, url=details.endpoint
         )
 
@@ -667,7 +680,7 @@ async def get_service_parameters(
         )
 
         logger.debug("Exchanging user token for OGC API Process execution...")
-        exchanged_token = await exchange_token(
+        exchanged_token = await self._get_token_for_api(
             user_token=user_token, url=details.endpoint
         )
 

docker-compose-t2.yml

@@ -0,0 +1,60 @@
+version: "3.9"
+
+volumes:
+  apex-db-data:
+
+services:
+  db:
+    image: postgres:15
+    restart: unless-stopped
+    environment:
+      POSTGRES_USER: apex_user
+      POSTGRES_PASSWORD: apex_pass
+      POSTGRES_DB: apex_db
+    ports:
+      - "5432:5432"
+    volumes:
+      - apex-db-data:/var/lib/postgresql/data
+    healthcheck:
+      test: ["CMD-SHELL", "pg_isready -U apex_user -d apex_db"]
+      interval: 5s
+      timeout: 5s
+      retries: 10
+
+  app:
+    build:
+      context: .
+      dockerfile: dockerfile
+    image: apex-dispatch-api:latest
+    restart: unless-stopped
+    environment:
+      APP_ENV: development
+      KEYCLOAK_HOST: ${KEYCLOAK_HOST}
+      KEYCLOAK_REALM: ${KEYCLOAK_REALM}
+      KEYCLOAK_CLIENT_ID: ${KEYCLOAK_CLIENT_ID}
+      KEYCLOAK_CLIENT_SECRET: ${KEYCLOAK_CLIENT_SECRET}
+      DATABASE_URL: ${DATABASE_URL}
+      BACKENDS: '{"https://processing.geohazards-tep.eu": {"auth_method": "USER_CREDENTIALS", "token_provider": "gep", "token_prefix": ""}}'
+    depends_on:
+      db:
+        condition: service_healthy
+    ports:
+      - "${APP_PORT:-8000}:${APP_PORT:-8000}"
+    healthcheck:
+      # adjust path if your health endpoint differs
+      test: ["CMD-SHELL", "curl -f http://localhost:${APP_PORT:-8000}/health || exit 1"]
+      interval: 5s
+      timeout: 3s
+      retries: 12
+
+  migrate:
+    image: apex-dispatch-api:latest
+    environment:
+      DATABASE_URL: ${DATABASE_URL}
+    depends_on:
+      db:
+        condition: service_healthy
+    # run the migration only after the app is reachable, then exit
+    entrypoint: ["/bin/sh", "-c"]
+    command: >
+      "alembic upgrade head"