switch to jmespath

icekube/models/pod.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 
-import json
 from functools import cached_property
 from itertools import product
 from pathlib import Path
 from typing import Any, Dict, List, Optional, cast
 
+import jmespath
 from icekube.models.base import RELATIONSHIP, Resource
 from icekube.models.node import Node
 from icekube.models.secret import Secret
@@ -64,7 +64,7 @@ class Pod(Resource):
     @computed_field  # type: ignore
     @cached_property
     def service_account(self) -> Optional[ServiceAccount]:
-        sa = self.data.get("spec", {}).get("serviceAccountName")
+        sa = jmespath.search("spec.serviceAccountName", self.data)
 
         if sa:
             return ServiceAccount(name=sa, namespace=self.namespace)
@@ -74,7 +74,7 @@ def service_account(self) -> Optional[ServiceAccount]:
     @computed_field  # type: ignore
     @cached_property
     def node(self) -> Optional[Node]:
-        node = self.data.get("spec", {}).get("nodeName")
+        node = jmespath.search("spec.nodeName", self.data)
 
         if node:
             return Node(name=node)
@@ -85,8 +85,7 @@ def node(self) -> Optional[Node]:
     @cached_property
     def containers(self) -> List[Dict[str, Any]]:
         return cast(
-            List[Dict[str, Any]],
-            self.data.get("spec", {}).get("containers", []),
+            List[Dict[str, Any]], jmespath.search("spec.containers[]", self.data) or []
         )
 
     @computed_field  # type: ignore
@@ -95,9 +94,7 @@ def capabilities(self) -> List[str]:
         capabilities = set()
 
         for container in self.containers:
-            security_context = container.get("securityContext") or {}
-            caps = security_context.get("capabilities") or {}
-            addl = caps.get("add") or []
+            addl = jmespath.search("securityContext.capabilities.add", container) or []
             addl = [x.upper() for x in addl]
             add = set(addl)
 
@@ -112,33 +109,26 @@ def capabilities(self) -> List[str]:
     @computed_field  # type: ignore
     @cached_property
     def privileged(self) -> bool:
-        containers = self.data.get("spec", {}).get("containers", [])
-        privileged = False
-        for container in containers:
-            context = container.get("securityContext") or {}
-
-            if context.get("privileged", False):
-                privileged = True
-
-        return privileged
+        privileged = (
+            jmespath.search("spec.containers[].securityContext.privileged", self.data)
+            or []
+        )
+        return any(privileged)
 
     @computed_field  # type: ignore
     @cached_property
     def host_path_volumes(self) -> List[str]:
-        volumes = self.data.get("spec", {}).get("volumes") or []
-        host_volumes = [x for x in volumes if "hostPath" in x and x["hostPath"]]
-
-        return [x["hostPath"]["path"] for x in host_volumes]
+        return jmespath.search("spec.volumes[].hostPath.path", self.data) or []
 
     @computed_field  # type: ignore
     @cached_property
     def hostPID(self) -> bool:
-        return self.data.get("spec", {}).get("hostPID") or False
+        return jmespath.search("spec.hostPID", self.data) or False
 
     @computed_field  # type: ignore
     @cached_property
     def hostNetwork(self) -> bool:
-        return self.data.get("spec", {}).get("hostNetwork") or False
+        return jmespath.search("spec.hostNetwork", self.data) or False
 
     @property
     def dangerous_host_path(self) -> bool:
@@ -180,27 +170,13 @@ def dangerous_host_path(self) -> bool:
 
     @property
     def mounted_secrets(self) -> List[str]:
-        if self.raw:
-            data = json.loads(self.raw)
-        else:
-            return []
-
-        secrets = []
-
-        volumes = data.get("spec", {}).get("volumes") or []
-
-        for volume in volumes:
-            if volume.get("secret"):
-                secrets.append(volume["secret"]["secretName"])
-
-        for container in data.get("spec", {}).get("containers") or []:
-            if not container.get("env"):
-                continue
-            for env in container["env"]:
-                try:
-                    secrets.append(env["valueFrom"]["secretKeyRef"]["name"])
-                except (KeyError, TypeError):
-                    pass
+        secrets = jmespath.search("spec.volumes[].secret.secretName", self.data) or []
+        secrets += (
+            jmespath.search(
+                "spec.containers[].env[].valueFrom.secretKeyRef.name", self.data
+            )
+            or []
+        )
 
         return secrets
 

icekube/relationships.py

@@ -22,22 +22,22 @@ class Relationship:
     WITHIN_NAMESPACE: ClassVar[str] = "WITHIN_NAMESPACE"
 
     GRANTS_PODS_CREATE: ClassVar[str] = "GRANTS_PODS_CREATE"
-    GRANTS_REPLICATIONCONTROLLERS_CREATE: ClassVar[
-        str
-    ] = "GRANTS_REPLICATIONCONTROLLERS_CREATE"
+    GRANTS_REPLICATIONCONTROLLERS_CREATE: ClassVar[str] = (
+        "GRANTS_REPLICATIONCONTROLLERS_CREATE"
+    )
     GRANTS_DAEMONSETS_CREATE: ClassVar[str] = "GRANTS_DAEMONSETS_CREATE"
     GRANTS_DEPLOYMENTS_CREATE: ClassVar[str] = "GRANTS_DEPLOYMENTS_CREATE"
     GRANTS_REPLICASETS_CREATE: ClassVar[str] = "GRANTS_REPLICASETS_CREATE"
     GRANTS_STATEFULSETS_CREATE: ClassVar[str] = "GRANTS_STATEFULSETS_CREATE"
     GRANTS_CRONJOBS_CREATE: ClassVar[str] = "GRANTS_CRONJOBS_CREATE"
     GRANTS_JOBS_CREATE: ClassVar[str] = "GRANTS_JOBS_CREATE"
 
-    GRANTS_AZUREPODIDENTITYEXCEPTIONS_CREATE: ClassVar[
-        str
-    ] = "GRANTS_AZUREPODIDENTITYEXCEPTIONS_CREATE"
-    GRANTS_CERTIFICATESIGNINGREQUESTS_CREATE: ClassVar[
-        str
-    ] = "GRANTS_CERTIFICATESIGNINGREQUESTS_CREATE"
+    GRANTS_AZUREPODIDENTITYEXCEPTIONS_CREATE: ClassVar[str] = (
+        "GRANTS_AZUREPODIDENTITYEXCEPTIONS_CREATE"
+    )
+    GRANTS_CERTIFICATESIGNINGREQUESTS_CREATE: ClassVar[str] = (
+        "GRANTS_CERTIFICATESIGNINGREQUESTS_CREATE"
+    )
     GRANTS_PROXY_CREATE: ClassVar[str] = "GRANTS_PROXY_CREATE"
 
     GRANTS_GET: ClassVar[str] = "GRANTS_GET"