#!/usr/bin/env python3 import contextlib import json from typing import Iterator from typing import Literal from typing import Protocol from urllib import request from urllib.parse import urlencode DOCKER_URLS = { "auth": "https://auth.docker.io", "registry": "https://registry-1.docker.io", } HttpMethods = Literal["GET", "POST", "PUT", "DELETE"] def http_request( method: HttpMethods, url: str, body: dict | None = None, *, headers: dict | None = None, ) -> dict: if not headers: headers = {} if not body: body = {} req = request.Request( method=method, url=url, data=json.dumps(body).encode("utf-8"), headers=headers, ) res = request.urlopen(req) return json.load(res) class HttpReqFunc(Protocol): def __call__( self, method: HttpMethods, url: str, *, body: dict | None = None, headers: dict | None = None, ) -> dict: ... def generate_token_params(image: str) -> str: return urlencode( dict( service="registry.docker.io", scope=f"repository:{image}:pull", ), ) @contextlib.contextmanager def authorized_requests(image_name: str) -> Iterator[HttpReqFunc]: token_query_params = generate_token_params(image_name) token_uri = f"{DOCKER_URLS['auth']}/token?{token_query_params}" auth_token = http_request( "GET", token_uri, ).get("token") def wrapped_http_request(*args, **kwargs) -> dict: headers = kwargs.pop("headers", dict()) headers["Authorization"] = f"Bearer {auth_token}" return http_request(*args, **kwargs, headers=headers) yield wrapped_http_request def main() -> int: image_names = ["library/traefik"] for image_name in image_names: with authorized_requests(image_name) as make_request: info = make_request( "GET", f"{DOCKER_URLS['registry']}/v2/{image_name}/tags/list?n=50&last=v3.1.0", ) print(info) return 0 if __name__ == "__main__": raise SystemExit(main())