[CDF-24214] 👮Standardize Transformation Authentication

cognite_toolkit/_cdf_tk/loaders/_resource_loaders/function_loaders.py

@@ -415,7 +415,13 @@ def load_resource_file(
         # is used to compare with the CDF resource.
         for resource in resources:
             identifier = self.get_id(resource)
-            credentials = read_auth(identifier, resource, self.client, "function schedule", self.console)
+            credentials = read_auth(
+                resource.get("authentication"),
+                self.client.config,
+                identifier,
+                "function schedule",
+                console=self.console,
+            )
             self.authentication_by_id[identifier] = credentials
             auth_hash = calculate_secure_hash(credentials.dump(camel_case=True), shorten=True)
             extra_str = f" {self._hash_key}: {auth_hash}"

cognite_toolkit/_cdf_tk/loaders/_resource_loaders/transformation_loaders.py

@@ -29,11 +29,13 @@
 import warnings
 from collections import defaultdict
 from collections.abc import Hashable, Iterable, Sequence
+from copy import deepcopy
 from functools import lru_cache
 from pathlib import Path
-from typing import Any, cast, final
+from typing import Any, Literal, cast, final
 
 from cognite.client.data_classes import (
+    ClientCredentials,
     OidcCredentials,
     Transformation,
     TransformationList,
@@ -54,23 +56,26 @@
     DataModelId,
     ViewId,
 )
+from cognite.client.data_classes.transformations import NonceCredentials
 from cognite.client.data_classes.transformations.notifications import (
     TransformationNotificationWrite,
     TransformationNotificationWriteList,
 )
 from cognite.client.exceptions import CogniteAPIError, CogniteAuthError, CogniteDuplicatedError, CogniteNotFoundError
 from cognite.client.utils.useful_types import SequenceNotStr
 from rich import print
+from rich.console import Console
 
 from cognite_toolkit._cdf_tk._parameters import ANY_INT, ParameterSpec, ParameterSpecSet
+from cognite_toolkit._cdf_tk.client import ToolkitClient
 from cognite_toolkit._cdf_tk.client.data_classes.raw import RawDatabase, RawTable
 from cognite_toolkit._cdf_tk.constants import BUILD_FOLDER_ENCODING
 from cognite_toolkit._cdf_tk.exceptions import (
     ResourceCreationError,
     ToolkitFileNotFoundError,
     ToolkitInvalidParameterNameError,
+    ToolkitNotSupported,
     ToolkitRequiredValueError,
-    ToolkitTypeError,
     ToolkitYAMLFormatError,
 )
 from cognite_toolkit._cdf_tk.loaders._base_loaders import ResourceLoader
@@ -82,7 +87,7 @@
     quote_int_value_by_key_in_yaml,
     safe_read,
 )
-from cognite_toolkit._cdf_tk.utils.cdf import try_find_error
+from cognite_toolkit._cdf_tk.utils.cdf import read_auth, try_find_error
 from cognite_toolkit._cdf_tk.utils.diff_list import diff_list_hashable
 
 from .auth_loaders import GroupAllScopedLoader
