#!/usr/bin/env python3
"""OCI Registry Helper
Usage:
ocirh <subcommand> [<args>...]
Subcommands:
repos Lists repositories in the registry. Repos correspond to images
pushed to the registry.
tags Lists tags of the given repository.
manifests Lists manifests of the given repository for the given tag.
rmi Removes a tag from an image. If given tag is the only tag,
removes the image.
gc Runs garbage collection on the registry. Requires SSH public key
access to registry server.
rmr Removes given repository from the registry. Requires SSH public
key access to registry server.
Examples:
Suppose we have an image called 'fedora-toolbox' tagged with 'latest'.
ocirh repos
ocirh tags fedora-toolbox
ocirh manifests fedora-toolbox latest
ocirh rmi fedora-toolbox latest
ocirh gc
ocirh rmr fedora-toolbox
"""
import http.client
import json
import logging
import math
import subprocess
from docopt import docopt
from rich import print
from rich.console import Group
from rich.logging import RichHandler
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from rich.traceback import install
from rich.tree import Tree
install(show_locals=True)
# Rich logging handler
FORMAT = "%(message)s"
logging.basicConfig(
level="NOTSET",
format=FORMAT,
datefmt="[%X]",
handlers=[RichHandler(rich_tracebacks=True)],
)
log = logging.getLogger("rich")
# Taken from https://stackoverflow.com/a/14822210
#
# How this function works:
# If size_bytes == 0, returns 0 B.
# size_name is a tuple containing binary prefixes for bytes.
#
# math.log takes the logarithm of size_bytes to base 1024.
# math.floor rounds down the result of math.log to the nearest integer.
# int ensures the result of math.floor is of type int, and stores it in i.
# The value of i is used to determine which binary prefix to use from
# size_name.
#
# math.pow returns the value of 1024 raised to the power of i, stores it in p.
#
# round takes the value of size_bytes, divides it by p, and stores the result
# in s at precision of 2 decimal places.
#
# A formatted string with size s and binary prefix size_name[i] is returned.
def convert_size(size_bytes: int) -> str:
"""
Converts a decimal integer of bytes to its respective binary-prefixed size.
Parameters:
size_bytes (int): A decimal integer.
Returns:
(str): Binary-prefixed size of size_bytes formatted as a string.
"""
if size_bytes == 0:
return "0 B"
size_name = ("B", "KiB", "MiB", "GiB")
i = int(math.floor(math.log(size_bytes, 1024)))
p = math.pow(1024, i)
s = round(size_bytes / p, 2)
return "%s %s" % (s, size_name[i])
REGISTRY_URL = "registry.hyperreal.coffee"
def get_auth() -> str:
"""
Get the base64 encoded password for registry autentication.
Returns:
auth (str): A string containing the base64 encoded password.
"""
try:
with open("/run/user/1000/containers/auth.json", "r") as authfile:
json_data = json.loads(authfile.read())
except Exception as ex:
log.exception(ex)
auth = json_data["auths"][REGISTRY_URL]["auth"]
return auth
def get_headers() -> dict:
"""
Returns headers for HTTP request authentication to the registry server.
Returns:
headers (dict): A dict of HTTP headers
"""
return {
"Accept": "application/vnd.oci.image.manifest.v1+json",
"Authorization": "Basic " + get_auth(),
}
def get_json_response(request: str, url: str) -> dict:
"""
Connects to registry and returns response data as JSON.
Parameters:
request (str): A string like "GET" or "DELETE"
url (str) : A string containing the URL of the requested data
Returns:
json_data (dict): JSON data as a dict object
"""
conn = http.client.HTTPSConnection(REGISTRY_URL)
headers = get_headers()
try:
conn.request(request, url, "", headers)
res = conn.getresponse()
data = res.read()
json_data = json.loads(data.decode("utf-8"))
except Exception as ex:
log.exception(ex)
return json_data
def get_repositories():
"""
Prints a Rich Tree that lists the repositories of the registry.
"""
json_data = get_json_response("GET", "/v2/_catalog")
repo_tree = Tree("[green]Repositories")
for repo in json_data["repositories"]:
repo_tree.add("[blue]%s" % repo)
print(repo_tree)
def get_tags(repo: str):
"""
Prints a Rich Tree that lists the tags for the given repository.
Parameters:
repo (str): A string containing the name of the repo
"""
json_data = get_json_response("GET", "/v2/" + repo + "/tags/list")
tags_tree = Tree("[green]%s tags" % repo)
for tag in json_data["tags"]:
tags_tree.add("[cyan]:%s" % tag)
print(tags_tree)
def get_manifests(repo: str, tag: str):
"""
Prints a Rich grid table that displays the manifests and metadata of the
image repository.
Parameters:
repo (str): A string containing the name of the repo
tag (str) : A string containing the tag of the desired image
"""
json_data = get_json_response("GET", "/v2/" + repo + "/manifests/" + tag)
grid_meta = Table.grid(expand=True)
grid_meta.add_column()
grid_meta.add_column()
meta_schema_version_key = Text("Schema version")
meta_schema_version_key.stylize("bold green", 0)
meta_schema_version_value = Text(str(json_data["schemaVersion"]))
meta_media_type_key = Text("Media type")
meta_media_type_key.stylize("bold green", 0)
meta_media_type_value = Text(json_data["mediaType"])
grid_meta.add_row(meta_schema_version_key, meta_schema_version_value)
grid_meta.add_row(meta_media_type_key, meta_media_type_value)
grid_config = Table.grid(expand=True)
grid_config.add_column()
grid_config.add_column()
config_media_type_key = Text("Media type")
config_media_type_key.stylize("bold green", 0)
config_media_type_value = Text(json_data["config"]["mediaType"])
config_digest_key = Text("Digest")
config_digest_key.stylize("bold green", 0)
config_digest_value = Text(json_data["config"]["digest"])
config_size_key = Text("Size")
config_size_key.stylize("bold green", 0)
config_size_value = Text(convert_size(json_data["config"]["size"]))
grid_config.add_row(config_media_type_key, config_media_type_value)
grid_config.add_row(config_digest_key, config_digest_value)
grid_config.add_row(config_size_key, config_size_value)
grid_annotations = Table.grid(expand=True)
grid_annotations.add_column()
grid_annotations.add_column()
for item in json_data["annotations"].items():
annotations_item_key = Text(item[0])
annotations_item_key.stylize("bold green", 0)
annotations_item_value = Text(item[1])
grid_annotations.add_row(annotations_item_key, annotations_item_value)
total_size = sum(layer.get("size") for layer in json_data["layers"])
table_layers = Table(box=None, show_footer=True)
table_layers.add_column(
"Digest", justify="right", style="yellow", no_wrap=True, footer="Total size:"
)
table_layers.add_column(
"Size",
justify="left",
style="cyan",
no_wrap=True,
footer=convert_size(total_size),
)
for layer in json_data["layers"]:
table_layers.add_row(layer.get("digest"), convert_size(layer.get("size")))
panel_group = Group(
Panel(grid_meta, title="[bold blue]Metadata"),
Panel(grid_config, title="[bold blue]Config"),
Panel(grid_annotations, title="Annotations"),
Panel(
table_layers,
title="[bold blue]Layers: %s" % json_data["layers"][0].get("mediaType"),
),
)
print(Panel(panel_group, title="[bold blue]%s:%s" % (repo, tag)))
def delete_image(repo: str, tag: str):
"""
Removes the given tag from the image. If the given tag is the only tag,
removes the image.
Parameters:
repo (str): A string containing the name of the repo
tag (str) : A string containing the tag to be removed
"""
try:
conn = http.client.HTTPSConnection(REGISTRY_URL)
headers = get_headers()
conn.request("GET", "/v2/" + repo + "/manifests/" + tag, "", headers)
res = conn.getresponse()
docker_content_digest = res.getheader("Docker-Content-Digest")
except Exception as ex:
log.exception(ex)
try:
conn.request(
"DELETE", "/v2/" + repo + "/manifests/" + docker_content_digest, "", headers
)
except Exception as ex:
log.exception(ex)
print("Untagged %s:%s successfully" % (repo, tag))
def garbage_collection():
"""
Runs garbage collection command on the remote registry server. Requires
SSH public key access.
"""
command = "/usr/local/bin/registry-gc"
try:
ssh = subprocess.Popen(
["ssh", "%s" % REGISTRY_URL, command],
shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
result = ssh.stdout.readlines()
if result == []:
log.error(ssh.stderr.readlines())
else:
print(result)
except Exception as ex:
log.exception(ex)
def remove_repo(repo: str):
"""
Runs command on remote registry server to remove the given repo.
Parameters:
repo (str): A string containing the name of the repo.
"""
command = "/usr/local/bin/registry-rm-repo " + repo
try:
ssh = subprocess.Popen(
["ssh", "%s" % REGISTRY_URL, command],
shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
)
result = ssh.stdout.readlines()
if result == []:
log.error(ssh.stderr.readlines())
else:
print(result)
except Exception as ex:
log.exception(ex)
if __name__ == "__main__":
args = docopt(__doc__, options_first=True)
match args["<subcommand>"]:
case "repos":
get_repositories()
case "tags":
get_tags(args["<args>"][0])
case "manifests":
get_manifests(args["<args>"][0], args["<args>"][1])
case "rmi":
delete_image(args["<args>"][0], args["<args>"][1])
case "gc":
garbage_collection()
case "rmr":
remove_repo(args["<args>"])
case _:
if args["<subcommand>"] in ["help", None]:
exit(subprocess.call(["python3", "ocirh", "--help"]))
else:
exit(
"%r is not a ocirh subcommand. See 'ocirh --help."
% args["<subcommand>"]
)