RegBrowser/explorer/registry_client.py

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"