GetDockerImageLatestTag

Gets docker image latest tag. Script simulates the docker pull flow but doesn't actually pull the image. Returns an entry with the docker image latest tag if all is good, otherwise will return an error.

Type

python

Pack

CommonScripts

Source

import re

import demistomock as demisto
import requests
from CommonServerPython import *
from packaging.version import parse as parse_version

from CommonServerUserPython import *

ACCEPT_HEADER = {
    "Accept": "application/json, "
    "application/vnd.docker.distribution.manifest.v2+json, "
    "application/vnd.docker.distribution.manifest.list.v2+json"
}

# use 10 seconds timeout for requests
TIMEOUT = 10
DEFAULT_REGISTRY = "registry-1.docker.io"

# OCI artifact tag suffixes that are not runnable Docker images.
# These are produced by tools like cosign (.sig), OCI referrers (.att, .sbom), etc.
NON_RUNNABLE_TAG_SUFFIXES = (".sig", ".att", ".sbom")


def parse_www_auth(www_auth):
    """Parse realm and service from www-authenticate string of the form:
    Bearer realm="https://auth.docker.io/token",service="registry.docker.io"

    :param www_auth: www-authenticate header value
    :type www_auth: string
    """
    match = re.match(r'.*realm="(.+)",service="(.+)".*', www_auth, re.IGNORECASE)
    if not match:
        return None
    return (match.group(1), match.group(2))


def docker_auth(image_name, verify_ssl=True, registry=DEFAULT_REGISTRY, gateway_creds=""):
    """
    Authenticate to the docker service. Return an authentication token if authentication is required.
    """
    res = requests.get(f"https://{registry}/v2/", headers=ACCEPT_HEADER, timeout=TIMEOUT, verify=verify_ssl)
    if res.status_code == 401:  # need to authenticate
        # defaults in case we fail for some reason
        realm = "https://auth.docker.io/token"
        service = "registry.docker.io"
        # Should contain header: Www-Authenticate
        www_auth = res.headers.get("www-authenticate")
        if www_auth:
            parse_auth = parse_www_auth(www_auth)
            if parse_auth:
                realm, service = parse_auth
            else:
                demisto.info(f"Failed parsing www-authenticate header: {www_auth}")
        else:
            demisto.info(f"Failed extracting www-authenticate header from registry: {registry}, final url: {res.url}")
        headers = ACCEPT_HEADER.copy()
        if gateway_creds and registry != DEFAULT_REGISTRY:
            headers["Authorization"] = f"Basic {gateway_creds}"
        res = requests.get(
            f"{realm}?scope=repository:{image_name}:pull&service={service}", headers=headers, timeout=TIMEOUT, verify=verify_ssl
        )
        res.raise_for_status()
        res_json = res.json()
        return res_json.get("token")
    else:
        res.raise_for_status()
        return None


def is_runnable_tag(tag: Any) -> bool:
    """Return True if *tag* represents a runnable Docker image tag.

    Filters out OCI artifact tags such as cosign signature tags (``*.sig``),
    attestation tags (``*.att``), and SBOM tags (``*.sbom``) which are not
    valid Docker image tags and cannot be used to run a container.

    Note that only tags *ending* with one of the artifact suffixes are
    filtered, so a legitimate tag that merely contains the substring
    (e.g. ``my.sig.image``) is preserved.

    Args:
        tag: A Docker image tag name. Non-string values are treated as
            non-runnable rather than raising.

    Returns:
        True when the tag is a runnable image tag, False otherwise.
    """
    if not isinstance(tag, str) or not tag:
        return False
    return not tag.endswith(NON_RUNNABLE_TAG_SUFFIXES)


def clear_non_numbered_tags(tags):
    """Clears a given tags list to only keep numbered tags

    Args:
        tags(list): list of docker image tag names - ordered in lexical order

    Returns:
        a tag list with only numbered tags
    """
    only_numbered_tags = []
    for tag in tags:
        number_token = 1
        split_tag = tag.split(".")
        for sub_section in split_tag:
            if not sub_section.isdigit():
                number_token = 0

        if number_token:
            only_numbered_tags.append(tag)

    return only_numbered_tags


