feat(grants): add privilege enums for SPCS objects

docs/resources/grant.md

@@ -59,6 +59,21 @@ grants:
   - priv: MONITOR
     on: cortex search service somedb.someschema.someservice
     to: search_observability_role
+
+  # SPCS: USAGE on a compute pool
+  - priv: usage
+    on_compute_pool: ds_gpu_pool
+    to_role: data_engineer
+
+  # SPCS: READ on an image repository
+  - priv: read
+    on_image_repository: sandbox.spcs.vllm_repo
+    to_role: data_engineer
+
+  # SPCS: MONITOR on a service
+  - priv: monitor
+    on_service: sandbox.spcs.lora_service
+    to_role: data_engineer
 ```
 
 #### Future Grants
@@ -115,6 +130,11 @@ grant = Grant(priv="CREATE TABLE", on_schema="foo", to="somerole")
 
 # Table Privileges:
 grant = Grant(priv=["SELECT", "INSERT", "DELETE"], on_table="sometable", to="somerole")
+
+# Snowpark Container Services (SPCS) Privileges:
+grant = Grant(priv="USAGE", on_compute_pool="ds_gpu_pool", to="data_engineer")
+grant = Grant(priv="READ", on_image_repository="sandbox.spcs.vllm_repo", to="data_engineer")
+grant = Grant(priv="MONITOR", on_service="sandbox.spcs.lora_service", to="data_engineer")
 ```
 
 #### Future Grants
