Refactor iam client into task_handler and make toke url configurable

Author: mario-winkler

config-landsat.yaml

@@ -4,6 +4,8 @@ worker:
     host: https://registration-harvester-api.develop.eoepca.org/flowable-rest
     tls: true
     cacert: ./etc/eoepca-ca-chain.pem
+  iam:
+    oidc_token_endpoint_url: https://iam-auth.develop.eoepca.org/realms/eoepca/protocol/openid-connect/token    
   topics:
     landsat_discover_data:
       module: worker.landsat.tasks

config-sentinel.yaml

@@ -4,6 +4,8 @@ worker:
     host: https://registration-harvester-api.develop.eoepca.org/flowable-rest
     tls: true
     cacert: ./etc/eoepca-ca-chain.pem
+  iam:
+    oidc_token_endpoint_url: https://iam-auth.develop.eoepca.org/realms/eoepca/protocol/openid-connect/token    
   topics:
     sentinel_discover_data:
       module: worker.sentinel.tasks

config-static-catalog.yaml

@@ -4,6 +4,8 @@ worker:
     host: https://registration-harvester-api.develop.eoepca.org/flowable-rest
     tls: true
     cacert: ./etc/eoepca-ca-chain.pem
+  iam:
+    oidc_token_endpoint_url: https://iam-auth.develop.eoepca.org/realms/eoepca/protocol/openid-connect/token
   topics:
     stac_publish_catalog:
       module: worker.stac.tasks

src/worker/common/task_handler.py

@@ -1,3 +1,6 @@
+from worker.common.config import worker_config
+from worker.common.iam import IAMClient
+from worker.common.secrets import worker_secrets
 from worker.common.types import ExternalJob, JobResult, JobResultBuilder
 
 
@@ -17,6 +20,20 @@ def __init__(self, handlers_config: dict):
             **self.config_all.get("subscription", {}),
         }
 
+        # IAM client
+        iam_client_id = worker_secrets.get_secret("iam_client_id", None)
+        iam_client_secret = worker_secrets.get_secret("iam_client_secret", None)
+        iam_oidc_token_endpoint_url = "https://iam-auth.develop.eoepca.org/realms/eoepca/protocol/openid-connect/token"
+        iam_config = worker_config.get_all().get("iam")
+        if iam_config is not None:
+            token_url = iam_config.get("oidc_token_endpoint_url")
+            if token_url is not None:
+                iam_oidc_token_endpoint_url = token_url
+
+        self.iam_client = IAMClient(
+            token_endpoint_url=iam_oidc_token_endpoint_url, client_id=iam_client_id, client_secret=iam_client_secret
+        )
+
     def execute(self, job: ExternalJob, result: JobResultBuilder, config: dict) -> JobResult:
         raise NotImplementedError
 

src/worker/stac/tasks.py

@@ -6,21 +6,12 @@
 from httpx import HTTPStatusError
 from pystac import Catalog, Collection, Item, StacIO
 
-from worker.common.iam import IAMClient
 from worker.common.log_utils import configure_logging, log_with_context
-from worker.common.secrets import worker_secrets
 from worker.common.task_handler import TaskHandler
 from worker.common.types import ExternalJob, JobResult, JobResultBuilder
 
 configure_logging()
 
-iam_client_id = worker_secrets.get_secret("iam_client_id", None)
-iam_client_secret = worker_secrets.get_secret("iam_client_secret", None)
-iam_oidc_token_endpoint_url = "https://iam-auth.apx.develop.eoepca.org/realms/eoepca/protocol/openid-connect/token"
-iam_client = IAMClient(
-    token_endpoint_url=iam_oidc_token_endpoint_url, client_id=iam_client_id, client_secret=iam_client_secret
-)
-
 
 class StacCatalogHandler(TaskHandler):
     def execute(self, job: ExternalJob, result: JobResultBuilder, config: dict) -> JobResult:
@@ -140,7 +131,7 @@ def execute(self, job: ExternalJob, result: JobResultBuilder, config: dict) -> J
             return result.failure()
 
         # Get token to access protected endpoints of catalog
-        token = iam_client.get_access_token()
+        token = self.iam_client.get_access_token()
         headers = {"Content-Type": "application/json", "Authorization": f"Bearer {token}"}
 
         # Publish stac collection
@@ -236,7 +227,7 @@ def execute(self, job: ExternalJob, result: JobResultBuilder, config: dict) -> J
             return result.failure()
 
         # Get token to access protected endpoints of catalog
-        token = iam_client.get_access_token()
+        token = self.iam_client.get_access_token()
         headers = {"Content-Type": "application/json", "Authorization": f"Bearer {token}"}
 
         # Publish STAC item