community.okd/plugins/module_utils/openshift_images_common.py

#!/usr/bin/env python

from __future__ import absolute_import, division, print_function

__metaclass__ = type

from datetime import datetime
from ansible_collections.community.okd.plugins.module_utils.openshift_docker_image import (
    parse_docker_image_ref,
)
from ansible.module_utils.six import iteritems


def get_image_blobs(image):
    blobs = [layer["image"] for layer in image["dockerImageLayers"] if "image" in layer]
    docker_image_metadata = image.get("dockerImageMetadata")
    if not docker_image_metadata:
        return blobs, "failed to read metadata for image %s" % image["metadata"]["name"]
    media_type_manifest = (
        "application/vnd.docker.distribution.manifest.v2+json",
        "application/vnd.oci.image.manifest.v1+json",
    )
    media_type_has_config = image["dockerImageManifestMediaType"] in media_type_manifest
    docker_image_id = docker_image_metadata.get("Id")
    if media_type_has_config and docker_image_id and len(docker_image_id) > 0:
        blobs.append(docker_image_id)
    return blobs, None


def is_created_after(creation_timestamp, max_creation_timestamp):
    if not max_creation_timestamp:
        return False
    creationTimestamp = datetime.strptime(creation_timestamp, "%Y-%m-%dT%H:%M:%SZ")
    return creationTimestamp > max_creation_timestamp


def is_too_young_object(obj, max_creation_timestamp):
    return is_created_after(
        obj["metadata"]["creationTimestamp"], max_creation_timestamp
    )


