Support DefaultDataCredential in Service (Workflow Data, Logs, Apps)

src/lib/data/storage/backends/tests/test_backends.py

@@ -534,6 +534,53 @@ def test_workflow_config_with_null_credential(self):
         # Assert
         self.assertIsNone(config.workflow_data.credential)
 
+    def test_workflow_config_data_with_default_credential(self):
+        """Test WorkflowConfig DataConfig accepts DefaultDataCredential."""
+        default_cred = credentials.DefaultDataCredential(
+            endpoint='s3://bucket.io/workflows',
+            region='us-west-2',
+        )
+
+        config = postgres.WorkflowConfig(
+            workflow_data=postgres.DataConfig(credential=default_cred),
+        )
+
+        credential = config.workflow_data.credential
+        self.assertIsInstance(credential, credentials.DefaultDataCredential)
+        assert isinstance(credential, credentials.DefaultDataCredential)
+        self.assertEqual(credential.endpoint, 's3://bucket.io/workflows')
+        self.assertEqual(credential.region, 'us-west-2')
+
+    def test_workflow_config_log_with_default_credential(self):
+        """Test WorkflowConfig LogConfig accepts DefaultDataCredential."""
+        default_cred = credentials.DefaultDataCredential(
+            endpoint='s3://log-bucket.io/logs',
+            region='us-east-1',
+        )
+
+        config = postgres.WorkflowConfig(
+            workflow_log=postgres.LogConfig(credential=default_cred),
+        )
+
+        self.assertIsInstance(
+            config.workflow_log.credential,
+            credentials.DefaultDataCredential,
+        )
+
+    def test_default_credential_to_decrypted_dict_no_keys(self):
+        """Test DefaultDataCredential.to_decrypted_dict has no access keys."""
+        default_cred = credentials.DefaultDataCredential(
+            endpoint='s3://bucket.io/data',
+            region='us-west-2',
+        )
+
+        result = default_cred.to_decrypted_dict()
+
+        self.assertEqual(result['endpoint'], 's3://bucket.io/data')
+        self.assertEqual(result['region'], 'us-west-2')
+        self.assertNotIn('access_key_id', result)
+        self.assertNotIn('access_key', result)
+
 
 if __name__ == '__main__':
     unittest.main()

src/runtime/pkg/data/data.go

@@ -414,8 +414,17 @@ func MountURL(downloadType string, credentialInfo ConfigInfo, urlPath string,
 		osmoChan <- fmt.Sprintf("Missing data credential for %s.", storageBackend.GetProfile())
 		return isEmpty
 	}
-	os.Setenv("AWS_ACCESS_KEY_ID", dataCredential.AccessKeyId)
-	os.Setenv("AWS_SECRET_ACCESS_KEY", dataCredential.AccessKey)
+	// Only set static key env vars when keys are provided.
+	// When using DefaultDataCredential (ambient credentials via Pod Identity,
+	// IRSA, etc.), keys are empty — setting empty env vars would clobber the
+	// SDK's default credential chain.
+	if dataCredential.AccessKeyId != "" {
+		os.Setenv("AWS_ACCESS_KEY_ID", dataCredential.AccessKeyId)
+		os.Setenv("AWS_SECRET_ACCESS_KEY", dataCredential.AccessKey)
+	}
+	if dataCredential.Region != "" {
+		os.Setenv("AWS_REGION", dataCredential.Region)
+	}
 
 	var commandArgs []string
 

src/utils/connectors/postgres.py

@@ -2686,7 +2686,7 @@ def construct_path(endpoint: str, bucket: str, path: str):
 
 class LogConfig(ExtraArgBaseModel):
     """ Config for storing information about data. """
-    credential: credentials.StaticDataCredential | None = None
+    credential: credentials.DataCredential | None = None
 
 
 class WorkflowInfo(ExtraArgBaseModel):
@@ -2703,7 +2703,7 @@ def validate_name(self, name: str):
 
 class DataConfig(ExtraArgBaseModel):
     """ Config for storing information about data. """
-    credential: credentials.StaticDataCredential | None = None
+    credential: credentials.DataCredential | None = None
 
     base_url: str = ''
     # Timeout in mins for osmo-ctrl to retry connecting to the OSMO service until exiting the task

src/utils/job/task.py

@@ -26,6 +26,7 @@
 import re
 import secrets
 import time
+from collections.abc import Mapping
 from typing import Any, Dict, List, Optional, Set, Tuple, Union
 from urllib.parse import urlencode
 
