Implement workload_identity for USER object type

Author: littleK0i

CHANGELOG.md

@@ -1,8 +1,14 @@
 # Changelog
 
+## [0.58.0] - 2025-10-06
+
+- Introduced `workload_identity` authenticator and relevant CLI options (thanks to @jbylina).
+- Introduced `workload_identity` parameter for `USER` object type and `--refresh-workload-identity` CLI option.
+- Made `object_type` and `object_name` mandatory for `SNAPSHOT_SET`.
+
 ## [0.57.1] - 2025-09-19
 
-- Fix typo in `SHARE` object type definition (thanks to @rex911).
+- Fixed typo in `SHARE` object type definition (thanks to @rex911).
 - Update AbstractRoleResolver to not transfer ownership of notebooks and shares (thanks to @rex911).
 
 ## [0.57.0] - 2025-09-16

snowddl/app/base.py

@@ -251,6 +251,9 @@ def init_arguments_parser(self):
         parser.add_argument(
             "--refresh-user-passwords", help="Additionally refresh passwords of users", default=False, action="store_true"
         )
+        parser.add_argument(
+            "--refresh-workload-identity", help="Additionally refresh workload identites of users", default=False, action="store_true"
+        )
         parser.add_argument(
             "--refresh-future-grants",
             help="Additionally refresh missing grants for existing objects derived from future grants",
@@ -504,6 +507,9 @@ def init_settings(self):
         if self.args.get("refresh_user_passwords"):
             settings.refresh_user_passwords = True
 
+        if self.args.get("refresh_workload_identity"):
+            settings.refresh_workload_identity = True
+
         if self.args.get("refresh_future_grants"):
             settings.refresh_future_grants = True
 

snowddl/blueprint/blueprint.py

@@ -477,6 +477,7 @@ class UserBlueprint(AbstractBlueprint):
     default_warehouse: Optional[AccountObjectIdent] = None
     default_namespace: Optional[Union[DatabaseIdent, SchemaIdent]] = None
     session_params: Dict[str, Union[bool, float, int, str]] = {}
+    workload_identity: Optional[Dict[str, Union[bool, float, int, str, list]]] = None
 
 
 class ViewBlueprint(SchemaObjectBlueprint, DependsOnMixin):

snowddl/parser/user.py

@@ -79,6 +79,12 @@
             "network_policy": {
                 "type": "string",
             },
+            "workload_identity": {
+                "type": "object",
+                "additionalProperties": {
+                    "type": ["array", "boolean", "number", "string"]
+                }
+            },
         },
         "additionalProperties": False
     }
@@ -132,6 +138,7 @@ def process_user(self, user_name, user_params, default_wh_map):
             default_warehouse=AccountObjectIdent(self.env_prefix, default_warehouse) if default_warehouse else None,
             default_namespace=build_default_namespace_ident(self.env_prefix, user_params.get("default_namespace")) if user_params.get("default_namespace") else None,
             session_params=self.normalise_params_dict(user_params.get("session_params", {})),
+            workload_identity=self.normalise_params_dict(user_params.get("workload_identity")),
             business_roles=business_roles,
             comment=user_params.get("comment"),
         )

snowddl/resolver/user.py

@@ -37,6 +37,8 @@ def get_existing_objects(self):
                 "has_password": r["has_password"] == "true",
                 "has_rsa_public_key": r["has_rsa_public_key"] == "true",
                 "has_mfa": r["has_mfa"] == "true",
+                "has_pat": r["has_pat"] == "true",
+                "has_workload_identity": r["has_workload_identity"] == "true",
                 "comment": r["comment"] if r["comment"] else None,
             }
 
@@ -96,6 +98,12 @@ def create_object(self, bp: UserBlueprint):
         if bp.type:
             query.append_nl("TYPE = {type}", {"type": bp.type})
 
+        # Workload identity
+        if bp.workload_identity:
+            query.append_nl("WORKLOAD_IDENTITY = (")
+            query.append(self._build_workload_identity_parameters(bp))
+            query.append_nl(")")
+
         # Object and session parameters
         query.append(self._build_common_parameters(bp))
 
@@ -123,9 +131,15 @@ def compare_object(self, bp: UserBlueprint, row: dict):
         if self._compare_public_keys(bp, row):
             result = ResolveResult.ALTER
 
