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

@@ -8,6 +8,7 @@
 
 from app.error import AuthException, DispatcherException
 from app.schemas.websockets import WSStatusMessage
+from app.config.schemas import BackendAuthConfig
 
 from .config.settings import settings
 
@@ -33,12 +34,19 @@ def _decode_token(token: str):
     try:
         logger.debug(f"Decoding token for user authentication: {token} with "
                      f"issuer {KEYCLOAK_BASE_URL}")
-        signing_key = jwks_client.get_signing_key_from_jwt(token).key
+        try:
+            signing_key = jwks_client.get_signing_key_from_jwt(token).key
+        except Exception as e:
+            raise
+            logger.warning(f"Signing key error: {str(e)}")
+            signing_key = ""
+        logger.warning("Before decode")
         payload = jwt.decode(
             token,
             signing_key,
             algorithms=[ALGORITHM],
             issuer=KEYCLOAK_BASE_URL,
+            options={"verify_aud": False},
         )
         return payload
     except Exception:
@@ -101,64 +109,93 @@ async def exchange_token(user_token: str, url: str) -> str:
     :return: The bearer token as a string.
     """
 
-    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:
+    if not backend_idp.token_provider:
         raise ValueError(
             f"Backend '{url}' must define 'token_provider'"
         )
 
     platform_token = await _exchange_token_for_provider(
-        initial_token=user_token, provider=provider
+        initial_token=user_token,
+        backend_idp=backend_idp
     )
     return (
-        f"{token_prefix}/{platform_token['access_token']}"
-        if token_prefix
+        f"{backend_idp.token_prefix}/{platform_token['access_token']}"
+        if backend_idp.token_prefix
         else platform_token["access_token"]
     )
 
 
 async def _exchange_token_for_provider(
-    initial_token: str, provider: str
+    initial_token: str, backend_idp: BackendAuthConfig
 ) -> 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.
+    :param backend_idp: target IDP.
 
     :return: The token response (dict) on success.
 
     :raise: Raises AuthException 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:
-        raise AuthException(
-            http_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            message="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,
-    }
-
+    if backend_idp.token_url:
+        # Cross-Domain Federation/Trusted Token Delegation
+        # Payload will be sent to the backend IDP and receive a valid access token from it
+        logger.info(f"Using Cross-Domain Federation/Trusted Token Delegation with IDP {backend_idp.token_url}")
+        if not backend_idp.client_id:
+            raise AuthException(
+                http_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                message="Token exchange not configured on the server (missing client credentials).",
+            )
+        token_url = 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"
+        }
+
+    else:
+        # Internal-to-External Token Exchange
+        # Payload will be sent to the backend IDP and receive a valid access token from it
+        logger.info(f"Using Internal-to-External Token Exchange with IDP {backend_idp.token_provider}")
+
+        token_url = f"{KEYCLOAK_BASE_URL}/protocol/openid-connect/token"
+        #token_url = "https://iam.terradue.com/realms/master/protocol/openid-connect/token"
+
+        # Check if the necessary settings are in place
+        if not settings.keycloak_client_id:
+            raise AuthException(
+                http_status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                message="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": backend_idp.token_provider,
+        }
+
+    provider_str = backend_idp.token_provider or backend_idp.token_url
     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}")
+        logger.error(
+            f"Token exchange network error for provider={provider_str}: {exc}")
         raise AuthException(
             http_status=status.HTTP_502_BAD_GATEWAY,
             message=(
-                f"Could not authenticate with {provider}. Please contact APEx support or reach out "
+                f"Could not authenticate with {provider_str}. Please contact APEx support or reach out "
                 "through the <a href='https://forum.apex.esa.int/'>APEx User Forum</a>."
             ),
         )
@@ -173,7 +210,7 @@ async def _exchange_token_for_provider(
         raise AuthException(
             http_status=status.HTTP_502_BAD_GATEWAY,
             message=(
-                f"Could not authenticate with {provider}. Please contact APEx support or reach out "
+                f"Could not authenticate with {provider_str}. Please contact APEx support or reach out "
                 "through the <a href='https://forum.apex.esa.int/'>APEx User Forum</a>."
             ),
         )
@@ -182,7 +219,7 @@ async def _exchange_token_for_provider(
         # Keycloak returns error and error_description fields for token errors
         err = body.get("error_description") or body.get("error") or resp.text
         logger.error(
-            f"Token exchange failed for provider={provider}, status={resp.status_code}, error={err}"
+            f"Token exchange failed for provider={provider_str}, status={resp.status_code}, error={err}"
         )
         # Map common upstream statuses to meaningful client statuses
         client_status = (
@@ -191,15 +228,20 @@ async def _exchange_token_for_provider(
             else status.HTTP_502_BAD_GATEWAY
         )
 
-        raise AuthException(
-            http_status=client_status,
-            message=(
-                f"Please link your account with {provider} in your "
+        if backend_idp.token_provider:
+            message = (
+                f"Please link your account with {provider_str} in your "
                 f"<a href='{settings.keycloak_host}/realms/{settings.keycloak_realm}/"
                 "account'>Account Dashboard</a>"
                 if body.get("error", "") == "not_linked"
-                else f"Could not authenticate with {provider}: {err}"
-            ),
+                else f"Could not authenticate with {provider_str}: {err}"
+            )
+        else:
+            message = f"Could not authenticate with {provider_str}: {err}"
+
+        raise AuthException(
+            http_status=client_status,
+            message=message
         )
 
     # Successful exchange, return token response (access_token, expires_in, etc.)

app/config/schemas.py

@@ -13,3 +13,11 @@ class BackendAuthConfig(BaseModel):
     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
+    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/database/models/processing_job.py

@@ -3,8 +3,8 @@
 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.dialects.mysql import LONGTEXT
 from sqlalchemy.orm import Mapped, Session, mapped_column
 
 from app.database.db import Base
@@ -26,8 +26,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/platforms/implementations/ogc_api_process.py

@@ -3,6 +3,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
@@ -160,6 +162,23 @@ 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(f"Skipping token exchange (token issued by {parsed_uri.netloc})")
+            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,
@@ -174,7 +193,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
         )
 
@@ -298,7 +317,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
         )
 
@@ -317,7 +336,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
         )
 
@@ -429,7 +448,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"

env-t2

@@ -0,0 +1,18 @@
+# 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=postgresql+psycopg2://apex_user:apex_pass@db:5432/apex_db
+
+# Keycloak Settings
+KEYCLOAK_HOST="https://iam.terradue.com"
+KEYCLOAK_REALM=master
+KEYCLOAK_CLIENT_ID=apex-client-id
+KEYCLOAK_CLIENT_SECRET=apex-client-secret
+
+
+BACKENDS='"https://openeo.backend2.com": {"auth_method": "USER_CREDENTIALS",  "token_provider": "backend", "token_prefix": "oidc/backend"}}'