class OpenShiftAnalyzeImageStream(object):
    def __init__(self, ignore_invalid_refs, max_creation_timestamp, module):
        self.max_creationTimestamp = max_creation_timestamp
        self.used_tags = {}
        self.used_images = {}
        self.ignore_invalid_refs = ignore_invalid_refs
        self.module = module

    def analyze_reference_image(self, image, referrer):
        result, error = parse_docker_image_ref(image, self.module)
        if error:
            return error

        if not result["hostname"] or not result["namespace"]:
            # image reference does not match hostname/namespace/name pattern - skipping
            return None

        if not result["digest"]:
            # Attempt to dereference istag. Since we cannot be sure whether the reference refers to the
            # integrated registry or not, we ignore the host part completely. As a consequence, we may keep
            # image otherwise sentenced for a removal just because its pull spec accidentally matches one of
            # our imagestreamtags.

            # set the tag if empty
            if result["tag"] == "":
                result["tag"] = "latest"
            key = "%s/%s:%s" % (result["namespace"], result["name"], result["tag"])
            if key not in self.used_tags:
                self.used_tags[key] = []
            self.used_tags[key].append(referrer)
        else:
            key = "%s/%s@%s" % (result["namespace"], result["name"], result["digest"])
            if key not in self.used_images:
                self.used_images[key] = []
            self.used_images[key].append(referrer)

    def analyze_refs_from_pod_spec(self, podSpec, referrer):
        for container in podSpec.get("initContainers", []) + podSpec.get(
            "containers", []
        ):
            image = container.get("image")
            if len(image.strip()) == 0:
                # Ignoring container because it has no reference to image
                continue
            err = self.analyze_reference_image(image, referrer)
            if err:
                return err
        return None

    def analyze_refs_from_pods(self, pods):
        for pod in pods:
            # A pod is only *excluded* from being added to the graph if its phase is not
            # pending or running. Additionally, it has to be at least as old as the minimum
            # age threshold defined by the algorithm.
            too_young = is_too_young_object(pod, self.max_creationTimestamp)
            if pod["status"]["phase"] not in ("Running", "Pending") and too_young:
                continue
            referrer = {
                "kind": pod["kind"],
                "namespace": pod["metadata"]["namespace"],
                "name": pod["metadata"]["name"],
            }
            err = self.analyze_refs_from_pod_spec(pod["spec"], referrer)
            if err:
                return err
        return None

    def analyze_refs_pod_creators(self, resources):
        keys = (
            "ReplicationController",
            "DeploymentConfig",
            "DaemonSet",
            "Deployment",
            "ReplicaSet",
            "StatefulSet",
            "Job",
            "CronJob",
        )

        for k, objects in iteritems(resources):
            if k not in keys:
                continue
            for obj in objects:
                if k == "CronJob":
                    spec = obj["spec"]["jobTemplate"]["spec"]["template"]["spec"]
                else:
                    spec = obj["spec"]["template"]["spec"]
                referrer = {
                    "kind": obj["kind"],
                    "namespace": obj["metadata"]["namespace"],
                    "name": obj["metadata"]["name"],
                }
                err = self.analyze_refs_from_pod_spec(spec, referrer)
                if err:
                    return err
        return None

    def analyze_refs_from_strategy(self, build_strategy, namespace, referrer):
        # Determine 'from' reference
        def _determine_source_strategy():
            for src in ("sourceStrategy", "dockerStrategy", "customStrategy"):
                strategy = build_strategy.get(src)
                if strategy:
                    return strategy.get("from")
            return None

        def _parse_image_stream_image_name(name):
            v = name.split("@")
            if len(v) != 2:
                return (
                    None,
                    None,
                    "expected exactly one @ in the isimage name %s" % name,
                )
            name = v[0]
            tag = v[1]
            if len(name) == 0 or len(tag) == 0:
                return (
                    None,
                    None,
                    "image stream image name %s must have a name and ID" % name,
                )
            return name, tag, None

        def _parse_image_stream_tag_name(name):
            if "@" in name:
                return (
                    None,
                    None,
                    "%s is an image stream image, not an image stream tag" % name,
                )
            v = name.split(":")
            if len(v) != 2:
                return (
                    None,
                    None,
                    "expected exactly one : delimiter in the istag %s" % name,
                )
            name = v[0]
            tag = v[1]
            if len(name) == 0 or len(tag) == 0:
                return (
                    None,
                    None,
                    "image stream tag name %s must have a name and a tag" % name,
                )
            return name, tag, None

        from_strategy = _determine_source_strategy()
        if from_strategy:
            if from_strategy.get("kind") == "DockerImage":
                docker_image_ref = from_strategy.get("name").strip()
                if len(docker_image_ref) > 0:
                    err = self.analyze_reference_image(docker_image_ref, referrer)
            elif from_strategy.get("kind") == "ImageStreamImage":
                name, tag, error = _parse_image_stream_image_name(
                    from_strategy.get("name")
                )
                if error:
                    if not self.ignore_invalid_refs:
                        return error
                else:
                    namespace = from_strategy.get("namespace") or namespace
                    self.used_images.append(
                        {"namespace": namespace, "name": name, "tag": tag}
                    )
            elif from_strategy.get("kind") == "ImageStreamTag":
                name, tag, error = _parse_image_stream_tag_name(
                    from_strategy.get("name")
                )
                if error:
                    if not self.ignore_invalid_refs:
                        return error
                else:
                    namespace = from_strategy.get("namespace") or namespace
                    self.used_tags.append(
                        {"namespace": namespace, "name": name, "tag": tag}
                    )

    def analyze_refs_from_build_strategy(self, resources):
        # Json Path is always spec.strategy
        keys = ("BuildConfig", "Build")
        for k, objects in iteritems(resources):
            if k not in keys:
                continue
            for obj in objects:
                referrer = {
                    "kind": obj["kind"],
                    "namespace": obj["metadata"]["namespace"],
                    "name": obj["metadata"]["name"],
                }
                error = self.analyze_refs_from_strategy(
                    obj["spec"]["strategy"], obj["metadata"]["namespace"], referrer
                )
                if error is not None:
                    return "%s/%s/%s: %s" % (
                        referrer["kind"],
                        referrer["namespace"],
                        referrer["name"],
                        error,
                    )

    def analyze_image_stream(self, resources):
        # Analyze image reference from Pods
        error = self.analyze_refs_from_pods(resources["Pod"])
        if error:
            return None, None, error

        # Analyze image reference from Resources creating Pod
        error = self.analyze_refs_pod_creators(resources)
        if error:
            return None, None, error

        # Analyze image reference from Build/BuildConfig
        error = self.analyze_refs_from_build_strategy(resources)
        return self.used_tags, self.used_images, error