]>
Commit | Line | Data |
---|---|---|
a4b75251 TL |
1 | import requests |
2 | from typing import List, Dict, Tuple | |
3 | from requests import Response | |
4 | ||
5 | ||
6 | class Registry: | |
7 | ||
8 | def __init__(self, url: str): | |
9 | self._url: str = url | |
10 | ||
11 | @property | |
12 | def api_domain(self) -> str: | |
13 | if self._url == 'docker.io': | |
14 | return 'registry-1.docker.io' | |
15 | return self._url | |
16 | ||
17 | def get_token(self, response: Response) -> str: | |
18 | realm, params = self.parse_www_authenticate(response.headers['Www-Authenticate']) | |
19 | r = requests.get(realm, params=params) | |
20 | r.raise_for_status() | |
21 | ret = r.json() | |
22 | if 'access_token' in ret: | |
23 | return ret['access_token'] | |
24 | if 'token' in ret: | |
25 | return ret['token'] | |
26 | raise ValueError(f'Unknown token reply {ret}') | |
27 | ||
28 | def parse_www_authenticate(self, text: str) -> Tuple[str, Dict[str, str]]: | |
29 | # 'Www-Authenticate': 'Bearer realm="https://auth.docker.io/token",service="registry.docker.io",scope="repository:ceph/ceph:pull"' | |
30 | r: Dict[str, str] = {} | |
31 | for token in text.split(','): | |
32 | key, value = token.split('=', 1) | |
33 | r[key] = value.strip('"') | |
34 | realm = r.pop('Bearer realm') | |
35 | return realm, r | |
36 | ||
37 | def get_tags(self, image: str) -> List[str]: | |
38 | tags = [] | |
39 | headers = {'Accept': 'application/json'} | |
40 | url = f'https://{self.api_domain}/v2/{image}/tags/list' | |
41 | while True: | |
2a845540 TL |
42 | try: |
43 | r = requests.get(url, headers=headers) | |
44 | except requests.exceptions.ConnectionError as e: | |
45 | msg = f"Cannot get tags from url '{url}': {e}" | |
46 | raise ValueError(msg) from e | |
a4b75251 TL |
47 | if r.status_code == 401: |
48 | if 'Authorization' in headers: | |
49 | raise ValueError('failed authentication') | |
50 | token = self.get_token(r) | |
51 | headers['Authorization'] = f'Bearer {token}' | |
52 | continue | |
53 | r.raise_for_status() | |
54 | ||
55 | new_tags = r.json()['tags'] | |
56 | tags.extend(new_tags) | |
57 | ||
58 | if 'Link' not in r.headers: | |
59 | break | |
60 | ||
61 | # strip < > brackets off and prepend the domain | |
62 | url = f'https://{self.api_domain}' + r.headers['Link'].split(';')[0][1:-1] | |
63 | continue | |
64 | ||
65 | return tags |