diff --git a/alembic/versions/833e4a41c2ad_generation_of_tables.py b/alembic/versions/833e4a41c2ad_generation_of_tables.py index 189823e..72f521f 100644 --- a/alembic/versions/833e4a41c2ad_generation_of_tables.py +++ b/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), diff --git a/app/auth.py b/app/auth.py index 56af4b1..5fedd1f 100644 --- a/app/auth.py +++ b/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: diff --git a/app/config/schemas.py b/app/config/schemas.py index d19bd11..60a5f3e 100644 --- a/app/config/schemas.py +++ b/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 diff --git a/app/config/settings.py b/app/config/settings.py index 308b651..42f1842 100644 --- a/app/config/settings.py +++ b/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): diff --git a/app/database/models/processing_job.py b/app/database/models/processing_job.py index 3611b8e..13c309a 100644 --- a/app/database/models/processing_job.py +++ b/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 ) diff --git a/app/database/models/upscaling_task.py b/app/database/models/upscaling_task.py index f887b4b..9df0ef9 100644 --- a/app/database/models/upscaling_task.py +++ b/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 ) diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py index f9cca3d..44d6287 100644 --- a/app/platforms/implementations/ogc_api_process.py +++ b/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 ) diff --git a/docker-compose-t2.yml b/docker-compose-t2.yml new file mode 100644 index 0000000..4626a7d --- /dev/null +++ b/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" diff --git a/tests/platforms/test_ogc_api_process_platform.py b/tests/platforms/test_ogc_api_process_platform.py index 9d9cbea..b8e76e8 100644 --- a/tests/platforms/test_ogc_api_process_platform.py +++ b/tests/platforms/test_ogc_api_process_platform.py @@ -3,10 +3,18 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch +import jwt import pytest import requests +TEST_USER_TOKEN = jwt.encode( + {"iss": "https://issuer.example.com/realms/test"}, + "test-secret-for-ogc-api-tests-at-least-32-bytes", + algorithm="HS256", +) + + def _install_ogc_api_processes_client_stub(): root_module = types.ModuleType("ogc_api_processes_client") api_client_wrapper_module = types.ModuleType( @@ -455,7 +463,7 @@ async def test_execute_job_returns_namespaced_job_id( mock_create_api_client.return_value = api_client result = await platform.execute_job( - user_token="token", + user_token=TEST_USER_TOKEN, title="My job", details=ServiceDetails( endpoint="https://example.com", @@ -510,7 +518,7 @@ async def test_execute_job_returns_plain_job_id_without_namespace( mock_create_api_client.return_value = api_client result = await platform.execute_job( - user_token="token", + user_token=TEST_USER_TOKEN, title="My job", details=ServiceDetails( endpoint="https://example.com", @@ -561,7 +569,7 @@ async def test_execute_job_omits_missing_optional_user_fields( mock_create_api_client.return_value = api_client await platform.execute_job( - user_token="token", + user_token=TEST_USER_TOKEN, title="My job", details=ServiceDetails( endpoint="https://example.com", @@ -615,7 +623,7 @@ async def test_get_job_status_maps_client_status( mock_create_api_client.return_value = api_client result = await platform.get_job_status( - user_token="token", + user_token=TEST_USER_TOKEN, job_id="ns:job-123", details=ServiceDetails( endpoint="https://example.com", @@ -655,7 +663,7 @@ async def test_get_job_results_returns_stac_collection( mock_create_api_client.return_value = api_client result = await platform.get_job_results( - user_token="token", + user_token=TEST_USER_TOKEN, job_id="ns:job-123", details=ServiceDetails( endpoint="https://example.com", @@ -710,7 +718,7 @@ async def test_get_job_results_follows_geojson_collection_link( ) result = await platform.get_job_results( - user_token="token", + user_token=TEST_USER_TOKEN, job_id="ns:job-123", details=ServiceDetails( endpoint="https://example.com", @@ -745,7 +753,7 @@ async def test_get_job_results_returns_empty_collection_when_no_supported_result mock_create_api_client.return_value = api_client result = await platform.get_job_results( - user_token="token", + user_token=TEST_USER_TOKEN, job_id="ns:job-123", details=ServiceDetails( endpoint="https://example.com", @@ -799,7 +807,7 @@ async def test_get_service_parameters_maps_geojson_and_options( mock_create_api_client.return_value = api_client result = await platform.get_service_parameters( - user_token="token", + user_token=TEST_USER_TOKEN, details=ServiceDetails( endpoint="https://example.com", namespace="ns", @@ -891,7 +899,7 @@ async def test_get_service_parameters_maps_all_supported_types( mock_create_api_client.return_value = api_client result = await platform.get_service_parameters( - user_token="token", + user_token=TEST_USER_TOKEN, details=ServiceDetails( endpoint="https://example.com", namespace="ns",