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 4dd88d6..14e4df7 100644
--- a/app/auth.py
+++ b/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 APEx User Forum."
),
)
@@ -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 APEx User Forum."
),
)
@@ -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"Account Dashboard"
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.)
diff --git a/app/config/schemas.py b/app/config/schemas.py
index d19bd11..f64d470 100644
--- a/app/config/schemas.py
+++ b/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
diff --git a/app/database/models/processing_job.py b/app/database/models/processing_job.py
index 3611b8e..20d699d 100644
--- a/app/database/models/processing_job.py
+++ b/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
)
diff --git a/app/platforms/implementations/ogc_api_process.py b/app/platforms/implementations/ogc_api_process.py
index 546f74f..1a15d08 100644
--- a/app/platforms/implementations/ogc_api_process.py
+++ b/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
)
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/env-t2 b/env-t2
new file mode 100644
index 0000000..ebc284c
--- /dev/null
+++ b/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"}}'