import base64 import requests from requests.auth import HTTPBasicAuth from urllib.parse import urljoin class RegistryClient: def __init__(self, url, username='', password='', verify_ssl=True): self.base_url = url.rstrip('/') self.username = username self.password = password self.verify_ssl = verify_ssl self.session = requests.Session() self.session.verify = verify_ssl self.token = None self._auth_method = None if username and password: self.session.auth = HTTPBasicAuth(username, password) def _url(self, path): return f"{self.base_url}{path}" def _get_headers(self): headers = {'Accept': 'application/json'} if self.token: headers['Authorization'] = f'Bearer {self.token}' return headers def _request(self, method, path, **kwargs): url = self._url(path) kwargs.setdefault('headers', self._get_headers()) kwargs.setdefault('verify', self.verify_ssl) response = self.session.request(method, url, **kwargs) if response.status_code == 401: self._handle_401(response) kwargs['headers'] = self._get_headers() response = self.session.request(method, url, **kwargs) response.raise_for_status() return response def _handle_401(self, response): www_auth = response.headers.get('WWW-Authenticate', '') if 'Bearer' in www_auth: self._extract_token(www_auth) elif 'Basic' in www_auth and self.username and self.password: token_resp = self.session.get( self._url('/v2/'), auth=HTTPBasicAuth(self.username, self.password), verify=self.verify_ssl ) token_resp.raise_for_status() def _extract_token(self, www_auth): params = {} for part in www_auth.split(','): if '=' in part: key, value = part.split('=', 1) key = key.strip() value = value.strip().strip('"') if key == 'realm': params['realm'] = value else: params[key] = value if 'realm' in params: token_url = params.pop('realm') token_resp = self.session.get( token_url, params=params, verify=self.verify_ssl ) token_resp.raise_for_status() self.token = token_resp.json().get('token') def check_connection(self): try: response = self._request('GET', '/v2/') return True, 'Conexión exitosa' except requests.exceptions.ConnectionError as e: return False, f'Error de conexión: {str(e)}' except requests.exceptions.HTTPError as e: return False, f'Error HTTP: {e.response.status_code}' except Exception as e: return False, f'Error: {str(e)}' def list_catalog(self): response = self._request('GET', '/v2/_catalog') return response.json().get('repositories', []) def list_tags(self, repository): response = self._request('GET', f'/v2/{repository}/tags/list') data = response.json() tags = data.get('tags', []) return sorted(tags) if tags else [] def get_manifest(self, repository, reference): headers = { 'Accept': ', '.join([ 'application/vnd.docker.distribution.manifest.v2+json', 'application/vnd.docker.distribution.manifest.list.v2+json', 'application/vnd.oci.image.manifest.v1+json', 'application/vnd.oci.image.index.v1+json', ]) } response = self._request('GET', f'/v2/{repository}/manifests/{reference}', headers=headers) return response.json(), response.headers def get_blob_summaries(self, repository, manifest): layers = manifest.get('layers', []) config = manifest.get('config', {}) result = { 'layers': [], 'config_size': config.get('size', 0), 'config_digest': config.get('digest', ''), 'total_size': 0, } for layer in layers: digest = layer.get('digest', '') size = layer.get('size', 0) result['layers'].append({ 'digest': digest, 'size': size, 'size_human': self._human_size(size), }) result['total_size'] += size result['total_size_human'] = self._human_size(result['total_size']) return result def get_image_detail(self, repository, reference): manifest, headers = self.get_manifest(repository, reference) media_type = manifest.get('mediaType', '') manifest_list = 'manifest.list' in media_type or 'index' in media_type digest = headers.get('Docker-Content-Digest', headers.get('docker-content-digest', '')) result = { 'repository': repository, 'reference': reference, 'digest': digest, 'media_type': media_type, 'manifest_list': manifest_list, 'schema_version': manifest.get('schemaVersion'), 'media_type_display': self._media_type_display(media_type), } if manifest_list: manifests = manifest.get('manifests', []) result['sub_manifests'] = [] for sub in manifests: sub_platform = sub.get('platform', {}) result['sub_manifests'].append({ 'digest': sub.get('digest', ''), 'size': sub.get('size', 0), 'size_human': self._human_size(sub.get('size', 0)), 'architecture': sub_platform.get('architecture', 'unknown'), 'os': sub_platform.get('os', 'unknown'), }) else: config_data = self._get_config(repository, config_digest=manifest.get('config', {}).get('digest', '')) result['architecture'] = config_data.get('architecture', 'unknown') result['os'] = config_data.get('os', 'unknown') result['created'] = config_data.get('created', '') result['author'] = config_data.get('author', '') history = config_data.get('history', []) result['history'] = [] for h in history: created_by = h.get('created_by', '') if created_by and created_by != '/bin/sh -c #(nop) ADD file:' in created_by: result['history'].append({ 'created_by': created_by[:200] + ('...' if len(created_by) > 200 else ''), }) blob_info = self.get_blob_summaries(repository, manifest) result['layers'] = blob_info['layers'] result['total_size'] = blob_info['total_size'] result['total_size_human'] = blob_info['total_size_human'] return result def _get_config(self, repository, config_digest): if not config_digest: return {} try: response = self._request('GET', f'/v2/{repository}/blobs/{config_digest}') return response.json() except Exception: return {} def delete_manifest(self, repository, reference): try: manifest, resp_headers = self.get_manifest(repository, reference) media_type = manifest.get('mediaType', 'application/vnd.docker.distribution.manifest.v2+json') digest = resp_headers.get('Docker-Content-Digest', resp_headers.get('docker-content-digest', '')) if not digest: return False, 'No se pudo obtener el digest del manifest. Intenta eliminar desde el panel de administración del registry.' response = self._request( 'DELETE', f'/v2/{repository}/manifests/{digest}', headers={ 'Accept': media_type, }, ) return True, 'Imagen eliminada correctamente' except requests.exceptions.HTTPError as e: status = e.response.status_code if status == 404: return False, 'Imagen no encontrada (404). Es posible que el registry requiera eliminar por digest. Prueba obtener el digest desde los detalles de la imagen.' elif status == 405: return False, 'El registry no permite eliminación de imágenes (405 Method Not Allowed). Es necesario habilitar el "delete" en la configuración del registry.' elif status == 401 or status == 403: return False, f'No tienes permisos para eliminar imágenes ({status}). Verifica las credenciales y que el usuario tenga permisos de escritura.' else: body = '' try: body = e.response.text[:300] except Exception: pass return False, f'Error al eliminar ({status}): {body}' except Exception as e: return False, f'Error: {str(e)}' def _media_type_display(self, media_type): types = { 'application/vnd.docker.distribution.manifest.v2+json': 'Docker Manifest v2', 'application/vnd.docker.distribution.manifest.list.v2+json': 'Docker Manifest List v2', 'application/vnd.oci.image.manifest.v1+json': 'OCI Image Manifest v1', 'application/vnd.oci.image.index.v1+json': 'OCI Image Index v1', } return types.get(media_type, media_type or 'Desconocido') @staticmethod def _human_size(size_bytes): for unit in ['B', 'KB', 'MB', 'GB', 'TB']: if size_bytes < 1024.0: return f"{size_bytes:.1f} {unit}" size_bytes /= 1024.0 return f"{size_bytes:.1f} PB"