Short and dirty poc of docker retagging without pulling. Same registry, different repositories.

retag.py

"""
Re-tag images on distant registry without pulling them, even if not in the same
repository. Only compatible with v2 manifests.

Usage:
    retag <source> <destination> [--debug] [--insecure]
    retag -h | --help
"""

# Disclaimer : no QA and no lint yet, poc quality script.
# Does not support auth yet.

import json
import logging

import colorlog
import docopt
import requests

class DifferentRegistriesError(Exception):
    pass

class ImageNameParsingError(Exception):
    pass

class ImageNotFoundError(Exception):
    pass

class MountingBlobError(Exception):
    pass

class NotAuthenticatedError(Exception):
    pass

class PostingManifestError(Exception):
    pass

def spawn_logger(debug=False):
    """
    Creates and configure logging.
    """
    log_format = "%(levelname)s - " "%(message)s"
    colorlog_format = "%(log_color)s " f"{log_format}"
    colorlog.basicConfig(format=colorlog_format)
    log = logging.getLogger("retag")
    if debug:
        log.setLevel(logging.DEBUG)
    else:
        log.setLevel(logging.INFO)
    return log

def check_registry(registry_url):
    response = requests.get(registry_url)
    if response.status_code != 200:
        raise NotAuthenticatedError(registry_url, response)
    return True

def parse_image_name(image_name:str):
    name_parts = image_name.split("/")
    registry = name_parts[0]
    if len(name_parts[-1].split(":")) == 2:
        # nginx:apline
        tag = name_parts[-1].split(":")[1]
    elif len(name_parts[-1].split(":")) == 1:
        # nginx
        tag = "latest"
    else:
        # nginx:yolo:invalid
        raise ImageNameParsingError(image_name)
    repository = "/".join(name_parts[1:-1] + [name_parts[-1].split(":")[0]])
    return registry, repository, tag

def get_manifest(registry_url, repository, tag):
    headers = {"Accept": "application/vnd.docker.distribution.manifest.v2+json" }
    response = requests.get(f"{registry_url}{repository}/manifests/{tag}", headers=headers)
    if response.status_code != 200:
        raise ImageNotFoundError(registry_url, repository, tag)
    manifest = response.json()
    layers = [layer["digest"] for layer in manifest["layers"]]
    config = manifest["config"]["digest"]
    return layers, config, manifest

def layer_already_present(registry_url, repository, digest):
    response = requests.head(f"{registry_url}{repository}/blobs/{digest}")
    return response.status_code == 200

def mount_layer(registry_url, source_repository, destination_repository, digest):
    params = {"from":source_repository, "mount": digest}
    response = requests.post(f"{registry_url}{destination_repository}/blobs/uploads/", params=params)
    if response.status_code != 201:
        raise MountingBlobError(digest)

def post_manifest(registry_url, destination_repository, destination_tag, manifest):
    headers = {"Content-Type": "application/vnd.docker.distribution.manifest.v2+json" }
    response = requests.put(f"{registry_url}{destination_repository}/manifests/{destination_tag}", json=manifest, headers=headers)
    if response.status_code != 201:
        raise PostingManifestError(destination_repository, destination_tag, manifest)

def main():
    """ Main """
    args = docopt.docopt(__doc__)
    log = spawn_logger(debug=args["--debug"])
    log.debug(args)

    if args["--insecure"]:
        pattern = "http://"
    else:
        pattern = "https://"

    source_registry, source_repository, source_tag = parse_image_name(args["<source>"])
    destination_registry, destination_repository, destination_tag = parse_image_name(args["<destination>"])

    if source_registry != destination_registry:
        raise DifferentRegistriesError(source_registry, destination_registry)

    registry_url = f"{pattern}{source_registry}/v2/"
    log.debug("Registry to contact : %s", registry_url)
    check_registry(registry_url)
    log.info("Registry reachable")

    layers, config, manifest = get_manifest(registry_url, source_repository, source_tag)
    log.debug("layers : %r", layers)
    log.debug("config : %r", config)
    log.debug("manifest : %r", manifest)

    for layer in layers:
        if not layer_already_present(registry_url, destination_repository, layer):
            log.info("Mounting layer %s", layer)
            mount_layer(registry_url, source_repository, destination_repository, layer)
        else:
            log.info("Layer already mounted %s", layer)

    log.info("All layers mounted on destination")
    if not layer_already_present(registry_url, destination_repository, config):
        log.info("Mounting configuration %s", config)
        mount_layer(registry_url, source_repository, destination_repository, config)
    else:
        log.info("Configuration already mounted %s", config)

    log.info("Configuration mounted")
    post_manifest(registry_url, destination_repository, destination_tag, manifest)

    log.info("Image successfully re-tagged")

if __name__ == "__main__":
    main()