@@ -122,6 +127,13 @@ class TransformationLoader(
     _doc_url = "Transformations/operation/createTransformations"
     _hash_key = "-- cdf-auth"
 
+    def __init__(self, client: ToolkitClient, build_dir: Path | None, console: Console | None = None):
+        super().__init__(client, build_dir, console)
+        self._authentication_by_id_operation: dict[
+            tuple[str, Literal["read", "write"]], OidcCredentials | ClientCredentials
+        ] = {}
+        self._nonce_cache: dict[str, NonceCredentials] = {}
+
     @property
     def display_name(self) -> str:
         return "transformations"
@@ -234,21 +246,73 @@ def load_resource_file(
                 item["query"] = safe_read(query_file, encoding=BUILD_FOLDER_ENCODING)
 
             auth_dict: dict[str, Any] = {}
-            for key in [
-                "authentication",
-                "sourceOidcCredentials",
-                "destinationOidcCredentials",
-                "sourceNonce",
-                "destinationNonce",
-            ]:
-                if key in item:
-                    auth_dict[key] = item[key]
+            if "authentication" in item:
+                auth_dict["authentication"] = item["authentication"]
             if auth_dict:
                 auth_hash = calculate_secure_hash(auth_dict, shorten=True)
                 if "query" in item:
                     hash_str = f"{self._hash_key}: {auth_hash}"
                     if not item["query"].startswith(self._hash_key):
                         item["query"] = f"{hash_str}\n{item['query']}"
+
+            if "sourceOidcCredentials" in item:
+                raise ToolkitNotSupported(
+                    "The property 'sourceOidcCredentials' is not supported. Use 'authentication.read' instead."
+                )
+
+            if "destinationOidcCredentials" in item:
+                raise ToolkitNotSupported(
+                    "The property 'destinationOidcCredentials' is not supported. Use 'authentication.write' instead."
+                )
+
+            if "sourceNonce" in item:
+                raise ToolkitNotSupported(
+                    "The property 'sourceNonce' is not supported by Toolkit. Use 'authentication.read' instead,"
+                    "then Toolkit will dynamically set the nonce."
+                )
+
+            if "destinationNonce" in item:
+                raise ToolkitNotSupported(
+                    "The property 'destinationNonce' is not supported by Toolkit. Use 'authentication.write' instead,"
+                    "then Toolkit will dynamically set the nonce."
+                )
+            auth = item.pop("authentication", None)
+            if isinstance(auth, dict) and "read" in auth:
+                self._authentication_by_id_operation[(external_id, "read")] = read_auth(
+                    auth["read"],
+                    self.client.config,
+                    external_id,
+                    "transformation",
+                    allow_oidc=True,
+                    console=self.console,
+                )
+            elif isinstance(auth, dict) and "write" not in auth:
+                self._authentication_by_id_operation[(external_id, "read")] = read_auth(
+                    auth,
+                    self.client.config,
+                    external_id,
+                    "transformation",
+                    allow_oidc=True,
+                    console=self.console,
+                )
+            if isinstance(auth, dict) and "write" in auth:
+                self._authentication_by_id_operation[(external_id, "write")] = read_auth(
+                    auth["write"],
+                    self.client.config,
+                    external_id,
+                    "transformation",
+                    allow_oidc=True,
+                    console=self.console,
+                )
+            elif isinstance(auth, dict) and "read" not in auth:
+                self._authentication_by_id_operation[(external_id, "write")] = read_auth(
+                    auth,
+                    self.client.config,
+                    external_id,
+                    "transformation",
+                    allow_oidc=True,
+                    console=self.console,
+                )
         return raw_list
 
     def load_resource(self, resource: dict[str, Any], is_dry_run: bool = False) -> TransformationWrite:
@@ -270,29 +334,7 @@ def load_resource(self, resource: dict[str, Any], is_dry_run: bool = False) -> T
         if "conflictMode" not in resource:
             # Todo; Bug SDK missing default value
             resource["conflictMode"] = "upsert"
-
-        source_oidc_credentials = (
-            resource.get("authentication", {}).get("read") or resource.get("authentication") or None
-        )
-        destination_oidc_credentials = (
-            resource.get("authentication", {}).get("write") or resource.get("authentication") or None
-        )
-        transformation = TransformationWrite._load(resource)
-        try:
-            if transformation.source_oidc_credentials is None:
-                transformation.source_oidc_credentials = source_oidc_credentials and OidcCredentials.load(
-                    source_oidc_credentials
-                )
-            if transformation.destination_oidc_credentials is None:
-                transformation.destination_oidc_credentials = destination_oidc_credentials and OidcCredentials.load(
-                    destination_oidc_credentials
-                )
-        except KeyError as e:
-            item_id = self.get_id(resource)
-            raise ToolkitTypeError(
-                f"Ill-formed Transformation {item_id}: Authentication property is missing required fields"
-            ) from e
-        return transformation
+        return TransformationWrite._load(resource)
 
     def dump_resource(self, resource: Transformation, local: dict[str, Any] | None = None) -> dict[str, Any]:
         dumped = resource.as_write().dump()
@@ -327,12 +369,13 @@ def create(self, items: Sequence[TransformationWrite]) -> TransformationList:
             warnings.simplefilter("ignore")
             # Ignoring warnings from SDK about session unauthorized. Motivation is CDF is not fast enough to
             # handle first a group that authorizes the session and then the transformation.
-            try:
-                return self.client.transformations.create(items)
-            except CogniteAuthError as e:
-                if error := self._create_auth_creation_error(items):
-                    raise error from e
-                raise e
+            self._update_nonce(items)
+        try:
+            return self.client.transformations.create(items)
+        except CogniteAuthError as e:
+            if error := self._create_auth_creation_error(items):
+                raise error from e
+            raise e
 
     def retrieve(self, ids: SequenceNotStr[str | int]) -> TransformationList:
         internal_ids, external_ids = self._split_ids(ids)
@@ -345,6 +388,7 @@ def update(self, items: Sequence[TransformationWrite]) -> TransformationList:
             warnings.simplefilter("ignore")
             # Ignoring warnings from SDK about session unauthorized. Motivation is CDF is not fast enough to
             # handle first a group that authorizes the session and then the transformation.
+            self._update_nonce(items)
             try:
                 return self.client.transformations.update(items, mode="replace")
             except CogniteAuthError as e:
@@ -377,6 +421,34 @@ def delete(self, ids: SequenceNotStr[str | int]) -> int:
             self.client.transformations.delete(id=existing, ignore_unknown_ids=True)
         return len(existing)
 
+    def _update_nonce(self, items: Sequence[TransformationWrite]) -> None:
+        for item in items:
+            if not item.external_id:
+                raise ToolkitRequiredValueError("Transformation must have external_id set.")
+            if read_credentials := self._authentication_by_id_operation.get((item.external_id, "read")):
+                item.source_nonce = self._create_nonce(read_credentials)
+            if write_credentials := self._authentication_by_id_operation.get((item.external_id, "write")):
+                item.destination_nonce = self._create_nonce(write_credentials)
+
+    def _create_nonce(self, credentials: OidcCredentials | ClientCredentials) -> NonceCredentials:
+        key = calculate_secure_hash(credentials.dump(), shorten=True)
+        if key in self._nonce_cache:
+            return self._nonce_cache[key]
+        if isinstance(credentials, ClientCredentials):
+            session = self.client.iam.sessions.create(credentials)
+            nonce = NonceCredentials(session.id, session.nonce, self.client.config.project)
+        elif isinstance(credentials, OidcCredentials):
+            config = deepcopy(self.client.config)
+            config.project = credentials.cdf_project_name
+            config.credentials = credentials.as_credential_provider()
+            other_client = ToolkitClient(config)
+            session = other_client.iam.sessions.create(credentials.as_client_credentials())
+            nonce = NonceCredentials(session.id, session.nonce, credentials.cdf_project_name)
+        else:
+            raise ValueError(f"Error in TransformationLoader: {type(credentials)} is not a valid credentials type")
+        self._nonce_cache[key] = nonce
+        return nonce
+
     def _iterate(
         self,
         data_set_external_id: str | None = None,
@@ -430,6 +502,12 @@ def sensitive_strings(self, item: TransformationWrite) -> Iterable[str]:
         if item.destination_oidc_credentials:
             yield item.destination_oidc_credentials.client_secret
 
+        external_id = self.get_id(item)
+        if read_credentials := self._authentication_by_id_operation.get((external_id, "read")):
+            yield read_credentials.client_secret
+        if write_credentials := self._authentication_by_id_operation.get((external_id, "write")):
+            yield write_credentials.client_secret
+
 
 @final
 class TransformationScheduleLoader(

cognite_toolkit/_cdf_tk/loaders/_resource_loaders/workflow_loaders.py

@@ -664,7 +664,9 @@ def load_resource_file(
         # is used to compare with the CDF resource.
         for resource in resources:
             identifier = self.get_id(resource)
-            credentials = read_auth(identifier, resource, self.client, "workflow trigger", self.console)
+            credentials = read_auth(
+                resource.get("authentication"), self.client.config, identifier, "workflow trigger", console=self.console
+            )
             self._authentication_by_id[identifier] = credentials
             if "metadata" not in resource:
                 resource["metadata"] = {}

cognite_toolkit/_cdf_tk/utils/cdf.py

@@ -12,7 +12,7 @@
 from cognite.client.exceptions import CogniteAPIError
 from rich.console import Console
 
-from cognite_toolkit._cdf_tk.client import ToolkitClient
+from cognite_toolkit._cdf_tk.client import ToolkitClient, ToolkitClientConfig
 from cognite_toolkit._cdf_tk.constants import ENV_VAR_PATTERN
 from cognite_toolkit._cdf_tk.exceptions import (
     ToolkitRequiredValueError,
@@ -115,31 +115,54 @@ def iterate_instances(
         body["cursor"] = next_cursor
 
 
+@overload
 def read_auth(
+    authentication: object,
+    client_config: ToolkitClientConfig,
+    identifier: Hashable,
+    resource_name: str,
+    allow_oidc: Literal[False] = False,
+    console: Console | None = None,
+) -> ClientCredentials: ...
+
+
+@overload
+def read_auth(
+    authentication: object,
+    client_config: ToolkitClientConfig,
+    identifier: Hashable,
+    resource_name: str,
+    allow_oidc: Literal[True],
+    console: Console | None = None,
+) -> ClientCredentials | OidcCredentials: ...
+
+
+def read_auth(
+    authentication: object,
+    client_config: ToolkitClientConfig,
     identifier: Hashable,
-    resource: dict[str, Any],
-    client: ToolkitClient,
     resource_name: str,
+    allow_oidc: bool = False,
     console: Console | None = None,
-) -> ClientCredentials:
-    auth = resource.get("authentication")
-    if auth is None:
-        if client.config.is_strict_validation or not isinstance(client.config.credentials, OAuthClientCredentials):
+) -> ClientCredentials | OidcCredentials:
+    if authentication is None:
+        if client_config.is_strict_validation or not isinstance(client_config.credentials, OAuthClientCredentials):
             raise ToolkitRequiredValueError(f"Authentication is missing for {resource_name} {identifier!r}.")
         else:
             HighSeverityWarning(
                 f"Authentication is missing for {resource_name} {identifier!r}. Falling back to the Toolkit credentials"
             ).print_warning(console=console)
-        credentials = ClientCredentials(client.config.credentials.client_id, client.config.credentials.client_secret)
-    elif not isinstance(auth, dict):
+        return ClientCredentials(client_config.credentials.client_id, client_config.credentials.client_secret)
+    elif not isinstance(authentication, dict):
         raise ToolkitTypeError(f"Authentication must be a dictionary for {resource_name} {identifier!r}")
-    elif "clientId" not in auth or "clientSecret" not in auth:
+    elif "clientId" not in authentication or "clientSecret" not in authentication:
         raise ToolkitRequiredValueError(
             f"Authentication must contain clientId and clientSecret for {resource_name} {identifier!r}"
         )
+    elif allow_oidc and "tokenUri" in authentication and "cdfProjectName" in authentication:
+        return OidcCredentials.load(authentication)
     else:
-        credentials = ClientCredentials(auth["clientId"], auth["clientSecret"])
-    return credentials
+        return ClientCredentials(authentication["clientId"], authentication["clientSecret"])
 
 
 def metadata_key_counts(