+        if self._compare_workload_identity_pre_type(bp, row):
+            result = ResolveResult.ALTER
+
         if self._compare_type(bp, row):
             result = ResolveResult.ALTER
 
+        if self._compare_workload_identity_post_type(bp, row):
+            result = ResolveResult.ALTER
+
         if self._compare_parameters(bp):
             result = ResolveResult.ALTER
 
@@ -158,6 +172,23 @@ def _build_common_parameters(self, bp: UserBlueprint):
 
         return query
 
+    def _build_workload_identity_parameters(self, bp: UserBlueprint):
+        query = self.engine.query_builder()
+
+        for param_name, param_value in bp.workload_identity.items():
+            query.append_nl(
+                "    {param_name:r} = {param_value:dp}",
+                {
+                    "param_name": param_name,
+                    # ISSUER + SUBJECT is the unique key in Snowflake
+                    # SnowDDL has to append env_prefix in order to prevent duplicate key error
+                    # Feel free to adjust this logic for your own custom test environment
+                    "param_value": f"{param_value}:{self.config.env_prefix.rstrip('_$')}" if param_name == "SUBJECT" else param_value,
+                },
+            )
+
+        return query
+
     def _compare_properties(self, bp: UserBlueprint, row: dict):
         query = self.engine.query_builder()
 
@@ -336,6 +367,39 @@ def _compare_parameters(self, bp: UserBlueprint):
 
         return False
 
+    def _compare_workload_identity_pre_type(self, bp: UserBlueprint, row: dict):
+        if not bp.workload_identity and row["has_workload_identity"]:
+            self.engine.execute_safe_ddl(
+                "ALTER USER {name:i} UNSET WORKLOAD_IDENTITY",
+                {
+                    "name": bp.full_name,
+                },
+            )
+
+            return True
+
+        return False
+
+    def _compare_workload_identity_post_type(self, bp: UserBlueprint, row: dict):
+        if bp.workload_identity and (not row["has_workload_identity"] or self.engine.settings.refresh_workload_identity):
+            query = self.engine.query_builder()
+
+            query.append(
+                "ALTER USER {name:i} SET WORKLOAD_IDENTITY = (",
+                {
+                    "name": bp.full_name,
+                }
+            )
+
+            query.append(self._build_workload_identity_parameters(bp))
+            query.append_nl(")")
+
+            self.engine.execute_safe_ddl(query)
+
+            return True
+
+        return False
+
     def _check_user_role_grant(self, bp: UserBlueprint):
         user_role = self._get_user_role_ident(bp)
 

snowddl/settings.py

@@ -20,6 +20,7 @@ class SnowDDLSettings(BaseModelWithConfig):
     execute_resource_monitor: bool = False
     execute_outbound_share: bool = False
     refresh_user_passwords: bool = False
+    refresh_workload_identity: bool = False
     refresh_future_grants: bool = False
     refresh_stage_encryption: bool = False
     refresh_secrets: bool = False

snowddl/version.py

@@ -1 +1 @@
-__version__ = "0.57.1"
+__version__ = "0.58.0"

test/_config/step1/user.yaml

@@ -37,6 +37,14 @@ us002_us1:
   type: person
   password: YQYAtgqkbS5h
 
+us003_us1:
+  disabled: true
+  type: service
+  workload_identity:
+    type: OIDC
+    issuer: https://token.actions.githubusercontent.com
+    subject: repo:littleK0i/SnowDDL:ref:refs/heads/master
+
 np001_us1:
   disabled: true
   network_policy: np001_np1

test/_config/step2/user.yaml

@@ -18,6 +18,10 @@ us002_us1:
   type: legacy_service
   password: YQYAtgqkbS5h
 
+us003_us1:
+  disabled: true
+  type: person
+
 np001_us1:
   disabled: true
   network_policy: np001_np2

test/_config/step3/user.yaml

@@ -10,6 +10,14 @@ us002_us1:
     d4A3+Wp/pkTiYUh2GvjHTZrGViZXBPRjciP+6ktLMuXP4bW2DeS1xEYIUeYhxaNI
     IwIDAQAB
 
+us003_us1:
+  disabled: true
+  type: service
+  workload_identity:
+    type: OIDC
+    issuer: https://token.actions.githubusercontent.com
+    subject: repo:littleK0i/SnowDDL:ref:refs/heads/master
+
 np001_us1:
   disabled: true