@@ -102,7 +103,7 @@ def create_login_dict(user: str,
 
 
 def create_config_dict(
-    data_info: dict[str, credentials.StaticDataCredential],
+    data_info: Mapping[str, credentials.DataCredential],
 ) -> dict:
     '''
     Creates the config dict where the input should be a dict containing key values like:
@@ -2699,7 +2700,7 @@ def convert_to_pod_spec(
         service_config: connectors.ServiceConfig | None = None,
         dataset_config: connectors.DatasetConfig | None = None,
         pool_info: connectors.Pool | None = None,
-        data_endpoints: Dict[str, credentials.StaticDataCredential] | None = None,
+        data_endpoints: Mapping[str, credentials.DataCredential] | None = None,
         skip_refresh_token: bool = False,
         auth_token: str | None = None,
     ) -> Tuple[Dict, Dict[str, kb_objects.FileMount], Optional[Tuple[str, str]]]:
@@ -3193,10 +3194,10 @@ def decode_hstore(tasks: str) -> Set[str]:
 
 def fetch_creds(
     user: str,
-    data_creds: dict[str, credentials.StaticDataCredential],
+    data_creds: Mapping[str, credentials.DataCredential],
     path: str,
     disabled_data: list[str] | None = None,
-) -> credentials.StaticDataCredential | None:
+) -> credentials.DataCredential | None:
     backend_info = storage.construct_storage_backend(path)
 
     if backend_info.profile not in data_creds:

src/utils/job/tests/BUILD

@@ -67,6 +67,7 @@ py_test(
     tags = ["manual"],
     deps = [
         "//src/lib/utils:common",
+        "//src/lib/utils:credentials",
         "//src/utils/connectors",
         "//src/utils/job",
     ]

src/utils/job/tests/test_task.py

@@ -21,7 +21,7 @@
 from unittest import mock
 import unittest
 
-from src.lib.utils import common
+from src.lib.utils import common, credentials
 from src.utils.job import task, kb_objects
 from src.utils import connectors
 
@@ -869,5 +869,67 @@ def test_file_mount_credential_not_in_secrets(self):
         self.assertEqual(len(user_secret_call), 0)
 
 
+class CreateConfigDictTest(unittest.TestCase):
+    """Tests for create_config_dict with different credential types."""
+
+    def test_static_credential(self):
+        """Test create_config_dict with StaticDataCredential."""
+        static_cred = credentials.StaticDataCredential(
+            endpoint='s3://my-bucket',
+            access_key_id='AKIAIOSFODNN7EXAMPLE',
+            access_key='wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY',
+            region='us-east-1',
+        )
+
+        result = task.create_config_dict({'s3://my-bucket': static_cred})
+
+        data_entry = result['auth']['data']['s3://my-bucket']
+        self.assertEqual(data_entry['access_key_id'], 'AKIAIOSFODNN7EXAMPLE')
+        self.assertEqual(data_entry['access_key'], 'wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY')
+        self.assertEqual(data_entry['endpoint'], 's3://my-bucket')
+        self.assertEqual(data_entry['region'], 'us-east-1')
+
+    def test_default_credential(self):
+        """Test create_config_dict with DefaultDataCredential produces no access keys."""
+        default_cred = credentials.DefaultDataCredential(
+            endpoint='s3://ambient-bucket',
+            region='us-west-2',
+        )
+
+        result = task.create_config_dict({'s3://ambient-bucket': default_cred})
+
+        data_entry = result['auth']['data']['s3://ambient-bucket']
+        self.assertEqual(data_entry['endpoint'], 's3://ambient-bucket')
+        self.assertEqual(data_entry['region'], 'us-west-2')
+        self.assertNotIn('access_key_id', data_entry)
+        self.assertNotIn('access_key', data_entry)
+
+    def test_mixed_credentials(self):
+        """Test create_config_dict with both credential types."""
+        static_cred = credentials.StaticDataCredential(
+            endpoint='s3://static-bucket',
+            access_key_id='AKIAIOSFODNN7EXAMPLE',
+            access_key='secret',
+        )
+        default_cred = credentials.DefaultDataCredential(
+            endpoint='s3://ambient-bucket',
+            region='eu-west-1',
+        )
+
+        result = task.create_config_dict({
+            's3://static-bucket': static_cred,
+            's3://ambient-bucket': default_cred,
+        })
+
+        static_entry = result['auth']['data']['s3://static-bucket']
+        self.assertIn('access_key_id', static_entry)
+        self.assertIn('access_key', static_entry)
+
+        ambient_entry = result['auth']['data']['s3://ambient-bucket']
+        self.assertNotIn('access_key_id', ambient_entry)
+        self.assertNotIn('access_key', ambient_entry)
+        self.assertEqual(ambient_entry['region'], 'eu-west-1')
+
+
 if __name__ == '__main__':
     unittest.main()