| 111 | return params.get("realm", ""), params.get("service", ""), params.get("scope", "") |
| 112 | |
| 113 | def _validate_realm(self, registry_url, realm): |
| 114 | registry_host = self.helpers.urlparse(registry_url).hostname or "" |
| 115 | realm_host = self.helpers.urlparse(realm).hostname or "" |
| 116 | _, registry_domain = self.helpers.split_domain(registry_host) |
| 117 | _, realm_domain = self.helpers.split_domain(realm_host) |
| 118 | if not realm_domain or realm_domain != registry_domain: |
| 119 | self.warning(f"Auth realm TLD ({realm_domain}) does not match registry TLD ({registry_domain}), skipping") |
| 120 | return False |
| 121 | return True |
| 122 | |
| 123 | async def get_tags(self, registry, repository): |
| 124 | url = f"{registry}/v2/{repository}/tags/list" |