"""Inventory from the Kubernetes API: what is actually running, and which workloads run it.""" import json import logging import os import re import ssl import urllib.error import urllib.parse import urllib.request log = logging.getLogger("vuln-scanner.inventory") _ID_PREFIXES = ("docker-pullable://", "docker://", "containerd://", "cri-o://") _DIGEST = re.compile(r"sha256:[0-9a-f]{64}") _DOCKER_ALIASES = {"index.docker.io", "registry-1.docker.io", "docker.io"} _CONTAINER_FIELDS = ( ("containers", "containerStatuses"), ("initContainers", "initContainerStatuses"), ("ephemeralContainers", "ephemeralContainerStatuses"), ) _VMI_PATH = "/apis/kubevirt.io/v1/virtualmachineinstances" PATHS = { "pods": "/api/v1/pods", "replicasets": "/apis/apps/v1/replicasets", "jobs": "/apis/batch/v1/jobs", "cronjobs": "/apis/batch/v1/cronjobs", "nodes": "/api/v1/nodes", } class InventoryError(Exception): """The API could not be listed; the previous inventory stands.""" class NotFound(InventoryError): pass class KubeClient: """Paginated GET list against the API using the pod's service account.""" def __init__(self, base_url=None, token_file=None, ca_file=None): sa = "/var/run/secrets/kubernetes.io/serviceaccount" if base_url is None: host = os.environ.get("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc") port = os.environ.get("KUBERNETES_SERVICE_PORT", "443") host = "[" + host + "]" if ":" in host else host base_url = "https://" + host + ":" + port self.base_url = base_url.rstrip("/") self.token_file = token_file or os.path.join(sa, "token") self.ca_file = ca_file or os.path.join(sa, "ca.crt") def _get_json(self, path, params): url = self.base_url + path + "?" + urllib.parse.urlencode(params) with open(self.token_file, encoding="utf-8") as handle: # projected tokens rotate token = handle.read().strip() request = urllib.request.Request(url, headers={"Authorization": "Bearer " + token}) context = ssl.create_default_context(cafile=self.ca_file) if url.startswith("https") else None try: with urllib.request.urlopen(request, timeout=30, context=context) as response: return json.loads(response.read()) except urllib.error.HTTPError as exc: if exc.code == 404: raise NotFound(path) from None raise InventoryError(path + ": HTTP " + str(exc.code)) from None except (urllib.error.URLError, OSError, ValueError) as exc: raise InventoryError(path + ": " + type(exc).__name__) from None def list_all(self, path): items = [] params = {"limit": "500"} while True: page = self._get_json(path, params) items.extend(page.get("items") or []) token = (page.get("metadata") or {}).get("continue") if not token: return items params = {"limit": "500", "continue": token} def collect(client): """Every list the inventory needs. A VMI API that 404s means no KubeVirt, hence no VMs.""" lists = {name: client.list_all(path) for name, path in PATHS.items()} try: lists["vmis"] = client.list_all(_VMI_PATH) except NotFound: lists["vmis"] = [] return lists def normalise_reference(reference): """(registry/repository, tag or None, digest or None) for an image reference.""" text = reference.strip() for prefix in _ID_PREFIXES: if text.startswith(prefix): text = text[len(prefix):] digest = None if "@" in text: text, _, raw = text.partition("@") digest = raw if _DIGEST.fullmatch(raw) else None tag = None tail = text.rsplit("/", 1)[-1] if ":" in tail: text, _, tag = text.rpartition(":") first, slash, rest = text.partition("/") if slash and ("." in first or ":" in first or first == "localhost"): registry, repository = first.lower(), rest else: registry, repository = "docker.io", text if registry in _DOCKER_ALIASES: registry = "docker.io" if "/" not in repository: repository = "library/" + repository return registry + "/" + repository, tag, digest def digest_from_image_id(image_id): """The repo digest in a container status imageID, or None. A bare `sha256:...` with no repository is a local config id, not a pullable digest, so it is treated as absent (the target is then by-tag) rather than scanned as something it is not. """ text = image_id or "" for prefix in _ID_PREFIXES: if text.startswith(prefix): text = text[len(prefix):] if "@" not in text: return None candidate = text.rsplit("@", 1)[1] return candidate if _DIGEST.fullmatch(candidate) else None def _controller(obj): references = (obj.get("metadata") or {}).get("ownerReferences") or [] for ref in references: if ref.get("controller"): return ref return references[0] if references else None def resolve_owner(pod, replicasets, jobs): """The top-level workload (kind, name) a pod belongs to.""" meta = pod.get("metadata") or {} namespace = meta.get("namespace", "") owner = _controller(pod) if owner is None: return "Pod", meta.get("name", "") kind, name = owner.get("kind", ""), owner.get("name", "") index = {"ReplicaSet": replicasets, "Job": jobs}.get(kind) if index is not None: parent = index.get((namespace, name)) upstream = _controller(parent) if parent else None if upstream is not None: return upstream.get("kind", kind), upstream.get("name", name) return kind, name def _index(items): return {((i.get("metadata") or {}).get("namespace", ""), (i.get("metadata") or {}).get("name", "")): i for i in items} class _Builder: def __init__(self): self.targets = {} def add(self, target_id, image, digest, ref, workload, container): target = self.targets.get(target_id) if target is None: target = self.targets[target_id] = { "id": target_id, "type": "image", "image": image, "digest": digest, "ref": ref, "reason": None, "workloads": {}, } if digest and not target["digest"]: target["digest"] = digest target["workloads"].setdefault(workload, set()).add(container) def add_image(self, spec_image, image_id, workload, container): image, tag, spec_digest = normalise_reference(spec_image) digest = digest_from_image_id(image_id) if digest: self.add(image + "@" + digest, image, digest, image + "@" + digest, workload, container) else: ref = image + "@" + spec_digest if spec_digest else image + ":" + (tag or "latest") target_id = image + ":" + tag if tag else (ref if spec_digest else image + ":latest") self.add(target_id, image, None, ref, workload, container) def add_uncovered(self, target_id, kind, reason, namespace, name, node_kind): self.targets[target_id] = { "id": target_id, "type": kind, "image": "", "digest": None, "ref": None, "reason": reason, "workloads": {(namespace, node_kind, name): set()}, } def finish(self): result = [] for target in sorted(self.targets.values(), key=lambda t: t["id"]): target["workloads"] = [ {"namespace": ns, "kind": kind, "name": name, "containers": sorted(containers)} for (ns, kind, name), containers in sorted(target["workloads"].items()) ] result.append(target) return result def build(lists): """Resolve raw API lists into target dicts.""" replicasets = _index(lists.get("replicasets") or []) jobs = _index(lists.get("jobs") or []) builder = _Builder() running = set() for pod in lists.get("pods") or []: status = pod.get("status") or {} # A Succeeded/Failed pod runs nothing; counting it would let a stale evicted pod pin a digest. if status.get("phase") in ("Succeeded", "Failed"): continue namespace = (pod.get("metadata") or {}).get("namespace", "") kind, name = resolve_owner(pod, replicasets, jobs) workload = (namespace, kind, name) if kind == "CronJob": running.add((namespace, name)) spec = pod.get("spec") or {} for spec_field, status_field in _CONTAINER_FIELDS: statuses = {s.get("name"): s for s in status.get(status_field) or []} for container in spec.get(spec_field) or []: image_id = (statuses.get(container.get("name")) or {}).get("imageID", "") builder.add_image(container.get("image", ""), image_id, workload, container.get("name", "")) for cronjob in lists.get("cronjobs") or []: meta = cronjob.get("metadata") or {} namespace, name = meta.get("namespace", ""), meta.get("name", "") if (namespace, name) in running: continue job_spec = ((cronjob.get("spec") or {}).get("jobTemplate") or {}).get("spec") or {} pod_spec = (job_spec.get("template") or {}).get("spec") or {} for field in ("containers", "initContainers"): for container in pod_spec.get(field) or []: builder.add_image(container.get("image", ""), "", (namespace, "CronJob", name), container.get("name", "")) for vmi in lists.get("vmis") or []: meta = vmi.get("metadata") or {} namespace, name = meta.get("namespace", ""), meta.get("name", "") builder.add_uncovered("vm/" + namespace + "/" + name, "vm", "guest OS packages are not assessed by this scanner", namespace, name, "VirtualMachineInstance") for node in lists.get("nodes") or []: name = (node.get("metadata") or {}).get("name", "") builder.add_uncovered("node/" + name, "node", "node OS packages and the k3s binary are not assessed by this scanner", "", name, "Node") return builder.finish()