85 lines
2.1 KiB
Text
85 lines
2.1 KiB
Text
|
#!/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())
|