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

@@ -13,13 +13,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(
@@ -91,19 +99,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:
@@ -206,12 +219,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 icekube.relationships import Relationship
 from pydantic import root_validator
 from pydantic.fields import Field
@@ -19,6 +18,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"])
@@ -48,8 +49,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),
  ),
@@ -63,6 +63,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):
@@ -88,11 +92,16 @@ def relationships(
  (subject, Relationship.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, Relationship.HAS_CSR_APPROVAL, get_cluster_object()),
+ (self, Relationship.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 icekube.relationships import Relationship
 from pydantic import root_validator
 
@@ -67,14 +66,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"),
  )
@@ -87,7 +86,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
 
@@ -258,7 +257,10 @@ def relationships(
  (
  self,
  Relationship.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

@@ -90,7 +90,7 @@ def affected_resource_query(
  "name": namespace,
  }
  else:
- query_filter = {"kind": "Cluster"}
+ query_filter = {"apiVersion": "N/A", "kind": "Cluster"}
  yield (
  Relationship.generate_grant("CREATE", resource),
  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

@@ -18,6 +18,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,14 +4,14 @@
 from typing import Any, Dict, List, cast
 
 from icekube.models.base import RELATIONSHIP, Resource
-from icekube.neo4j import mock
 from icekube.relationships import Relationship
 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):
@@ -45,8 +45,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),
  )