poc_regprox.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# import coloredlogs
import logging
import pprint
import requests
import sys


def get_next(headers):
    link = headers.get("Link")
    if link is None:
        return None
    log.info(pprint.pformat(link))
    link_url, link_rel = link.split(';')
    link_url = link_url.strip("<> ")
    if link_rel.strip(" ") == 'rel="next"':
        return link_url
    return None


def process_repositories(jason, _filter):
    repositories = jason["repositories"]
    log.debug("Got %d repositories" % len(repositories))
    log.debug("First repository is: %s " % repositories[0])
    filtered = list(filter(lambda elt: _filter in elt, repositories))
    log.info("Repositories matching %s: %s" % (_filter, pprint.pformat(filtered)))
    return filtered


logging.basicConfig(level=logging.ERROR)
log = logging.getLogger(__name__)
log.setLevel(logging.ERROR)
# coloredlogs.install(level=logging.ERROR)
# coloredlogs.install(level=logging.ERROR, logger=log)

base_url = "https://registry-proxy.engineering.redhat.com"

# Get a token
r = requests.get(
    base_url + "/v2/auth?scope=repository:rh-osbs:pull"
)
token = r.json()["token"]
headers = {"Authorization": "Bearer " + token}


def with_docker_api(_filter):
    next_link = "/v2/_catalog"
    repos = []
    while next_link is not None:
        r = requests.get(
            base_url + next_link, headers=headers
        )
        repos.extend(process_repositories(r.json(), _filter))
        next_link = get_next(r.headers)

    print("Repositories found:")
    print("===================")
    pprint.pprint(repos)
    print("==== Total: %d" % len(repos))


if __name__ == "__main__":
    if len(sys.argv) > 1:
        with_docker_api(sys.argv[1])
    else:
        print("Pass a string to filter on as argument (e.g.: %s rh-osbs/rhosp16)" % sys.argv[0])