def lexical_find_latest_tag(tags):
    """Will return the latest numeric docker image tag if possible - otherwise will return the last lexical tag.

    for example for the tag list: [2.0.2000, 2.1.2700 2.1.373, latest], will return 2.1.2700

    Non-runnable OCI artifact tags (e.g. ``*.sig``, ``*.att``, ``*.sbom``) are
    excluded before any comparison.

    Args:
        tags(list): list of docker image tag names - ordered in lexical order

    Returns:
        The latest runnable tag, or an empty string when no runnable tag exists.
    """
    runnable_tags = [tag for tag in tags if is_runnable_tag(tag)]
    if not runnable_tags:
        demisto.debug("No runnable tags found after filtering non-runnable artifact tags.")
        return ""

    only_numbered_tags = clear_non_numbered_tags(runnable_tags)

    if len(only_numbered_tags) == 0:
        return runnable_tags[-1]

    max_tag = only_numbered_tags[0]

    for num_tag in only_numbered_tags:
        if parse_version(max_tag) < parse_version(num_tag):
            max_tag = num_tag

    return max_tag


def find_latest_tag_by_date(tags):
    """Get the latest tags by datetime comparison.

    Non-runnable OCI artifact tags (e.g. ``*.sig``, ``*.att``, ``*.sbom``) are
    excluded before comparison so that cosign signature artifacts are never
    returned as the "latest" image tag.

    Args:
        tags(list): List of dictionaries representing the docker image tags

    Returns:
        The last updated docker image tag name
    """
    latest_tag_name = "latest"
    latest_tag_date = datetime.now() - timedelta(days=400000)
    for tag in tags:
        tag_name = tag.get("name")
        # is_runnable_tag() also rejects non-string / empty values, so past this
        # guard tag_name is guaranteed to be a usable string.
        if not is_runnable_tag(tag_name):
            demisto.debug(f"Skipping non-runnable or invalid tag: {tag_name!r}")
            continue

        last_updated = tag.get("last_updated")
        try:
            tag_date = datetime.strptime(last_updated, "%Y-%m-%dT%H:%M:%S.%fZ")
        except (TypeError, ValueError):
            demisto.debug(f"Skipping tag {tag_name!r} with unparsable last_updated value: {last_updated!r}")
            continue

        if tag_date >= latest_tag_date:
            latest_tag_date = tag_date
            latest_tag_name = tag_name

    return latest_tag_name


def main():
    if demisto.args().get("use_system_proxy") == "no":
        # Remove proxy environment variables if they exist
        for proxy_var in ["HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]:
            os.environ.pop(proxy_var, None)
    verify_ssl = demisto.args().get("trust_any_certificate") != "yes"
    docker_full_name = demisto.args()["docker_image"]
    gateway_creds = demisto.args().get("creds_for_opp", "")
    registry = DEFAULT_REGISTRY
    image_name = docker_full_name
    if docker_full_name.count("/") > 1:
        registry, image_name = docker_full_name.split("/", 1)
    try:
        auth_token = docker_auth(image_name, verify_ssl, registry, gateway_creds)
        headers = ACCEPT_HEADER.copy()
        if auth_token:
            headers["Authorization"] = f"Bearer {auth_token}"

        # first try to get the docker image tags using normal http request
        res = requests.get(
            url=f"https://hub.docker.com/v2/repositories/{image_name}/tags",
            timeout=TIMEOUT,
            verify=verify_ssl,
        )
        if res.status_code == 200:
            tags = res.json().get("results", [])
            # if http request successful find the latest tag by date in the response
            if tags:
                tag = find_latest_tag_by_date(tags)
            else:
                tag = ""
                demisto.debug(f"No tags, {tag=}")

        else:
            # if http request did not successed than get tags using the API.
            # See: https://docs.docker.com/registry/spec/api/#listing-image-tags
            res = requests.get(
                f"https://{registry}/v2/{image_name}/tags/list", headers=headers, timeout=TIMEOUT, verify=verify_ssl
            )
            res.raise_for_status()
            # the API returns tags in lexical order with no date info - so try an get the numeric highest tag
            tags = res.json().get("tags", [])
            if tags:
                tag = lexical_find_latest_tag(tags)
            else:
                tag = ""
                demisto.debug(f"No tags, {tag=}")

        demisto.results(tag)
    except Exception as ex:
        return_error(f"Failed getting tag for: {docker_full_name}. Err: {ex!s}")


# python2 uses __builtin__ python3 uses builtins
if __name__ in ("__main__", "__builtin__", "builtins"):
    main()

README