Make a request to the URL if that fails try to obtain an authentication token and try again.
(self, url: str)
| 75 | return registry, repository |
| 76 | |
| 77 | async def docker_api_request(self, url: str): |
| 78 | """Make a request to the URL if that fails try to obtain an authentication token and try again.""" |
| 79 | for _ in range(2): |
| 80 | response = await self.helpers.request(url, headers=self.headers, follow_redirects=True) |
| 81 | if response is not None and response.status_code != 401: |
| 82 | return response |
| 83 | www_auth = response.headers.get("www-authenticate", "") |
| 84 | realm, service, scope = self._parse_www_authenticate(www_auth) |
| 85 | if not all([realm, service, scope]): |
| 86 | self.log.warning(f"Could not obtain realm, service or scope from {url}") |
| 87 | break |
| 88 | if not self._validate_realm(url, realm): |
| 89 | break |
| 90 | auth_url = f"{realm}?service={service}&scope={scope}" |
| 91 | auth_response = await self.helpers.request(auth_url) |
| 92 | if not auth_response: |
| 93 | self.log.warning(f"Could not obtain token from {auth_url}") |
| 94 | break |
| 95 | auth_json = auth_response.json() |
| 96 | token = auth_json["token"] |
| 97 | self.headers.update({"Authorization": f"Bearer {token}"}) |
| 98 | return None |
| 99 | |
| 100 | @staticmethod |
| 101 | def _parse_www_authenticate(header): |
no test coverage detected