adds checks for apiVersion for resources

icekube/icekube.py

@@ -44,7 +44,7 @@ def enumerate_resource_kind(
         ignore = []
 
     with get_driver().session() as session:
-        cluster = Cluster(name=context_name(), version=kube_version())
+        cluster = Cluster(apiVersion="N/A", name=context_name(), version=kube_version())
         cmd, kwargs = create(cluster)
         session.run(cmd, **kwargs)
 

icekube/models/__init__.py

@@ -1,52 +1,33 @@
-from typing import List, Type
-
+from icekube.models import (
+    clusterrole,
+    clusterrolebinding,
+    group,
+    pod,
+    role,
+    rolebinding,
+    secret,
+    securitycontextconstraints,
+    serviceaccount,
+    user,
+)
 from icekube.models.api_resource import APIResource
 from icekube.models.base import Resource
 from icekube.models.cluster import Cluster
-from icekube.models.clusterrole import ClusterRole
-from icekube.models.clusterrolebinding import ClusterRoleBinding
-from icekube.models.group import Group
-from icekube.models.namespace import Namespace
-from icekube.models.pod import Pod
-from icekube.models.role import Role
-from icekube.models.rolebinding import RoleBinding
-from icekube.models.secret import Secret
-from icekube.models.securitycontextconstraints import (
-    SecurityContextConstraints,
-)
-from icekube.models.serviceaccount import ServiceAccount
 from icekube.models.signer import Signer
-from icekube.models.user import User
-
-enumerate_resource_kinds: List[Type[Resource]] = [
-    ClusterRole,
-    ClusterRoleBinding,
-    Namespace,
-    Pod,
-    Role,
-    RoleBinding,
-    Secret,
-    SecurityContextConstraints,
-    ServiceAccount,
-]
-
-
-# plurals: Dict[str, Type[Resource]] = {x.plural: x for x in enumerate_resource_kinds}
-
 
 __all__ = [
     "APIResource",
     "Cluster",
-    "ClusterRole",
-    "ClusterRoleBinding",
-    "Group",
-    "Namespace",
-    "Pod",
-    "Role",
-    "RoleBinding",
-    "Secret",
-    "SecurityContextConstraints",
-    "ServiceAccount",
     "Signer",
-    "User",
+    "Resource",
+    "clusterrole",
+    "clusterrolebinding",
+    "group",
+    "pod",
+    "role",
+    "rolebinding",
+    "secret",
+    "securitycontextconstraints",
+    "serviceaccount",
+    "user",
 ]

icekube/models/base.py

@@ -12,13 +12,21 @@
 logger = logging.getLogger(__name__)
 
 
+def api_group(api_version: str) -> str:
+    if "/" in api_version:
+        return api_version.split("/")[0]
+    # When the base APIGroup is ""
+    return ""
+
+
 class Resource(BaseModel):
     apiVersion: str = Field(default=...)
     kind: str = Field(default=...)
     name: str = Field(default=...)
     plural: str = Field(default=...)
     namespace: Optional[str] = Field(default=None)
     raw: Optional[str] = Field(default=None)
+    supported_api_groups: List[str] = Field(default_factory=list)
 
     def __new__(cls, **kwargs):
         kind_class = cls.get_kind_class(
@@ -90,19 +98,24 @@ def get_value(field):
 
     @classmethod
     def get_kind_class(cls, apiVersion: str, kind: str) -> Type[Resource]:
-        subclasses = {x.__name__: x for x in cls.__subclasses__()}
-        try:
-            return subclasses[kind]
-        except KeyError:
-            return cls
+        for subclass in cls.__subclasses__():
+            if subclass.__name__ != kind:
+                continue
+
+            supported = subclass.model_fields["supported_api_groups"].default
+            if not isinstance(supported, list):
+                continue
+
+            if api_group(apiVersion) not in supported:
+                continue
+
+            return subclass
+
+        return cls
 
     @property
     def api_group(self) -> str:
-        if "/" in self.apiVersion:
-            return self.apiVersion.split("/")[0]
-        else:
-            # When the base APIGroup is ""
-            return ""
+        return api_group(self.apiVersion)
 
     @property
     def resource_definition_name(self) -> str:
@@ -205,12 +218,10 @@ def relationships(
         logger.debug(
             f"Generating {'initial' if initial else 'second'} set of relationships",
         )
-        from icekube.neo4j import mock
-
         relationships: List[RELATIONSHIP] = []
 
         if self.namespace is not None:
-            ns = mock(Resource, name=self.namespace, kind="Namespace")
+            ns = Resource(name=self.namespace, kind="Namespace")
             relationships += [
                 (
                     self,

icekube/models/cluster.py

@@ -8,29 +8,26 @@ class Cluster(Resource):
     kind: str = "Cluster"
     apiVersion: str = "N/A"
     plural: str = "clusters"
+    supported_api_groups: List[str] = ["N"]
 
     def __repr__(self) -> str:
         return f"Cluster(name='{self.name}', version='{self.version}')"
 
     @property
-    def unique_identifiers(self) -> Dict[str, str]:
+    def db_labels(self) -> Dict[str, str]:
         return {
-            "name": self.name,
-            "kind": self.kind,
-            "apiVersion": self.apiVersion,
+            **self.unique_identifiers,
+            "plural": self.plural,
+            "version": self.version,
         }
 
-    @property
-    def db_labels(self) -> Dict[str, str]:
-        return {**self.unique_identifiers, "version": self.version}
-
     def relationships(
         self,
         initial: bool = True,
     ) -> List[RELATIONSHIP]:
         relationships = super().relationships()
 
-        query = "MATCH (src) WHERE NOT src:Cluster "
+        query = "MATCH (src) WHERE NOT src.apiVersion = 'N/A' "
 
         relationships += [((query, {}), "WITHIN_CLUSTER", self)]
 

icekube/models/clusterrole.py

@@ -11,6 +11,10 @@
 
 class ClusterRole(Resource):
     rules: List[PolicyRule] = Field(default_factory=list)
+    supported_api_groups: List[str] = [
+        "rbac.authorization.k8s.io",
+        "authorization.openshift.io",
+    ]
 
     @root_validator(pre=True)
     def inject_rules(cls, values):

icekube/models/clusterrolebinding.py

@@ -9,7 +9,6 @@
 from icekube.models.role import Role
 from icekube.models.serviceaccount import ServiceAccount
 from icekube.models.user import User
-from icekube.neo4j import find_or_mock, get_cluster_object, mock
 from pydantic import root_validator
 from pydantic.fields import Field
 
@@ -18,6 +17,8 @@ def get_role(
     role_ref: Dict[str, Any],
     namespace: Optional[str] = None,
 ) -> Union[ClusterRole, Role]:
+    from icekube.neo4j import find_or_mock
+
     role_ref["kind"] = role_ref.get("kind", "ClusterRole")
     if role_ref["kind"] == "ClusterRole":
         return find_or_mock(ClusterRole, name=role_ref["name"])
@@ -47,8 +48,7 @@ def get_subjects(
             results.append(Group(name=subject["name"]))
         elif subject["kind"] == "ServiceAccount":
             results.append(
-                mock(
-                    ServiceAccount,
+                ServiceAccount(
                     name=subject["name"],
                     namespace=subject.get("namespace", namespace),
                 ),
@@ -62,6 +62,10 @@ def get_subjects(
 class ClusterRoleBinding(Resource):
     role: Union[ClusterRole, Role]
     subjects: List[Union[ServiceAccount, User, Group]] = Field(default_factory=list)
+    supported_api_groups: List[str] = [
+        "rbac.authorization.k8s.io",
+        "authorization.openshift.io",
+    ]
 
     @root_validator(pre=True)
     def inject_role_and_subjects(cls, values):
@@ -85,12 +89,15 @@ def relationships(
         relationships += [(self, "GRANTS_PERMISSION", self.role)]
         relationships += [(subject, "BOUND_TO", self) for subject in self.subjects]
 
+        cluster_query = (
+            "MATCH ({prefix}) WHERE {prefix}.kind =~ ${prefix}_kind ",
+            {"apiVersion": "N/A", "kind": "Cluster"},
+        )
+
         if not initial:
             for role_rule in self.role.rules:
                 if role_rule.contains_csr_approval:
-                    relationships.append(
-                        (self, "HAS_CSR_APPROVAL", get_cluster_object()),
-                    )
+                    relationships.append((self, "HAS_CSR_APPROVAL", cluster_query))
                 for relationship, resource in role_rule.affected_resource_query():
                     relationships.append((self, relationship, resource))
 

icekube/models/group.py

@@ -1,13 +1,10 @@
 from __future__ import annotations
 
-from typing import Dict
+from typing import List
 
 from icekube.models.base import Resource
 
 
 class Group(Resource):
     plural: str = "groups"
-
-    @property
-    def unique_identifiers(self) -> Dict[str, str]:
-        return {**super().unique_identifiers, "plural": self.plural}
+    supported_api_groups: List[str] = ["", "user.openshift.io"]

icekube/models/node.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+from typing import List
+
 from icekube.models.base import Resource
 
 
 class Node(Resource):
-    ...
+    supported_api_groups: List[str] = [""]

icekube/models/pod.py

@@ -9,7 +9,6 @@
 from icekube.models.node import Node
 from icekube.models.secret import Secret
 from icekube.models.serviceaccount import ServiceAccount
-from icekube.neo4j import mock
 from pydantic import root_validator
 
 CAPABILITIES = [
@@ -66,14 +65,14 @@ class Pod(Resource):
     privileged: bool
     hostPID: bool
     hostNetwork: bool
+    supported_api_groups: List[str] = [""]
 
     @root_validator(pre=True)
     def inject_service_account(cls, values):
         data = json.loads(values.get("raw", "{}"))
         sa = data.get("spec", {}).get("serviceAccountName")
         if sa:
-            values["service_account"] = mock(
-                ServiceAccount,
+            values["service_account"] = ServiceAccount(
                 name=sa,
                 namespace=values.get("namespace"),
             )
@@ -86,7 +85,7 @@ def inject_node(cls, values):
         data = json.loads(values.get("raw", "{}"))
         node = data.get("spec", {}).get("nodeName")
         if node:
-            values["node"] = mock(Node, name=node)
+            values["node"] = Node(name=node)
         else:
             values["node"] = None
 
@@ -257,7 +256,10 @@ def relationships(
                 (
                     self,
                     "MOUNTS_SECRET",
-                    mock(Secret, namespace=cast(str, self.namespace), name=secret),
+                    Secret(  # type: ignore
+                        namespace=cast(str, self.namespace),
+                        name=secret,
+                    ),
                 ),
             ]
 

icekube/models/policyrule.py

@@ -89,7 +89,7 @@ def affected_resource_query(
                         "name": namespace,
                     }
                 else:
-                    query_filter = {"kind": "Cluster"}
+                    query_filter = {"apiVersion": "N/A", "kind": "Cluster"}
                     yield (
                         f"GRANTS_{resource}_CREATE".upper().replace("-", "_"),
                         generate_query(query_filter),

icekube/models/role.py

@@ -11,6 +11,10 @@
 
 class Role(Resource):
     rules: List[PolicyRule] = Field(default_factory=list)
+    supported_api_groups: List[str] = [
+        "rbac.authorization.k8s.io",
+        "authorization.openshift.io",
+    ]
 
     @root_validator(pre=True)
     def inject_role(cls, values):

icekube/models/rolebinding.py

@@ -17,6 +17,10 @@
 class RoleBinding(Resource):
     role: Union[ClusterRole, Role]
     subjects: List[Union[ServiceAccount, User, Group]] = Field(default_factory=list)
+    supported_api_groups: List[str] = [
+        "rbac.authorization.k8s.io",
+        "authorization.openshift.io",
+    ]
 
     @root_validator(pre=True)
     def inject_role_and_subjects(cls, values):

icekube/models/secret.py

@@ -4,13 +4,13 @@
 from typing import Any, Dict, List, cast
 
 from icekube.models.base import RELATIONSHIP, Resource
-from icekube.neo4j import mock
 from pydantic import root_validator
 
 
 class Secret(Resource):
     secret_type: str
     annotations: Dict[str, Any]
+    supported_api_groups: List[str] = [""]
 
     @root_validator(pre=True)
     def remove_secret_data(cls, values):
@@ -44,8 +44,7 @@ def relationships(self, initial: bool = True) -> List[RELATIONSHIP]:
 
             sa = self.annotations.get("kubernetes.io/service-account.name")
             if sa:
-                account = mock(
-                    ServiceAccount,
+                account = ServiceAccount(
                     name=sa,
                     namespace=cast(str, self.namespace),
                 )