Missing ACLs (#1556)

CHANGELOG.md

@@ -17,6 +17,12 @@ Changes are grouped as follows
 - `Fixed` for any bug fixes.
 - `Security` in case of vulnerabilities.
 
+## [7.7.1] - 2023-12-20
+### Fixed
+- Missing legacy capability ACLs: `modelHostingAcl` and `genericsAcl`.  
+- The `IAMAPI.compare_capabilities` fails with a `AttributeError: 'UnknownAcl' object has no attribute '_capability_name'`
+  if the user has an unknwon ACL. This is now fixed by skipping comparison of unknown ACLs and issuing a warning.
+
 ## [7.7.0] - 2023-12-20
 ### Added
 - Support for `ViewProperty` types `SingleReverseDirectRelation` and `MultiReverseDirectRelation` in data modeling.

cognite/client/_api/iam.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import warnings
 from itertools import groupby
 from operator import itemgetter
 from typing import TYPE_CHECKING, Any, Dict, Sequence, Union
@@ -24,9 +25,11 @@
 from cognite.client.data_classes.capabilities import (
     AllScope,
     Capability,
+    LegacyCapability,
     ProjectCapability,
     ProjectCapabilityList,
     RawAcl,
+    UnknownAcl,
 )
 from cognite.client.data_classes.iam import TokenInspection
 from cognite.client.utils._identifier import IdentifierSequence
@@ -61,11 +64,20 @@ def _convert_capability_to_tuples(capabilities: ComparableCapability, project: s
         capabilities = [cap for grp in capabilities for cap in grp.capabilities or []]
     if isinstance(capabilities, Sequence):
         tpls: set[tuple] = set()
+        has_skipped = False
         for cap in capabilities:
             if isinstance(cap, dict):
                 cap = Capability.load(cap)
+            if isinstance(cap, UnknownAcl):
+                warnings.warn(f"Unknown capability {cap.capability_name} will be ignored in comparison")
+                has_skipped = True
+                continue
+            if isinstance(cap, LegacyCapability):
+                # Legacy capabilities are no longer in use, so they are safe to skip.
+                has_skipped = True
+                continue
             tpls.update(cap.as_tuples())  # type: ignore [union-attr]
-        if tpls:
+        if tpls or has_skipped:
             return tpls
         raise ValueError("No capabilities given")
     raise TypeError(
@@ -93,6 +105,10 @@ def compare_capabilities(
     ) -> list[Capability]:
         """Helper method to compare capabilities across two groups (of capabilities) to find which are missing from the first.
 
+        Note:
+            Capabilities that are no longer in use by the API will be ignored. These have names prefixed with `Legacy` and
+            all inherit from the base class `LegacyCapability`. If you want to check for these, you must do so manually.
+
         Args:
             existing_capabilities (ComparableCapability): List of existing capabilities.
             desired_capabilities (ComparableCapability): List of wanted capabilities to check against existing.

cognite/client/_version.py

@@ -1,4 +1,4 @@
 from __future__ import annotations
 
-__version__ = "7.7.0"
+__version__ = "7.7.1"
 __api_subversion__ = "V20220125"

cognite/client/data_classes/capabilities.py

@@ -2,7 +2,9 @@
 
 import enum
 import inspect
+import itertools
 import logging
+import warnings
 from abc import ABC
 from dataclasses import asdict, dataclass, field
 from itertools import product
@@ -176,6 +178,13 @@ def as_tuples(self) -> set[tuple]:
         )
 
 
+@dataclass
+class LegacyCapability(Capability, ABC):
+    """This is a base class for capabilities that are no longer in use by the API."""
+
+    ...
+
+
 class ProjectScope(ABC):
     name: ClassVar[str] = "projectScope"
 
@@ -245,15 +254,23 @@ def _infer_project(self, project: str | None = None) -> str:
 
     def as_tuples(self, project: str | None = None) -> set[tuple]:
         project = self._infer_project(project)
-        return set().union(
-            *(
-                proj_cap.capability.as_tuples()
-                for proj_cap in self
-                if isinstance(proj_cap.project_scope, AllProjectsScope)
+
+        output: set[tuple] = set()
+        for proj_cap in self:
+            cap = proj_cap.capability
+            if isinstance(cap, UnknownAcl):
+                warnings.warn(f"Unknown capability {cap.capability_name} will be ignored in comparison")
+                continue
+            if isinstance(cap, LegacyCapability):
+                # Legacy capabilities are no longer in use, so they are safe to skip.
+                continue
+            if (
+                isinstance(proj_cap.project_scope, AllProjectsScope)
                 or isinstance(proj_cap.project_scope, ProjectsScope)
                 and project in proj_cap.project_scope.projects
-            )
-        )
+            ):
+                output |= cap.as_tuples()
+        return output
 
 
 @dataclass(frozen=True)
@@ -1128,8 +1145,40 @@ class Scope:
         All = AllScope
 
 
+@dataclass
+class LegacyModelHostingAcl(LegacyCapability):
+    _capability_name = "modelHostingAcl"
+    actions: Sequence[Action]
+    scope: AllScope = field(default_factory=AllScope)
+
+    class Action(Capability.Action):
+        Read = "READ"
+        Write = "WRITE"
+
+    class Scope:
+        All = AllScope
+
+
+@dataclass
+class LegacyGenericsAcl(LegacyCapability):
+    _capability_name = "genericsAcl"
+    actions: Sequence[Action]
+    scope: AllScope
+
+    class Action(Capability.Action):
+        Read = "READ"
+        Write = "WRITE"
+
+    class Scope:
+        All = AllScope
+
+
 _CAPABILITY_CLASS_BY_NAME: MappingProxyType[str, type[Capability]] = MappingProxyType(
-    {c._capability_name: c for c in Capability.__subclasses__() if c is not UnknownAcl}
+    {
+        c._capability_name: c
+        for c in itertools.chain(Capability.__subclasses__(), LegacyCapability.__subclasses__())
+        if c not in (UnknownAcl, LegacyCapability)
+    }
 )
 # Give all Actions a better error message (instead of implementing __missing__ for all):
 for acl in _CAPABILITY_CLASS_BY_NAME.values():

pyproject.toml

@@ -1,7 +1,7 @@
 [tool.poetry]
 name = "cognite-sdk"
 
-version = "7.7.0"
+version = "7.7.1"
 description = "Cognite Python SDK"
 readme = "README.md"
 documentation = "https://cognite-sdk-python.readthedocs-hosted.com"

tests/tests_unit/test_data_classes/test_capabilities.py

@@ -5,7 +5,7 @@
 
 import pytest
 
-import cognite.client.data_classes.capabilities as capabilities_module  # noqa F401
+import cognite.client.data_classes.capabilities as capabilities_module  # F401
 from cognite.client.data_classes.capabilities import (
     AllProjectsScope,
     AllScope,
@@ -55,10 +55,12 @@ def all_acls():
         {"filesAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
         {"filesAcl": {"actions": ["READ", "WRITE"], "scope": {"datasetScope": {"ids": ["2332579", "372"]}}}},
         {"functionsAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
+        {"genericsAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
         {"groupsAcl": {"actions": ["LIST", "READ", "DELETE", "UPDATE", "CREATE"], "scope": {"all": {}}}},
         {"groupsAcl": {"actions": ["READ", "CREATE", "UPDATE", "DELETE"], "scope": {"currentuserscope": {}}}},
         {"hostedExtractorsAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
         {"labelsAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
+        {"modelHostingAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
         {"monitoringTasksAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
         {"notificationsAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
         {"pipelinesAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}},
@@ -391,6 +393,24 @@ def test_raw_acl_database_scope(self, cognite_client, extra_existing):
             assert RawAcl([RawAcl.Action.Read], RawAcl.Scope.Table({"db1": ["t3"]})) in missing
             assert RawAcl([RawAcl.Action.Write], RawAcl.Scope.Table({"db1": ["t1"]})) in missing
 
+    def test_unknown_existing_capability(self, cognite_client):
+        desired = [
+            Capability.load({"datasetsAcl": {"actions": ["READ"], "scope": {"all": {}}}}),
+        ]
+        unknown = Capability.load(
+            {"dataproductAcl": {"actions": ["UTILIZE"], "scope": {"components": {"ids": [1, 2, 3]}}}}
+        )
+
+        missing = cognite_client.iam.compare_capabilities(unknown, desired)
+        assert missing == desired
+
+    def test_legacy_capability(self, cognite_client):
+        legacy = [Capability.load({"modelHostingAcl": {"actions": ["READ", "WRITE"], "scope": {"all": {}}}})]
+        desired = [Capability.load({"modelHostingAcl": {"actions": ["READ"], "scope": {"all": {}}}})]
+
+        missing = cognite_client.iam.compare_capabilities(legacy, desired)
+        assert not missing
+
 
 @pytest.mark.parametrize(
     "dct",
@@ -429,6 +449,8 @@ def test_show_example_usage(capability):
     if capability is UnknownAcl:
         with pytest.raises(NotImplementedError):
             capability.show_example_usage()
+    elif capability is capabilities_module.LegacyCapability:
+        pytest.skip("LegacyCapability is abstract")
     else:
         cmd = capability.show_example_usage()[15:]  # TODO PY39: .removeprefix
         exec(f"{capability.__name__} = capabilities_module.{capability.__name__}")