@@ -185,6 +205,9 @@ grant_on_all = Grant(
   - `"schema my_db.my_schema"` - for schema privileges
   - `"warehouse my_wh"` - for warehouse privileges
   - `"database my_db"` - for database privileges
+  - `"compute pool my_pool"` - for compute pool privileges
+  - `"image repository my_db.my_schema.my_repo"` - for image repository privileges
+  - `"service my_db.my_schema.my_service"` - for service privileges
   - `"future tables in schema my_schema"` - for future grants
   - `"all tables in database my_db"` - for grants on all existing objects
 

docs/snowflake-permissions.md

@@ -60,6 +60,14 @@ GRANT CREATE COMPUTE POOL ON ACCOUNT TO ROLE SNOWCAP_ADMIN;
 GRANT IMPORTED PRIVILEGES ON DATABASE SNOWFLAKE TO ROLE SNOWCAP_ADMIN;
 ```
 
+`CREATE COMPUTE POOL` above only covers creating new compute pools. Snowcap can also manage object-level grants on existing SPCS resources (see [Grant](resources/grant.md)):
+
+| Resource | Grantable privileges |
+|---|---|
+| Compute pool | `USAGE`, `MONITOR`, `MODIFY`, `OPERATE`, `OWNERSHIP` |
+| Image repository | `READ`, `WRITE`, `OWNERSHIP` |
+| Service | `USAGE`, `MONITOR`, `OPERATE`, `OWNERSHIP` |
+
 See [Optimizing Grant Fetching with ACCOUNT_USAGE](getting-started.md#optimizing-grant-fetching-with-account_usage) for when the ACCOUNT_USAGE optimization is worth enabling.
 
 ### Step 3: Create the service user

snowcap/privs.py

@@ -112,6 +112,15 @@ class DatabaseRolePriv(Priv):
     USAGE = "USAGE"
 
 
+class ComputePoolPriv(Priv):
+    ALL = "ALL"
+    MODIFY = "MODIFY"
+    MONITOR = "MONITOR"
+    OPERATE = "OPERATE"
+    OWNERSHIP = "OWNERSHIP"
+    USAGE = "USAGE"
+
+
 class CortexSearchServicePriv(Priv):
     ALL = "ALL"
     MONITOR = "MONITOR"
@@ -168,6 +177,13 @@ class IcebergTablePriv(Priv):
     UPDATE = "UPDATE"
 
 
+class ImageRepositoryPriv(Priv):
+    ALL = "ALL"
+    OWNERSHIP = "OWNERSHIP"
+    READ = "READ"
+    WRITE = "WRITE"
+
+
 class IntegrationPriv(Priv):
     ALL = "ALL"
     USAGE = "USAGE"
@@ -298,6 +314,14 @@ class SequencePriv(Priv):
     USAGE = "USAGE"
 
 
+class ServicePriv(Priv):
+    ALL = "ALL"
+    MONITOR = "MONITOR"
+    OPERATE = "OPERATE"
+    OWNERSHIP = "OWNERSHIP"
+    USAGE = "USAGE"
+
+
 class StagePriv(Priv):
     ALL = "ALL"
     OWNERSHIP = "OWNERSHIP"
@@ -391,7 +415,7 @@ class WarehousePriv(Priv):
     ResourceType.CATALOG_INTEGRATION: None,
     ResourceType.CLASS: None,
     ResourceType.COLUMN: None,
-    ResourceType.COMPUTE_POOL: None,
+    ResourceType.COMPUTE_POOL: ComputePoolPriv,
     ResourceType.CORTEX_SEARCH_SERVICE: CortexSearchServicePriv,
     ResourceType.DATABASE_ROLE: DatabaseRolePriv,
     ResourceType.DATABASE: DatabasePriv,
@@ -408,7 +432,7 @@ class WarehousePriv(Priv):
     ResourceType.GRANT: None,
     ResourceType.HYBRID_TABLE: None,
     ResourceType.ICEBERG_TABLE: IcebergTablePriv,
-    ResourceType.IMAGE_REPOSITORY: None,
+    ResourceType.IMAGE_REPOSITORY: ImageRepositoryPriv,
     ResourceType.INTEGRATION: IntegrationPriv,
     ResourceType.MATERIALIZED_VIEW: MaterializedViewPriv,
     ResourceType.NETWORK_POLICY: NetworkPolicyPriv,
@@ -428,7 +452,7 @@ class WarehousePriv(Priv):
     ResourceType.SECRET: SecretPriv,
     ResourceType.SECURITY_INTEGRATION: IntegrationPriv,
     ResourceType.SEQUENCE: SequencePriv,
-    ResourceType.SERVICE: None,
+    ResourceType.SERVICE: ServicePriv,
     ResourceType.SHARE: None,
     ResourceType.STAGE: StagePriv,
     ResourceType.STORAGE_INTEGRATION: IntegrationPriv,

tests/integration/test_grant_patterns.py

@@ -246,6 +246,112 @@ def test_no_drift_after_multi_priv_grant_apply(self, cursor, suffix, test_db, ma
         assert len(plan2) == 0, f"Expected no drift but got: {plan2}"
 
 
+class TestSPCSGrantsIntegration:
+    """Test that grants on compute pools, image repositories, and services work correctly."""
+
+    def test_grants_on_compute_pool_image_repository_and_service(
+        self, cursor, suffix, test_db, marked_for_cleanup
+    ):
+        """Test: USAGE on compute pool + ALL on image repository + ALL on service in one apply.
+
+        Compute pools are slow to provision, so this test creates one compute
+        pool, one image repository, and one service, then plans and applies
+        all three SPCS grants in a single blueprint cycle instead of one per
+        object type.
+        """
+        session = cursor.connection
+
+        role_name = f"SPCS_GRANT_ROLE_{suffix}"
+        role = res.Role(name=role_name)
+        cursor.execute(role.create_sql(if_not_exists=True))
+
+        schema_name = f"SPCS_SCHEMA_{suffix}"
+        schema = res.Schema(name=f"{test_db}.{schema_name}")
+        cursor.execute(schema.create_sql(if_not_exists=True))
+
+        pool_name = f"SPCS_POOL_{suffix}"
+        cursor.execute(
+            f"CREATE COMPUTE POOL IF NOT EXISTS {pool_name} "
+            "MIN_NODES = 1 MAX_NODES = 1 INSTANCE_FAMILY = CPU_X64_XS"
+        )
+
+        repo_name = f"SPCS_REPO_{suffix}"
+        repo_fqn = f"{test_db}.{schema_name}.{repo_name}"
+        cursor.execute(f"CREATE IMAGE REPOSITORY IF NOT EXISTS {repo_fqn}")
+
+        svc_name = f"SPCS_SVC_{suffix}"
+        svc_fqn = f"{test_db}.{schema_name}.{svc_name}"
+        cursor.execute(
+            f"""
+CREATE SERVICE IF NOT EXISTS {svc_fqn}
+  IN COMPUTE POOL {pool_name}
+  FROM SPECIFICATION $$
+spec:
+  container:
+  - name: main
+    image: /snowflake/images/snowflake_images/tutorial-image:latest
+  endpoint:
+  - name: apiendpoint
+    port: 8000
+    public: true
+$$
+  MIN_INSTANCES=1
+  MAX_INSTANCES=1
+"""
+        )
+        # Cleanup order: service and compute pool are billable and account-scoped
+        # (outside the DROP DATABASE safety net), so register them first - the
+        # cleanup loop drops in this append order, and an aborted loop must
+        # not leak the pool behind an earlier failure.
+        marked_for_cleanup.append(res.Service(name=svc_fqn, compute_pool=pool_name))
+        marked_for_cleanup.append(res.ComputePool(name=pool_name))
+        marked_for_cleanup.append(res.ImageRepository(name=repo_fqn))
+        marked_for_cleanup.append(schema)
+        marked_for_cleanup.append(role)
+
+        yaml_config = {
+            "grants": [
+                {"priv": "USAGE", "on_compute_pool": pool_name, "to_role": role_name},
+                {"priv": "ALL", "on_image_repository": repo_fqn, "to_role": role_name},
+                {"priv": "ALL", "on_service": svc_fqn, "to_role": role_name},
+            ],
+        }
+
+        bc = collect_blueprint_config(yaml_config)
+        blueprint = Blueprint.from_config(bc)
+
+        plan = blueprint.plan(session)
+        assert len(plan) == 3
+        assert all(isinstance(p, CreateResource) for p in plan)
+
+        blueprint.apply(session, plan)
+
+        # Verify USAGE grant on the compute pool
+        pool_grant = res.Grant(priv="USAGE", on_compute_pool=pool_name, to=role)
+        pool_data = safe_fetch(cursor, pool_grant.urn)
+        assert pool_data is not None, "USAGE grant on compute pool was not created"
+        assert pool_data["priv"] == "USAGE"
+
+        # Verify ALL grant on the image repository expands to exactly READ, WRITE
+        repo_grant = res.Grant(priv="ALL", on_image_repository=repo_fqn, to=role)
+        repo_data = safe_fetch(cursor, repo_grant.urn)
+        assert repo_data is not None, "ALL grant on image repository was not created"
+        assert repo_data["_privs"] == ["READ", "WRITE"]
+
+        # Verify ALL grant on the service expands to exactly MONITOR, OPERATE, USAGE
+        svc_grant = res.Grant(priv="ALL", on_service=svc_fqn, to=role)
+        svc_data = safe_fetch(cursor, svc_grant.urn)
+        assert svc_data is not None, "ALL grant on service was not created"
+        assert svc_data["_privs"] == ["MONITOR", "OPERATE", "USAGE"]
+
+        # Re-plan should show no changes
+        reset_cache()
+        bc2 = collect_blueprint_config(yaml_config)
+        blueprint2 = Blueprint.from_config(bc2)
+        plan2 = blueprint2.plan(session)
+        assert len(plan2) == 0, f"Expected no drift but got: {plan2}"
+
+
 class TestRoleGrantsIntegration:
     """Test that role grants work correctly in Snowflake."""
 

tests/test_grant.py

@@ -174,6 +174,55 @@ def test_grant_on_cortex_search_service():
     assert "MONITOR ON CORTEX SEARCH SERVICE" in monitor_grant.create_sql()
 
 
+def test_grant_on_compute_pool():
+    """USAGE on a COMPUTE POOL parses and renders correctly.
+
+    Compute pools are account-scoped, so grants target a bare name:
+        GRANT USAGE ON COMPUTE POOL <pool> TO ROLE r
+    """
+    grant = res.Grant(
+        priv="USAGE",
+        on_compute_pool="MY_POOL",
+        to="SOME_ROLE",
+    )
+    assert grant.on_type == ResourceType.COMPUTE_POOL
+    assert grant.on == "MY_POOL"
+    assert "GRANT USAGE ON COMPUTE POOL" in grant.create_sql()
+
+
+def test_grant_on_image_repository():
+    """READ on an IMAGE REPOSITORY parses and renders correctly.
+
+    Image repositories are schema-scoped, so grants target a fully
+    qualified name:
+        GRANT READ ON IMAGE REPOSITORY <db>.<schema>.<repo> TO ROLE r
+    """
+    grant = res.Grant(
+        priv="READ",
+        on_image_repository="DB.SPCS.REPO",
+        to="SOME_ROLE",
+    )
+    assert grant.on_type == ResourceType.IMAGE_REPOSITORY
+    assert grant.on == "DB.SPCS.REPO"
+    assert "GRANT READ ON IMAGE REPOSITORY DB.SPCS.REPO" in grant.create_sql()
+
+
+def test_grant_on_service():
+    """MONITOR on a SERVICE parses and renders correctly.
+
+    Services are schema-scoped, so grants target a fully qualified name:
+        GRANT MONITOR ON SERVICE <db>.<schema>.<svc> TO ROLE r
+    """
+    grant = res.Grant(
+        priv="MONITOR",
+        on_service="DB.SPCS.SVC",
+        to="SOME_ROLE",
+    )
+    assert grant.on_type == ResourceType.SERVICE
+    assert grant.on == "DB.SPCS.SVC"
+    assert "GRANT MONITOR ON SERVICE DB.SPCS.SVC" in grant.create_sql()
+
+
 def test_grant_database_role_to_database_role():
     database = res.Database(name="somedb")
     parent = res.DatabaseRole(name="parent", database=database)
@@ -328,6 +377,21 @@ def test_multi_priv_empty_list_raises_error(self):
         with pytest.raises(ValueError, match="at least one privilege"):
             res.Grant(priv=[], on_warehouse="somewh", to="somerole")
 
+    def test_all_priv_expands_on_compute_pool(self):
+        """Test: priv: ALL on a compute pool expands to the full non-ALL/non-OWNERSHIP list."""
+        grant = res.Grant(priv="ALL", on_compute_pool="MY_POOL", to="R")
+        assert grant._data._privs == ["MODIFY", "MONITOR", "OPERATE", "USAGE"]
+
+    def test_all_priv_expands_on_image_repository(self):
+        """Test: priv: ALL on an image repository expands to the full non-ALL/non-OWNERSHIP list."""
+        grant = res.Grant(priv="ALL", on_image_repository="D.S.REPO", to="R")
+        assert grant._data._privs == ["READ", "WRITE"]
+
+    def test_all_priv_expands_on_service(self):
+        """Test: priv: ALL on a service expands to the full non-ALL/non-OWNERSHIP list."""
+        grant = res.Grant(priv="ALL", on_service="D.S.SVC", to="R")
+        assert grant._data._privs == ["MONITOR", "OPERATE", "USAGE"]
+
     def test_single_priv_in_list_works(self):
         """Test: Single privilege in list works correctly."""
         grant = res.Grant(priv=["USAGE"], on_warehouse="somewh", to="somerole")