docker_pull.py

"""
Docker Image Puller - Pure Python CLI for pulling and saving Docker images

A pure Python tool for downloading container images from Registry v2 endpoints
without requiring Docker itself. Perfect for air-gapped environments, corporate
networks with proxies, and transferring images between systems.

Features:
- No Docker installation required (pure Python, standard library only)
- Docker Registry v2 and OCI endpoints, including private registries
- Full corporate proxy support with authentication
- Multi-architecture support (amd64, arm64, etc.)
- SSL/TLS configuration options
- Real-time download progress tracking
- Outputs Docker-compatible tar files

Usage:
    python docker_pull.py <image:tag> [options]

Examples:
    python docker_pull.py ubuntu:latest
    python docker_pull.py ghcr.io/nextcloud-releases/all-in-one:latest
    python docker_pull.py nginx:alpine --output my-nginx.tar
    python docker_pull.py --arch arm64 alpine:latest
    python docker_pull.py ubuntu:20.04 --proxy http://proxy:8080 --proxy-auth user:pass

Requirements:
- Python 3.11 or later
- Internet connection (or corporate network with proxy)

Copyright (c) 2025 ZacharyArthur
Licensed under the MIT License - see LICENSE file for details
"""

import argparse
import base64
import gzip
import hashlib
import ipaddress
import json
import logging
import os
import re
import shutil
import signal
import ssl
import sys
import tarfile
import tempfile
import threading
import time
import urllib.request
import zlib
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlencode, urlparse
from urllib.request import (
    HTTPRedirectHandler,
    ProxyHandler,
    Request,
    build_opener,
)

# Module-level logger
logger = logging.getLogger(__name__)

DOCKER_HUB_ALIASES = {"docker.io", "index.docker.io", "registry-1.docker.io"}
SUPPORTED_DIGEST_ALGORITHMS = {"sha256", "sha512"}
ARCHITECTURE_ALIASES = {
    "aarch64": "arm64",
    "i386": "386",
    "x86-64": "amd64",
    "x86_64": "amd64",
}
TAG_PATTERN = re.compile(r"[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}")
DIGEST_PATTERN = re.compile(r"([a-z0-9]+(?:[+._-][a-z0-9]+)*):([0-9a-fA-F]+)")
REPOSITORY_COMPONENT = r"[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*"
REPOSITORY_PATTERN = re.compile(rf"{REPOSITORY_COMPONENT}(?:/{REPOSITORY_COMPONENT})*")
REGISTRY_LABEL = r"(?:[A-Za-z0-9]|[A-Za-z0-9][A-Za-z0-9-]*[A-Za-z0-9])"
REGISTRY_HOST_PATTERN = re.compile(rf"{REGISTRY_LABEL}(?:\.{REGISTRY_LABEL})*")
MAX_MANIFEST_SIZE = 4 * 1024 * 1024


class RegistryError(Exception):
    """A registry response or image prevents a safe, complete pull."""


def validate_digest(digest):
    """Return a validated digest algorithm and lowercase encoded value."""
    if not isinstance(digest, str):
        raise RegistryError(f"Invalid descriptor digest: {digest!r}")
    match = DIGEST_PATTERN.fullmatch(digest or "")
    if not match:
        raise RegistryError(f"Invalid descriptor digest: {digest!r}")

    algorithm, encoded = match.groups()
    if algorithm not in SUPPORTED_DIGEST_ALGORITHMS:
        raise RegistryError(f"Unsupported digest algorithm: {algorithm}")
    hasher = hashlib.new(algorithm)

    if len(encoded) != hasher.digest_size * 2:
        raise RegistryError(f"Invalid {algorithm} digest length")
    return algorithm, encoded.lower()


def verify_digest(data, digest, subject="Content"):
    """Verify bytes against a validated descriptor digest."""
    algorithm, expected = validate_digest(digest)
    if hashlib.new(algorithm, data).hexdigest().lower() != expected:
        raise RegistryError(f"{subject} digest mismatch for {digest}")


def normalize_architecture(architecture, variant=None):
    """Return a canonical architecture name for platform comparisons."""
    architecture = architecture.lower() if isinstance(architecture, str) else ""
    variant = variant.lower() if isinstance(variant, str) else ""
    if architecture == "arm" and variant == "v8":
        return "arm64"
    return ARCHITECTURE_ALIASES.get(architecture, architecture)


def normalize_variant(architecture, variant=None):
    """Return the canonical variant used for platform comparisons."""
    variant = variant.lower() if isinstance(variant, str) else ""
    if normalize_architecture(architecture, variant) == "arm64" and variant in (
        "",
        "v8",
    ):
        return ""
    return variant


def get_auth_challenges(headers):
    """Return every WWW-Authenticate field from an HTTP header mapping."""
    values = (
        headers.get_all("WWW-Authenticate") if hasattr(headers, "get_all") else None
    )
    if values:
        return values
    value = headers.get("WWW-Authenticate")
    return [value] if value else []


def parse_auth_challenges(challenges):
    """Parse supported Bearer and Basic challenges, preferring Bearer."""
    if isinstance(challenges, str):
        challenges = [challenges]

    parsed = []
    scheme_pattern = re.compile(r"(bearer|basic)\s+(.+)", re.IGNORECASE)
    for header in challenges or []:
        current_scheme = None
        parameter_items = []
        for item in urllib.request.parse_http_list(header):
            match = scheme_pattern.fullmatch(item)
            if match:
                if current_scheme:
                    try:
                        values = urllib.request.parse_keqv_list(parameter_items)
                    except (IndexError, ValueError) as error:
                        raise ValueError(
                            "Malformed authentication challenge"
                        ) from error
                    parsed.append(
                        (
                            current_scheme,
                            {key.lower(): value for key, value in values.items()},
                        )
                    )
                current_scheme = match.group(1).lower()
                parameter_items = [match.group(2)]
            elif current_scheme:
                parameter_items.append(item)
        if current_scheme:
            try:
                values = urllib.request.parse_keqv_list(parameter_items)
            except (IndexError, ValueError) as error:
                raise ValueError("Malformed authentication challenge") from error
            parsed.append(
                (current_scheme, {key.lower(): value for key, value in values.items()})
            )

    return sorted(parsed, key=lambda item: item[0] != "bearer")


class OriginSafeRedirectHandler(HTTPRedirectHandler):
    """Keep credentials on same-origin redirects and strip them otherwise."""

    @staticmethod
    def _origin(url):
        parsed = urlparse(url)
        default_port = 443 if parsed.scheme.lower() == "https" else 80
        return (
            parsed.scheme.lower(),
            (parsed.hostname or "").lower(),
            parsed.port or default_port,
        )

    @staticmethod
    def _drop_header(request, name):
        """Remove a header case-insensitively from both urllib header stores."""
        lowered = name.lower()
        for store in (request.headers, request.unredirected_hdrs):
            for key in list(store):
                if key.lower() == lowered:
                    del store[key]

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        old_scheme = urlparse(req.full_url).scheme.lower()
        new_scheme = urlparse(newurl).scheme.lower()
        if new_scheme not in ("http", "https"):
            raise HTTPError(
                newurl, code, "Refusing redirect to unsupported scheme", headers, fp
            )
        if old_scheme == "https" and new_scheme != "https":
            raise HTTPError(
                newurl, code, "Refusing HTTPS downgrade redirect", headers, fp
            )

        try:
            old_origin = self._origin(req.full_url)
            new_origin = self._origin(newurl)
        except ValueError as error:
            raise HTTPError(
                newurl,
                code,
                f"Refusing redirect with invalid URL: {error}",
                headers,
                fp,
            ) from error

        redirected = super().redirect_request(req, fp, code, msg, headers, newurl)
        if redirected:
            self._drop_header(redirected, "Proxy-Authorization")
        if redirected and old_origin != new_origin:
            self._drop_header(redirected, "Authorization")
        return redirected


# =============================================================================
# UTILITY FUNCTIONS
# =============================================================================


def parse_image_spec(image_spec, plain_http=False):
    """Return registry URL, repository, manifest reference, load name, and tag."""
    if not image_spec or "://" in image_spec:
        raise ValueError("Image must be in [registry/]repository[:tag] format")

    if image_spec.count("@") > 1:
        raise ValueError("Image must contain at most one digest reference")
    image_with_tag, separator, digest = image_spec.rpartition("@")
    if separator:
        if "/" in digest or ":" not in digest:
            raise ValueError(
                "Image must be in [registry/]repository[:tag][@digest] format"
            )
        algorithm, encoded = validate_digest(digest)
        digest = f"{algorithm}:{encoded}"
    else:
        image_with_tag = image_spec
        digest = None

    last_slash = image_with_tag.rfind("/")
    last_colon = image_with_tag.rfind(":")
    explicit_tag = last_colon > last_slash
    if explicit_tag:
        image_name = image_with_tag[:last_colon]
        tag = image_with_tag[last_colon + 1 :]
    else:
        image_name, tag = image_with_tag, "latest"

    if not image_name or not tag:
        raise ValueError("Image name and tag must not be empty")
    if not TAG_PATTERN.fullmatch(tag):
        raise ValueError(f"Invalid image tag: {tag!r}")

    first, separator, remainder = image_name.partition("/")
    has_registry = bool(separator) and (
        "." in first or ":" in first or first.lower() == "localhost"
    )
    registry = first if has_registry else "docker.io"
    repository = remainder if has_registry else image_name

    if has_registry:
        if any(character.isspace() for character in registry):
            raise ValueError(f"Invalid registry host: {registry!r}")
        try:
            parsed_registry = urlparse(f"//{registry}")
            if registry.endswith(":") or parsed_registry.port == 0:
                raise ValueError("port must be between 1 and 65535")
        except ValueError as error:
            raise ValueError(f"Invalid registry host {registry!r}: {error}") from error
        registry_hostname = parsed_registry.hostname
        if (
            not registry_hostname
            or parsed_registry.username is not None
            or parsed_registry.password is not None
            or parsed_registry.path
            or parsed_registry.query
            or parsed_registry.fragment
        ):
            raise ValueError(f"Invalid registry host: {registry!r}")
        if ":" in registry_hostname:
            try:
                ipaddress.IPv6Address(registry_hostname)
            except ValueError as error:
                raise ValueError(f"Invalid registry host: {registry!r}") from error
        elif (
            len(registry_hostname) > 253
            or not REGISTRY_HOST_PATTERN.fullmatch(registry_hostname)
            or any(len(label) > 63 for label in registry_hostname.split("."))
        ):
            raise ValueError(f"Invalid registry host: {registry!r}")

    if not repository:
        raise ValueError("Image repository must not be empty")
    load_name = f"{registry}/{repository}" if has_registry else repository
    if registry.lower() in DOCKER_HUB_ALIASES and "/" not in repository:
        repository = f"library/{repository}"
    if repository.lower() != repository:
        raise ValueError(f"Image repository must be lowercase: {repository!r}")
    if len(repository) > 255 or not REPOSITORY_PATTERN.fullmatch(repository):
        raise ValueError(f"Invalid image repository: {repository!r}")

    registry_host = (
        "registry-1.docker.io" if registry.lower() in DOCKER_HUB_ALIASES else registry
    )
    scheme = "http" if plain_http else "https"
    reference = digest or tag
    archive_tag = tag if not digest or explicit_tag else None
    return (
        f"{scheme}://{registry_host}",
        repository,
        reference,
        load_name,
        archive_tag,
    )


def setup_logging(level=logging.INFO):
    """Configure logging for the application."""
    logging.basicConfig(
        level=level,
        format="%(message)s",  # Clean format for CLI tool
        handlers=[logging.StreamHandler(sys.stdout)],
    )
    logger.setLevel(level)


def format_bytes(size):
    """Format bytes into a human-readable string."""
    for unit in ["B", "KB", "MB", "GB"]:
        if size < 1024.0:
            return f"{int(size)} {unit}" if unit == "B" else f"{size:.1f} {unit}"
        size /= 1024.0
    return f"{size:.1f} TB"


def read_limited(response, limit, subject):
    """Read at most limit bytes from a small registry response."""
    body = response.read(limit + 1)
    if len(body) > limit:
        raise RegistryError(f"{subject} exceeds the {format_bytes(limit)} limit")
    return body


def log_http_error_body(error):
    """Log a bounded, escaped registry error response."""
    try:
        body = error.read(4096).decode("utf-8", "replace")
    except (AttributeError, OSError):
        return
    if body:
        logger.error("Error details: %r", body[:500])


def apply_default_file_mode(path):
    """Restore normal umask-derived permissions after mkstemp on POSIX."""
    if os.name == "posix":
        current_umask = os.umask(0)
        os.umask(current_umask)
        os.chmod(path, 0o666 & ~current_umask)


# =============================================================================
# CONFIGURATION CLASSES
# =============================================================================


class Config:
    """Configuration management for Docker Image Puller"""

    def __init__(self, proxy_config=None, timeout_config=None):
        # Proxy configuration
        self.proxy_config = self._validate_proxy_config(dict(proxy_config or {}))

        # Timeout configuration
        timeout_config = timeout_config or {}
        self.request_timeout = timeout_config.get("request_timeout", 30)
        self.download_timeout = timeout_config.get("download_timeout", 300)
        self.chunk_timeout = timeout_config.get("chunk_timeout", 60)

        # Validate configuration
        self._validate_config()

    def _validate_proxy_config(self, proxy_config):
        """Validate and normalize proxy configuration.

        Merges provided proxy settings with environment variables.
        Environment variables are used as fallback when not explicitly provided.

        Args:
            proxy_config (dict): User-provided proxy configuration

        Returns:
            dict: Validated and normalized proxy configuration
        """
        # Merge environment proxy settings
        if not proxy_config.get("http_proxy"):
            proxy_config["http_proxy"] = os.environ.get("HTTP_PROXY") or os.environ.get(
                "http_proxy"
            )

        if not proxy_config.get("https_proxy"):
            proxy_config["https_proxy"] = os.environ.get(
                "HTTPS_PROXY"
            ) or os.environ.get("https_proxy")

        if not proxy_config.get("no_proxy"):
            proxy_config["no_proxy"] = os.environ.get("NO_PROXY") or os.environ.get(
                "no_proxy"
            )

        for name in ("http_proxy", "https_proxy"):
            url = proxy_config.get(name)
            if not url:
                continue
            if not isinstance(url, str):
                raise ValueError(f"{name} must be a URL")
            try:
                parsed = urlparse(url)
                if parsed.netloc.endswith(":") or parsed.port == 0:
                    raise ValueError("port must be between 1 and 65535")
                hostname = parsed.hostname
            except ValueError as error:
                raise ValueError(f"Invalid {name} URL: {error}") from error
            if parsed.scheme not in ("http", "https") or not hostname:
                raise ValueError(
                    f"Invalid {name} URL: expected HTTP(S) with a hostname"
                )

        return proxy_config

    def _validate_config(self):
        """Validate configuration values"""
        for name in ("request_timeout", "download_timeout"):
            value = getattr(self, name)
            if (
                not isinstance(value, (int, float))
                or isinstance(value, bool)
                or value <= 0
            ):
                raise ValueError(f"{name} must be a positive number")
        if (
            not isinstance(self.chunk_timeout, int)
            or isinstance(self.chunk_timeout, bool)
            or self.chunk_timeout <= 0
        ):
            raise ValueError("chunk_timeout must be a positive integer")

    def has_proxy(self):
        """Check if proxy configuration is present"""
        return bool(
            self.proxy_config.get("http_proxy") or self.proxy_config.get("https_proxy")
        )


# =============================================================================
# SUPPORT CLASSES
# =============================================================================


class ProxyManager:
    """Handles proxy configuration and URL sanitization"""

    def __init__(self, config):
        self.config = config
        self.opener = None
        self.direct_opener = None
        self.setup_proxy()

    def _build_opener(self, proxy_handlers):
        handlers = [ProxyHandler(proxy_handlers), OriginSafeRedirectHandler()]
        if self.config.proxy_config.get("insecure"):
            context = ssl.create_default_context()
            context.check_hostname = False
            context.verify_mode = ssl.CERT_NONE
            handlers.append(urllib.request.HTTPSHandler(context=context))
        return build_opener(*handlers)

    def setup_proxy(self):
        """Build per-instance proxied and direct urllib openers."""
        if self.config.proxy_config.get("insecure"):
            logger.warning("SSL certificate verification is disabled")

        if not self.config.has_proxy():
            self._setup_no_proxy()
            return

        proxy_handlers = {}

        logger.info("Using proxy configuration:")

        http_proxy = self.config.proxy_config.get("http_proxy")
        https_proxy = self.config.proxy_config.get("https_proxy")

        if http_proxy:
            if self.config.proxy_config.get("proxy_auth"):
                http_proxy = self._add_proxy_auth(
                    http_proxy, self.config.proxy_config["proxy_auth"]
                )
            proxy_handlers["http"] = http_proxy
            logger.info(f"  HTTP Proxy: {self.sanitize_proxy_url(http_proxy)}")

        if https_proxy:
            if self.config.proxy_config.get("proxy_auth"):
                https_proxy = self._add_proxy_auth(
                    https_proxy, self.config.proxy_config["proxy_auth"]
                )
            proxy_handlers["https"] = https_proxy
            logger.info(f"  HTTPS Proxy: {self.sanitize_proxy_url(https_proxy)}")

        if self.config.proxy_config.get("no_proxy"):
            logger.info(f"  No Proxy: {self.config.proxy_config.get('no_proxy')}")

        self.opener = self._build_opener(proxy_handlers)
        self.direct_opener = self._build_opener({})

    def _setup_no_proxy(self):
        """Build an opener that explicitly ignores environment proxies."""
        self.opener = self._build_opener({})
        self.direct_opener = self.opener

    def open(self, request, timeout, bypass_proxy=False):
        """Open a request through this puller's configured network path."""
        opener = self.direct_opener if bypass_proxy else self.opener
        if opener is None:
            raise RuntimeError("Proxy manager is not initialized")
        return opener.open(request, timeout=timeout)

    def _add_proxy_auth(self, proxy_url, auth_string):
        """Add authentication credentials to proxy URL.

        Args:
            proxy_url (str): Base proxy URL
            auth_string (str): Authentication in format "username:password"

        Returns:
            str: Proxy URL with embedded authentication credentials
        """
        parsed = urlparse(proxy_url)
        if parsed.username is not None:
            return proxy_url
        if ":" in auth_string:
            username, password = auth_string.split(":", 1)
        else:
            logger.warning("Proxy authentication must be in format 'username:password'")
            return proxy_url

        credentials = f"{quote(username, safe='')}:{quote(password, safe='')}"
        return parsed._replace(netloc=f"{credentials}@{parsed.netloc}").geturl()

    def sanitize_proxy_url(self, url):
        """Remove credentials from proxy URL for display"""
        if not url:
            return url

        try:
            parsed = urlparse(url)
            if parsed.username or parsed.password:
                netloc = parsed.netloc.rpartition("@")[2]
                return parsed._replace(netloc=netloc).geturl()
        except (ValueError, TypeError, AttributeError):
            return self._mask_credentials_fallback(url)

        return url

    def _mask_credentials_fallback(self, text):
        """Fallback credential masking for malformed URLs or general text"""
        if not text:
            return text

        patterns = [
            r"://[^:/@]+:[^@]+@",  # ://user:pass@
            r"://[^:/@]+@",  # ://user@
        ]

        result = text
        for pattern in patterns:
            result = re.sub(pattern, "://***:***@", result)

        return result

    def sanitize_debug_output(self, text):
        """Sanitize any text that might contain credentials for debug output"""
        if not text:
            return text

        if isinstance(text, dict):
            return {
                key: (
                    "***"
                    if key.lower() in ("authorization", "proxy-authorization")
                    else value
                )
                for key, value in text.items()
            }

        return self._mask_credentials_fallback(str(text))

    def should_bypass_proxy(self, host):
        """Check if a hostname or host:port should bypass the proxy."""
        no_proxy = self.config.proxy_config.get("no_proxy")
        return bool(
            no_proxy
            and urllib.request.proxy_bypass_environment(  # pyright: ignore[reportAttributeAccessIssue]
                host, {"no": no_proxy}
            )
        )


class ProgressReporter:
    """Terminal-safe progress reporting with speed and ETA calculation."""

    def __init__(self, total_size=None, description="Download", show_speed=True):
        """Initialize progress reporter.

        Args:
            total_size (int, optional): Total expected size in bytes
            description (str): Description to show before progress bar
            show_speed (bool): Whether to show download speed and ETA
        """
        self.total_size = total_size
        self.downloaded = 0
        self.description = description
        self.show_speed = show_speed
        self.start_time = time.time()
        self.last_update = self.start_time

        # Terminal width detection
        self.terminal_width = self._get_terminal_width()

    def _get_terminal_width(self):
        """Get terminal width, default to 80 if detection fails"""
        try:
            return shutil.get_terminal_size().columns
        except (OSError, AttributeError):
            return 80

    def update(self, bytes_downloaded):
        """Update progress with new bytes downloaded.

        Args:
            bytes_downloaded (int): Additional bytes downloaded since last update
        """
        if bytes_downloaded <= 0:
            return

        self.downloaded += bytes_downloaded
        current_time = time.time()

        # Throttle display updates
        if current_time - self.last_update < 0.1 and self.downloaded < (
            self.total_size or float("inf")
        ):
            return

        self._display_progress()
        self.last_update = current_time

    def _display_progress(self):
        """Display current progress bar and stats"""
        if not logger.isEnabledFor(logging.INFO):
            return

        # Calculate progress percentage
        if self.total_size:
            progress_pct = min(100.0, (self.downloaded / self.total_size) * 100)
        else:
            progress_pct = 0

        # Format downloaded amount
        downloaded_str = format_bytes(self.downloaded)

        # Build progress bar
        if self.total_size:
            total_str = format_bytes(self.total_size)
            size_info = f"{downloaded_str}/{total_str}"
            progress_bar = self._build_progress_bar(progress_pct)
        else:
            size_info = downloaded_str
            progress_bar = f"[{'#' * 10}] ???%"

        # Calculate speed and ETA
        speed_info = ""
        if self.show_speed:
            elapsed = time.time() - self.start_time
            if elapsed > 0:
                speed = self.downloaded / elapsed
                speed_str = f"{format_bytes(speed)}/s"

                if self.total_size and speed > 0:
                    remaining = (self.total_size - self.downloaded) / speed
                    eta_str = self._format_duration(remaining)
                    speed_info = f" | {speed_str} | ETA: {eta_str}"
                else:
                    speed_info = f" | {speed_str}"

        # Build complete progress line
        progress_line = f"  {self.description}: {progress_bar} {size_info}{speed_info}"

        # Truncate to terminal width if needed
        if len(progress_line) > self.terminal_width:
            available = (
                self.terminal_width
                - len(f"  {self.description}: ")
                - len(size_info)
                - len(speed_info)
                - 3
            )
            if available > 10:  # Minimum width required
                progress_bar = self._build_progress_bar(progress_pct, available)
                progress_line = (
                    f"  {self.description}: {progress_bar} {size_info}{speed_info}"
                )
            else:
                # Minimal display for narrow terminals
                progress_line = f"  {downloaded_str} {int(progress_pct)}%"

        # Print with carriage return for overwrite
        print(f"\r{progress_line}", end="", flush=True)

    def _build_progress_bar(self, progress_pct, width=30):
        """Build a terminal-safe progress bar.

        Args:
            progress_pct (float): Progress percentage (0-100)
            width (int): Width of progress bar in characters

        Returns:
            str: Formatted progress bar
        """
        filled_width = int(width * progress_pct / 100)
        return (
            f"[{'#' * filled_width}{'-' * (width - filled_width)}] {progress_pct:5.1f}%"
        )

    def _format_duration(self, seconds):
        """Format duration into human readable string.

        Args:
            seconds (float): Duration in seconds

        Returns:
            str: Formatted duration string
        """
        if seconds < 60:
            return f"{int(seconds)}s"
        elif seconds < 3600:
            minutes = int(seconds / 60)
            secs = int(seconds % 60)
            return f"{minutes}m{secs:02d}s"
        else:
            hours = int(seconds / 3600)
            minutes = int((seconds % 3600) / 60)
            return f"{hours}h{minutes:02d}m"

    def finish(self):
        """Complete the progress display with a newline"""
        if self.downloaded > 0 and logger.isEnabledFor(logging.INFO):
            print()  # New line to finish progress display


# =============================================================================
# CORE FUNCTIONALITY
# =============================================================================


class DockerImagePuller:
    def __init__(
        self,
        auth_token=None,
        proxy_config=None,
        timeout_config=None,
        username=None,
        password=None,
    ):
        if (username is None) != (password is None):
            raise ValueError("username and password must be provided together")

        config = Config(proxy_config, timeout_config)
        self.registry_url = "https://registry-1.docker.io"
        self.auth_token = auth_token
        self.username = username
        self.password = password
        self.authorization = f"Bearer {auth_token}" if auth_token else None
        self.request_timeout = config.request_timeout
        self.download_timeout = config.download_timeout
        self.chunk_timeout = config.chunk_timeout
        self.proxy_manager = ProxyManager(config)

    def _stream_download(self, response, digest, expected_size=None, destination=None):
        """Download, validate, and optionally retain a blob on disk.

        Args:
            response: HTTP response object to stream from
            digest (str): Blob digest for identification in error messages
            expected_size (int, optional): Expected download size in bytes
            destination (str, optional): Final path for a disk-backed blob

        Returns:
            bytes or str: Validated data or destination path; None on failure
        """

        # Setup timeout handling for stuck downloads
        def timeout_handler(_signum, _frame):
            raise TimeoutError(f"Download chunk timeout after {self.chunk_timeout}s")

        algorithm, expected_digest = validate_digest(digest)
        temp_data = (
            open(destination + ".part", "w+b")  # noqa: SIM115
            if destination
            else tempfile.NamedTemporaryFile(delete=False)  # noqa: SIM115
        )
        total_size = 0
        chunk_size = 65536  # 64KB chunks
        hasher = hashlib.new(algorithm)
        alarm_signal = getattr(signal, "SIGALRM", None)
        set_alarm = getattr(signal, "alarm", None)
        if threading.current_thread() is not threading.main_thread():
            alarm_signal = set_alarm = None
        previous_alarm_handler = None
        last_fallback_update = 0.0

        # Initialize progress reporter for files > 1MB
        progress_reporter = None
        if expected_size and expected_size > 1024 * 1024:  # Over 1MB
            blob_name = digest[:12] + "..." if len(digest) > 12 else digest
            progress_reporter = ProgressReporter(expected_size, f"Layer {blob_name}")

        try:
            if alarm_signal is not None and set_alarm is not None:
                previous_alarm_handler = signal.signal(alarm_signal, timeout_handler)

            while True:
                if set_alarm is not None:
                    set_alarm(self.chunk_timeout)

                try:
                    chunk = response.read(chunk_size)
                    if not chunk:
                        break

                    chunk_len = len(chunk)
                    if (
                        expected_size is not None
                        and total_size + chunk_len > expected_size
                    ):
                        raise RegistryError(
                            f"Blob {digest} exceeds its descriptor size of "
                            f"{expected_size} bytes"
                        )
                    temp_data.write(chunk)
                    hasher.update(chunk)
                    total_size += chunk_len

                    # Update progress reporter
                    if progress_reporter:
                        progress_reporter.update(chunk_len)
                    elif total_size > 1024 * 1024 and logger.isEnabledFor(logging.INFO):
                        current_time = time.monotonic()
                        if current_time - last_fallback_update >= 0.1:
                            mb_downloaded = total_size / (1024 * 1024)
                            print(f"  Downloaded {mb_downloaded:.1f} MB...", end="\r")
                            last_fallback_update = current_time

                except TimeoutError as error:
                    raise TimeoutError("Download chunk timeout") from error
                finally:
                    if set_alarm is not None:
                        set_alarm(0)  # Cancel timeout

            # Finish progress reporting
            if progress_reporter:
                progress_reporter.finish()
            elif total_size > 1024 * 1024 and logger.isEnabledFor(logging.INFO):
                print()  # New line after fallback progress

            temp_data.close()
            if expected_size is not None and total_size != expected_size:
                logger.error(
                    "Blob %s size mismatch: expected %s bytes, received %s",
                    digest,
                    expected_size,
                    total_size,
                )
                return None
            if hasher.hexdigest().lower() != expected_digest:
                logger.error(f"Blob digest verification failed for {digest}")
                return None

            if destination:
                os.replace(temp_data.name, destination)
                return destination

            with open(temp_data.name, "rb") as f:
                return f.read()

        except TimeoutError as e:
            if progress_reporter:
                progress_reporter.finish()
            logger.error(f"Download timeout for blob {digest}: {e}")
            return None
        except Exception as e:  # noqa: BLE001
            if progress_reporter:
                progress_reporter.finish()
            logger.error(f"Streaming error for blob {digest}: {e}")
            return None
        finally:
            # Cleanup
            if alarm_signal is not None and set_alarm is not None:
                set_alarm(0)
                if previous_alarm_handler is not None:
                    signal.signal(alarm_signal, previous_alarm_handler)
            temp_data.close()
            try:
                os.unlink(temp_data.name)
            except OSError:
                pass

    def make_request(self, url, headers=None):
        """Make HTTP request with proxy handling"""
        req_headers = headers or {}

        logger.debug(f"Request URL: {self.proxy_manager.sanitize_debug_output(url)}")
        logger.debug(
            f"Headers: {self.proxy_manager.sanitize_debug_output(req_headers)}"
        )

        parsed_url = urlparse(url)
        bypass_proxy = self.proxy_manager.should_bypass_proxy(parsed_url.netloc)
        if bypass_proxy:
            logger.debug(f"Bypassing proxy for {parsed_url.hostname}")

        request = Request(url, headers=req_headers)
        response = self.proxy_manager.open(
            request, self.request_timeout, bypass_proxy=bypass_proxy
        )
        logger.debug(f"Response code: {response.code}")
        return response

    def get_auth_header(self, image_name, challenge=None):
        """Negotiate the registry's advertised authentication scheme."""
        if self.auth_token:
            return self.authorization

        try:
            challenge_has_repository_scope = challenge is not None
            if challenge is None:
                try:
                    with self.make_request(f"{self.registry_url}/v2/"):
                        return None
                except HTTPError as error:
                    if error.code != 401:
                        return None
                    challenge = get_auth_challenges(error.headers)

            if not challenge:
                raise RegistryError(
                    "Registry did not provide an authentication challenge"
                )

            parsed_challenges = parse_auth_challenges(challenge)
            if not parsed_challenges:
                raise RegistryError(
                    "Registry did not provide a supported authentication challenge"
                )
            scheme, values = parsed_challenges[0]
            credentials = None
            if self.username is not None:
                encoded = base64.b64encode(
                    f"{self.username}:{self.password}".encode()
                ).decode()
                credentials = f"Basic {encoded}"

            if scheme == "basic":
                if not credentials:
                    raise RegistryError("Registry requires --username and --password")
                self.authorization = credentials
                return credentials

            realm = values.get("realm")
            if not realm:
                raise RegistryError("Registry authentication challenge has no realm")

            registry = urlparse(self.registry_url)
            auth_server = urlparse(realm)
            if auth_server.scheme not in ("http", "https") or not auth_server.hostname:
                raise RegistryError(
                    "Registry authentication realm must be an absolute URL"
                )
            if auth_server.scheme != "https" and registry.scheme != "http":
                raise RegistryError(
                    "Refusing an insecure authentication realm for an HTTPS registry"
                )
            if credentials and auth_server.hostname != registry.hostname:
                logger.warning(
                    "Registry credentials are being sent to authentication host %s",
                    auth_server.hostname,
                )

            params = {
                "service": values.get("service"),
                "scope": (
                    values.get("scope") if challenge_has_repository_scope else None
                )
                or f"repository:{image_name}:pull",
            }
            query = urlencode({key: value for key, value in params.items() if value})
            separator = "&" if "?" in realm else "?"
            headers = {"Authorization": credentials} if credentials else None
            with self.make_request(f"{realm}{separator}{query}", headers) as response:
                data = json.loads(
                    read_limited(response, MAX_MANIFEST_SIZE, "Registry token response")
                )
                if not isinstance(data, dict):
                    raise RegistryError("Registry token response is not a JSON object")
                token = data.get("token") or data.get("access_token")
                if not isinstance(token, str) or not token:
                    raise RegistryError(
                        "Registry token response did not contain a token"
                    )
                if "\r" in token or "\n" in token:
                    raise RegistryError(
                        "Registry token response contained a multi-line token"
                    )
                self.authorization = f"Bearer {token}"
                return self.authorization
        except RegistryError:
            raise
        except (
            HTTPError,
            URLError,
            json.JSONDecodeError,
            UnicodeDecodeError,
            ValueError,
        ) as error:
            raise RegistryError(
                f"Error authenticating with registry: {error}"
            ) from error

    def get_manifest(
        self,
        image_name,
        tag,
        authorization,
        architecture="amd64",
        os_type="linux",
        variant=None,
        retry_with_new_auth=True,
    ):
        """Get image manifest from registry, handling multi-arch manifest lists"""
        if DIGEST_PATTERN.fullmatch(tag or ""):
            algorithm, encoded = validate_digest(tag)
            tag = f"{algorithm}:{encoded}"
        url = f"{self.registry_url}/v2/{image_name}/manifests/{tag}"

        # First, try to get manifest list (for multi-arch images)
        headers = {
            "Accept": "application/vnd.docker.distribution.manifest.list.v2+json,application/vnd.docker.distribution.manifest.v2+json,application/vnd.oci.image.index.v1+json,application/vnd.oci.image.manifest.v1+json",
        }
        if authorization:
            headers["Authorization"] = authorization

        try:
            with self.make_request(url, headers) as response:
                manifest_body = read_limited(
                    response, MAX_MANIFEST_SIZE, "Manifest response"
                )
                if DIGEST_PATTERN.fullmatch(tag or ""):
                    verify_digest(manifest_body, tag, "Manifest")
                manifest_data = json.loads(manifest_body)
                if not isinstance(manifest_data, dict):
                    raise RegistryError("Manifest response is not a JSON object")

                # Check if this is a manifest list (multi-arch)
                if "manifests" in manifest_data:
                    manifests = manifest_data["manifests"]
                    if not isinstance(manifests, list):
                        raise RegistryError("Image index manifests must be a list")
                    logger.info(
                        "Multi-architecture image detected. Available platforms:"
                    )

                    # List available platforms
                    valid_manifests = []
                    malformed_manifests = 0
                    for m in manifests:
                        if not isinstance(m, dict):
                            malformed_manifests += 1
                            continue
                        platform = m.get("platform", {})
                        if not isinstance(platform, dict):
                            malformed_manifests += 1
                            continue
                        arch = platform.get("architecture", "unknown")
                        platform_os = platform.get("os", "unknown")
                        entry_variant = platform.get("variant", "")

                        # Skip invalid entries
                        if (
                            not isinstance(arch, str)
                            or not isinstance(platform_os, str)
                            or (
                                entry_variant is not None
                                and not isinstance(entry_variant, str)
                            )
                            or arch == "unknown"
                            or platform_os == "unknown"
                        ):
                            malformed_manifests += 1
                            continue

                        valid_manifests.append(m)
                        variant_str = f"-{entry_variant}" if entry_variant else ""
                        logger.info("  - %r", f"{platform_os}/{arch}{variant_str}")

                    # Find the requested platform using canonical architecture names.
                    selected_manifest = None
                    requested_architecture = normalize_architecture(
                        architecture, variant
                    )
                    requested_variant = normalize_variant(architecture, variant)
                    requested_os = os_type.lower()
                    for m in valid_manifests:
                        platform = m.get("platform", {})
                        platform_variant = normalize_variant(
                            platform.get("architecture"), platform.get("variant")
                        )
                        if (
                            normalize_architecture(
                                platform.get("architecture"), platform.get("variant")
                            )
                            == requested_architecture
                            and platform.get("os").lower() == requested_os
                            and (
                                variant is None or platform_variant == requested_variant
                            )
                        ):
                            selected_manifest = m
                            break

                    if not selected_manifest:
                        available = ", ".join(
                            repr(
                                f"{item.get('platform', {}).get('os', 'unknown')}/"
                                f"{item.get('platform', {}).get('architecture', 'unknown')}"
                            )
                            for item in valid_manifests
                        )
                        raise RegistryError(
                            f"No manifest found for {os_type}/{architecture}"
                            + (f"/{variant}" if variant else "")
                            + "; "
                            f"available platforms: {available or 'none'}"
                            + (
                                f"; {malformed_manifests} malformed "
                                f"entr{'y' if malformed_manifests == 1 else 'ies'} omitted"
                                if malformed_manifests
                                else ""
                            )
                        )

                    platform = selected_manifest.get("platform", {})
                    logger.info(
                        "Selected platform: %r",
                        f"{platform.get('os')}/{platform.get('architecture')}",
                    )

                    # Now fetch the specific manifest using its digest
                    specific_digest = selected_manifest.get("digest")
                    if not specific_digest:
                        raise RegistryError("Selected manifest is missing a digest")
                    algorithm, encoded = validate_digest(specific_digest)
                    specific_digest = f"{algorithm}:{encoded}"
                    url = f"{self.registry_url}/v2/{image_name}/manifests/{specific_digest}"
                    expected_size = selected_manifest.get("size")
                    if expected_size is not None and (
                        not isinstance(expected_size, int)
                        or isinstance(expected_size, bool)
                        or expected_size < 0
                    ):
                        raise RegistryError("Selected manifest has an invalid size")

                    headers = {
                        "Accept": "application/vnd.docker.distribution.manifest.v2+json,application/vnd.oci.image.manifest.v1+json",
                    }
                    if authorization:
                        headers["Authorization"] = authorization

                    with self.make_request(url, headers) as response:
                        response_limit = (
                            min(expected_size, MAX_MANIFEST_SIZE)
                            if expected_size is not None
                            else MAX_MANIFEST_SIZE
                        )
                        specific_body = read_limited(
                            response, response_limit, "Selected manifest response"
                        )
                        if (
                            expected_size is not None
                            and len(specific_body) != expected_size
                        ):
                            raise RegistryError(
                                f"Manifest size mismatch for {specific_digest}: "
                                f"expected {expected_size} bytes, "
                                f"received {len(specific_body)}"
                            )
                        verify_digest(specific_body, specific_digest, "Manifest")
                        specific_manifest = json.loads(specific_body)
                        if not isinstance(specific_manifest, dict):
                            raise RegistryError(
                                "Selected manifest response is not a JSON object"
                            )

                        if "manifests" in specific_manifest:
                            raise RegistryError(
                                "Nested image indexes are not supported"
                            )

                        manifest_data = specific_manifest

                if manifest_data.get("schemaVersion") in (1, "1"):
                    raise RegistryError("Schema v1 images are not supported")

                media_type = manifest_data.get("mediaType")
                if media_type is not None and not isinstance(media_type, str):
                    raise RegistryError("Manifest has an invalid media type")

                # It's already a regular manifest
                if "config" in manifest_data or "mediaType" in manifest_data:
                    if media_type and "oci" in media_type:
                        logger.info("  (OCI format image)")
                    return manifest_data

                raise RegistryError(
                    f"Unknown manifest format. Keys found: {list(manifest_data.keys())}"
                )

        except HTTPError as e:
            if e.code == 401 and retry_with_new_auth:
                new_authorization = self.get_auth_header(
                    image_name, get_auth_challenges(e.headers)
                )
                return self.get_manifest(
                    image_name,
                    tag,
                    new_authorization,
                    architecture,
                    os_type,
                    variant,
                    retry_with_new_auth=False,
                )
            elif e.code == 404:
                logger.error(f"Image {image_name}:{tag} not found")
            else:
                logger.error(f"Error getting manifest: {e}")
                log_http_error_body(e)
            raise RegistryError(
                f"Unable to retrieve manifest for {image_name}:{tag}"
            ) from e
        except (json.JSONDecodeError, UnicodeDecodeError) as e:
            raise RegistryError(f"Error parsing manifest response: {e}") from e

    def download_blob(
        self,
        image_name,
        digest,
        authorization,
        retry_with_new_auth=True,
        expected_size=None,
        destination=None,
    ):
        """Download a blob (layer) from Docker registry.

        Handles authentication and origin-safe redirects.
        Automatically retries with fresh token on 401 errors.

        Args:
            image_name (str): Repository name (e.g., 'library/alpine')
            digest (str): SHA256 digest of the blob to download
            authorization (str): Authorization header value, if required
            retry_with_new_auth (bool): Retry once with fresh authentication on 401
            expected_size (int, optional): Descriptor size to verify
            destination (str, optional): Keep the validated blob at this path

        Returns:
            bytes or str: Blob data or destination path; None if download failed

        Raises:
            Various network and HTTP errors are caught and logged
        """
        algorithm, encoded = validate_digest(digest)
        digest = f"{algorithm}:{encoded}"
        url = f"{self.registry_url}/v2/{image_name}/blobs/{digest}"

        headers = {
            "Accept": "application/vnd.docker.image.rootfs.diff.tar.gzip,application/octet-stream,*/*",
        }
        if authorization:
            headers["Authorization"] = authorization

        try:
            request = Request(url, headers=headers)
            logger.debug(f"Downloading blob from: {url}")
            registry_host = urlparse(url).netloc
            with self.proxy_manager.open(
                request,
                self.download_timeout,
                bypass_proxy=self.proxy_manager.should_bypass_proxy(registry_host),
            ) as response:
                final_url = response.geturl()
                if final_url != url:
                    redirect_url = (
                        urlparse(final_url)._replace(query="", fragment="").geturl()
                    )
                    logger.debug(
                        "Followed redirect to: %s",
                        self.proxy_manager.sanitize_debug_output(redirect_url),
                    )

                content_length = response.headers.get("Content-Length")
                try:
                    response_size = int(content_length) if content_length else None
                    if response_size is not None and response_size < 0:
                        raise ValueError
                except (TypeError, ValueError):
                    logger.debug("Ignoring invalid Content-Length for blob %s", digest)
                    response_size = None
                if (
                    expected_size is not None
                    and response_size is not None
                    and expected_size != response_size
                ):
                    logger.error(
                        "Blob %s size mismatch: descriptor says %s bytes, "
                        "response says %s",
                        digest,
                        expected_size,
                        response_size,
                    )
                    return None

                verified_size = (
                    expected_size if expected_size is not None else response_size
                )
                return self._stream_download(
                    response, digest, verified_size, destination=destination
                )

        except HTTPError as e:
            if e.code == 401 and retry_with_new_auth:
                # Token might not have the right scope, get a new one
                logger.debug("Got 401, retrying with new token...")
                logger.info("Authorization failed, requesting new token...")
                new_authorization = self.get_auth_header(
                    image_name, get_auth_challenges(e.headers)
                )
                return self.download_blob(
                    image_name,
                    digest,
                    new_authorization,
                    retry_with_new_auth=False,
                    expected_size=expected_size,
                    destination=destination,
                )

            logger.error(f"Error downloading blob {digest}: HTTP {e.code} - {e.reason}")

            log_http_error_body(e)

            logger.debug(
                "Request headers were: %s",
                self.proxy_manager.sanitize_debug_output(headers),
            )

            return None

        except (URLError, OSError) as e:
            logger.error(f"Network error downloading blob {digest}: {e}")
            logger.debug("Exception traceback:", exc_info=True)
            return None
        except Exception as e:
            logger.error(f"Unexpected error downloading blob {digest}: {e}")
            logger.debug("Exception traceback:", exc_info=True)
            return None

    def create_docker_tar(
        self,
        image_name,
        tag,
        manifest,
        config_blob,
        config_data,
        layers,
        output_file,
    ):
        """Create a Docker-compatible tar file from downloaded components.

        Assembles the manifest, config, and layers into a standard Docker tar
        format that can be loaded with 'docker load'.

        Args:
            image_name (str): Full image name (e.g., 'library/alpine')
            tag (str): Image tag (e.g., 'latest')
            manifest (dict): Image manifest metadata
            config_blob (bytes): Image configuration blob
            config_data (dict): Parsed image configuration
            layers (list): Layer dictionaries with digest, size, and path
            output_file (str): Path where to save the tar file
        Creates:
            A tar file containing:
            - manifest.json: Docker format manifest
            - {config_digest}.json: Image configuration
            - {layer_digest}/layer.tar: Uncompressed image layers
        """

        rootfs = config_data.get("rootfs")
        diff_ids = rootfs.get("diff_ids") if isinstance(rootfs, dict) else None
        if not isinstance(diff_ids, list) or len(diff_ids) != len(layers):
            raise RegistryError(
                "Image configuration layer diff-ID count does not match the manifest"
            )

        normalized_diff_ids = []
        for diff_id in diff_ids:
            algorithm, encoded = validate_digest(diff_id)
            normalized_diff_ids.append((algorithm, encoded))

        # Create temporary directory for building tar
        with tempfile.TemporaryDirectory() as tmpdir:
            # Save config JSON
            _algorithm, config_digest = validate_digest(manifest["config"]["digest"])
            config_file = f"{config_digest}.json"
            config_path = os.path.join(tmpdir, config_file)

            with open(config_path, "wb") as f:
                f.write(config_blob)

            # Save layers
            layer_files = []
            written_diff_ids = {}
            for layer_info, (diff_algorithm, expected_diff_id) in zip(
                layers, normalized_diff_ids, strict=True
            ):
                _algorithm, layer_digest = validate_digest(layer_info["digest"])
                layer_files.append(layer_digest)
                if layer_digest in written_diff_ids:
                    if written_diff_ids[layer_digest] != (
                        diff_algorithm,
                        expected_diff_id,
                    ):
                        raise RegistryError(
                            f"Layer {layer_info['digest']} has inconsistent diff-IDs"
                        )
                    continue

                layer_dir = os.path.join(tmpdir, layer_digest)
                os.makedirs(layer_dir, exist_ok=True)

                layer_tar_path = os.path.join(layer_dir, "layer.tar")
                try:
                    with (
                        open(layer_info["path"], "rb") as source,
                        open(layer_tar_path, "wb") as layer_file,
                    ):
                        magic = source.read(4)
                        source.seek(0)
                        if magic.startswith(b"\x1f\x8b"):
                            with gzip.GzipFile(fileobj=source) as decompressed:
                                shutil.copyfileobj(decompressed, layer_file)
                        elif magic == b"\x28\xb5\x2f\xfd":
                            raise RegistryError(
                                "Zstandard-compressed OCI layers are not supported"
                            )
                        else:
                            shutil.copyfileobj(source, layer_file)
                except (OSError, EOFError, zlib.error) as error:
                    raise RegistryError(
                        f"Unable to decompress layer {layer_info['digest']}"
                    ) from error

                with open(layer_tar_path, "rb") as layer_file:
                    actual_diff_id = hashlib.file_digest(
                        layer_file, diff_algorithm
                    ).hexdigest()
                if actual_diff_id != expected_diff_id:
                    raise RegistryError(
                        f"Layer diff-ID mismatch for {layer_info['digest']}"
                    )
                written_diff_ids[layer_digest] = (diff_algorithm, expected_diff_id)

            # Create manifest.json
            manifest_json = [
                {
                    "Config": config_file,
                    "RepoTags": [f"{image_name}:{tag}"] if tag else None,
                    "Layers": [f"{lf}/layer.tar" for lf in layer_files],
                }
            ]

            manifest_path = os.path.join(tmpdir, "manifest.json")
            with open(manifest_path, "w") as f:
                json.dump(manifest_json, f)

            # Create the tar file
            with tarfile.open(output_file, "w") as tar:
                # Add manifest.json
                tar.add(manifest_path, arcname="manifest.json")

                # Add config
                tar.add(config_path, arcname=config_file)

                # Add each layer directory
                for layer_digest in dict.fromkeys(layer_files):
                    layer_dir = os.path.join(tmpdir, layer_digest)
                    for root, _dirs, files in os.walk(layer_dir):
                        for file in files:
                            file_path = os.path.join(root, file)
                            arcname = os.path.relpath(file_path, tmpdir)
                            tar.add(file_path, arcname=arcname)

    def pull_image(
        self,
        image_spec,
        output_file=None,
        architecture="amd64",
        os_type="linux",
        variant=None,
        plain_http=False,
    ):
        """Pull a Docker image and save as tar with enhanced progress reporting.

        Args:
            image_spec (str): Image specification (name:tag)
            output_file (str, optional): Output tar filename
            architecture (str): Target architecture
            os_type (str): Target operating system
            variant (str, optional): Target architecture variant
            plain_http (bool): Use HTTP instead of HTTPS for the registry
        """

        (
            self.registry_url,
            repository,
            reference,
            full_image_name,
            archive_tag,
        ) = parse_image_spec(image_spec, plain_http)
        if plain_http:
            logger.warning(
                "Using unencrypted HTTP; registry traffic and credentials are exposed"
            )
        self.authorization = f"Bearer {self.auth_token}" if self.auth_token else None

        if not output_file:
            safe_name = re.sub(r"[^A-Za-z0-9_.-]", "_", full_image_name)
            output_reference = archive_tag or reference.split(":", 1)[-1][:16]
            safe_reference = re.sub(r"[^A-Za-z0-9_.-]", "_", output_reference)
            output_file = f"{safe_name}_{safe_reference}.tar"

        output_directory = os.path.dirname(os.path.abspath(output_file))
        if not os.path.isdir(output_directory):
            raise ValueError(f"Output directory does not exist: {output_directory}")

        display_reference = (
            f"{full_image_name}:{archive_tag}"
            if archive_tag
            else f"{full_image_name}@{reference}"
        )
        logger.info(f"Downloading image: {display_reference}")

        # Get auth token
        authorization = self.get_auth_header(repository)

        # Get manifest
        logger.info("Retrieving image manifest...")
        manifest = self.get_manifest(
            repository, reference, authorization, architecture, os_type, variant
        )
        authorization = self.authorization
        if not isinstance(manifest, dict):
            raise RegistryError("Manifest response is not a JSON object")

        # Download config
        logger.info("Downloading image configuration...")
        config_descriptor = manifest.get("config")
        if not isinstance(config_descriptor, dict):
            raise RegistryError("Manifest config descriptor is not an object")
        config_digest = config_descriptor.get("digest")
        if not config_digest:
            raise RegistryError("No config digest found in manifest")
        algorithm, encoded = validate_digest(config_digest)
        config_digest = f"{algorithm}:{encoded}"

        config_size = config_descriptor.get("size")
        if config_size is not None and (
            not isinstance(config_size, int)
            or isinstance(config_size, bool)
            or config_size < 0
        ):
            raise RegistryError("Manifest config descriptor has an invalid size")
        config_blob = self.download_blob(
            repository,
            config_digest,
            authorization,
            expected_size=config_size,
        )

        if config_blob is None:
            raise RegistryError("Failed to download config")

        try:
            config_data = json.loads(config_blob)
        except (json.JSONDecodeError, UnicodeDecodeError, TypeError) as error:
            raise RegistryError("Image configuration is not valid JSON") from error
        if not isinstance(config_data, dict):
            raise RegistryError("Image configuration is not a JSON object")
        actual_architecture = config_data.get("architecture")
        actual_os = config_data.get("os")
        actual_variant = config_data.get("variant")
        if (
            not isinstance(actual_architecture, str)
            or not isinstance(actual_os, str)
            or (actual_variant is not None and not isinstance(actual_variant, str))
        ):
            raise RegistryError("Image configuration platform is invalid")
        if (
            normalize_architecture(actual_architecture, actual_variant)
            != normalize_architecture(architecture, variant)
            or actual_os.lower() != os_type.lower()
            or (
                variant is not None
                and normalize_variant(actual_architecture, actual_variant)
                != normalize_variant(architecture, variant)
            )
        ):
            actual_platform = (
                f"{actual_os or 'unknown'}/{actual_architecture or 'unknown'}"
            )
            requested_platform = f"{os_type}/{architecture}" + (
                f"/{variant}" if variant else ""
            )
            raise RegistryError(
                f"Image is {actual_platform!r}, not the requested "
                f"{requested_platform!r}"
            )

        # Download layers with overall progress tracking
        layers = []
        layer_list = manifest.get("layers", [])
        if not isinstance(layer_list, list):
            raise RegistryError("Manifest layers must be a list")
        total_layers = len(layer_list)
        downloaded_size = 0

        if total_layers == 0:
            raise RegistryError("Manifest contains no layers")

        validated_layers = []
        for i, layer in enumerate(layer_list, 1):
            if not isinstance(layer, dict):
                raise RegistryError(f"Layer {i} descriptor is not an object")
            digest = layer.get("digest")
            if not digest:
                raise RegistryError(f"Layer {i} is missing a digest")
            algorithm, encoded = validate_digest(digest)
            digest = f"{algorithm}:{encoded}"

            size = layer.get("size")
            if size is not None and (
                not isinstance(size, int) or isinstance(size, bool) or size < 0
            ):
                raise RegistryError(f"Layer {i} has an invalid size")

            media_type = layer.get("mediaType")
            if media_type is not None and not isinstance(media_type, str):
                raise RegistryError(f"Layer {i} has an invalid media type")
            media_type = (media_type or "").lower()
            if "foreign" in media_type or "nondistributable" in media_type:
                raise RegistryError(
                    f"Layer {i} uses an unsupported foreign/nondistributable media type"
                )
            if "zstd" in media_type:
                raise RegistryError(f"Layer {i} uses unsupported Zstandard compression")
            validated_layers.append((digest, size, encoded))

        total_download_size = sum(
            size or 0 for _digest, size, _encoded in validated_layers
        )
        logger.info(
            f"Downloading {total_layers} layers "
            f"({format_bytes(total_download_size)} total)..."
        )
        overall_progress = ProgressReporter(
            total_download_size if total_download_size > 0 else None,
            "Overall progress",
            show_speed=True,
        )

        with tempfile.TemporaryDirectory() as blob_dir:
            for i, (digest, size, encoded) in enumerate(validated_layers, 1):
                size_str = format_bytes(size) if size is not None else "unknown size"
                logger.info(f"Layer {i}/{total_layers} ({digest[:12]}... {size_str})")

                destination = os.path.join(blob_dir, encoded)
                blob_path = destination if os.path.exists(destination) else None
                if blob_path is None:
                    blob_path = self.download_blob(
                        repository,
                        digest,
                        self.authorization,
                        expected_size=size,
                        destination=destination,
                    )
                if blob_path is None:
                    raise RegistryError(f"Failed to download layer {digest}")
                if size is not None and os.path.getsize(blob_path) != size:
                    raise RegistryError(f"Layer {digest} has inconsistent sizes")

                downloaded_size += size or os.path.getsize(blob_path)
                overall_progress.downloaded = downloaded_size
                overall_progress._display_progress()
                if logger.isEnabledFor(logging.INFO):
                    print()
                layers.append({"digest": digest, "size": size, "path": blob_path})

            logger.info(f"Creating tar file: {output_file}")
            descriptor, temporary_output = tempfile.mkstemp(
                prefix=".docker-pull-", suffix=".tar", dir=output_directory
            )
            os.close(descriptor)
            try:
                self.create_docker_tar(
                    full_image_name,
                    archive_tag,
                    manifest,
                    config_blob,
                    config_data,
                    layers,
                    temporary_output,
                )
                apply_default_file_mode(temporary_output)
                os.replace(temporary_output, output_file)
            finally:
                try:
                    os.unlink(temporary_output)
                except OSError:
                    pass
        # Calculate final size
        file_size = os.path.getsize(output_file)
        logger.info(
            f"Successfully created {output_file} (size: {format_bytes(file_size)})"
        )
        logger.info(f"To load this image, run: docker load -i {output_file}")


# =============================================================================
# ENTRY POINT
# =============================================================================


def main():
    parser = argparse.ArgumentParser(
        description="Pull Registry v2 images and save them as Docker tar files",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
        Environment Variables:
        HTTP_PROXY / http_proxy     - HTTP proxy URL
        HTTPS_PROXY / https_proxy   - HTTPS proxy URL  
        NO_PROXY / no_proxy         - Comma-separated hosts or host:port pairs to bypass proxy

        Examples:
        %(prog)s ubuntu:20.04
        %(prog)s alpine --output alpine-latest.tar
        
        # Pull ARM64 image
        %(prog)s ubuntu:latest --arch arm64

        # Pull from another registry
        %(prog)s icr.io/codeengine/helloworld:latest
        
        # With proxy
        %(prog)s nginx:latest --proxy http://proxy.company.com:8080
        
        # With proxy authentication
        %(prog)s alpine --proxy http://proxy.company.com:8080 --proxy-auth username:password
        
        # Using environment variables
        export HTTPS_PROXY=http://proxy.company.com:8080
        export NO_PROXY=localhost,127.0.0.1,.local
        %(prog)s ubuntu:latest
        
        # Disable SSL verification for corporate proxies
        %(prog)s nginx --proxy https://proxy.company.com:8080 --insecure
                """,
    )

    parser.add_argument(
        "image",
        help="Container image to pull (e.g., ubuntu:20.04 or ghcr.io/org/image:tag)",
    )

    parser.add_argument(
        "-o",
        "--output",
        help="Output tar filename (default: imagename_tag.tar)",
        default=None,
    )

    parser.add_argument(
        "-t",
        "--token",
        help="Bearer token on the command line (prefer --token-stdin)",
        default=None,
    )

    parser.add_argument(
        "--token-stdin",
        action="store_true",
        help="Read a Bearer authentication token from standard input",
    )

    parser.add_argument(
        "--username", help="Registry username (for private repositories)", default=None
    )

    parser.add_argument(
        "--password",
        help="Registry password or API key on the command line",
        default=None,
    )

    parser.add_argument(
        "--password-stdin",
        action="store_true",
        help="Read the registry password or API key from standard input",
    )

    # Architecture and OS arguments
    parser.add_argument(
        "--arch",
        "--architecture",
        help="Target architecture (default: amd64)",
        default="amd64",
        choices=[
            "amd64",
            "arm64",
            "arm",
            "386",
            "ppc64le",
            "s390x",
            "mips64le",
            "riscv64",
        ],
    )

    parser.add_argument(
        "--os",
        help="Target operating system (default: linux)",
        default="linux",
        choices=["linux", "windows"],
    )

    parser.add_argument(
        "--variant",
        help="Target architecture variant (for example, v6 or v7 for arm)",
        default=None,
        type=str.lower,
    )

    # Proxy arguments
    parser.add_argument(
        "-p",
        "--proxy",
        help="Proxy URL (e.g., http://proxy.company.com:8080)",
        default=None,
    )

    parser.add_argument(
        "--proxy-auth",
        help="Proxy authentication in format username:password",
        default=None,
    )

    parser.add_argument(
        "--http-proxy",
        help="HTTP proxy URL (overrides environment variable)",
        default=None,
    )

    parser.add_argument(
        "--https-proxy",
        help="HTTPS proxy URL (overrides environment variable)",
        default=None,
    )

    parser.add_argument(
        "--no-proxy",
        help="Comma-separated hosts or host:port pairs to bypass proxy",
        default=None,
    )

    parser.add_argument(
        "-k",
        "--insecure",
        action="store_true",
        help="Disable SSL certificate verification (useful for corporate proxies)",
    )

    parser.add_argument(
        "--plain-http",
        action="store_true",
        help="Use unencrypted HTTP for a trusted development registry",
    )

    parser.add_argument(
        "--debug", action="store_true", help="Enable debug output for troubleshooting"
    )

    parser.add_argument(
        "-v",
        "--verbose",
        action="store_true",
        help="Use INFO logging (default)",
    )

    parser.add_argument(
        "-q", "--quiet", action="store_true", help="Quiet mode - only show errors"
    )

    parser.add_argument(
        "--log-level",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        help="Set logging level explicitly",
        default=None,
    )

    args = parser.parse_args()

    # Set up logging based on arguments
    log_level = logging.INFO  # Default level

    if args.debug:
        log_level = logging.DEBUG
    elif args.quiet:
        log_level = logging.ERROR
    elif args.log_level:
        log_level = getattr(logging, args.log_level)
    elif args.verbose:
        log_level = logging.INFO

    setup_logging(log_level)

    # Build proxy configuration
    proxy_config = {}

    # Handle simplified --proxy argument
    if args.proxy:
        proxy_config["http_proxy"] = args.proxy
        proxy_config["https_proxy"] = args.proxy

    # Handle specific proxy settings
    if args.http_proxy:
        proxy_config["http_proxy"] = args.http_proxy

    if args.https_proxy:
        proxy_config["https_proxy"] = args.https_proxy

    if args.no_proxy:
        proxy_config["no_proxy"] = args.no_proxy

    if args.proxy_auth:
        proxy_config["proxy_auth"] = args.proxy_auth

    if args.insecure:
        proxy_config["insecure"] = True

    if args.plain_http and args.insecure:
        parser.error("--plain-http and --insecure cannot be used together")
    if args.token and args.token_stdin:
        parser.error("--token and --token-stdin cannot be used together")
    if args.password and args.password_stdin:
        parser.error("--password and --password-stdin cannot be used together")
    if args.token_stdin and args.password_stdin:
        parser.error("--token-stdin and --password-stdin cannot be used together")
    if (args.token is not None or args.token_stdin) and (
        args.username is not None or args.password is not None or args.password_stdin
    ):
        parser.error("Bearer token and username/password modes cannot be combined")

    stdin_secret = (
        sys.stdin.read().rstrip("\r\n")
        if args.token_stdin or args.password_stdin
        else None
    )
    auth_token = stdin_secret if args.token_stdin else args.token
    password = stdin_secret if args.password_stdin else args.password
    if args.token_stdin and not auth_token:
        parser.error("--token-stdin received an empty value")
    if args.password_stdin and not password:
        parser.error("--password-stdin received an empty value")
    if auth_token and ("\r" in auth_token or "\n" in auth_token):
        parser.error("Bearer tokens must contain exactly one line")
    if (args.username is None) != (password is None):
        parser.error("--username and a password must be provided together")

    try:
        puller = DockerImagePuller(
            auth_token=auth_token,
            proxy_config=proxy_config,
            username=args.username,
            password=password,
        )
        puller.pull_image(
            args.image,
            args.output,
            architecture=args.arch,
            os_type=args.os,
            variant=args.variant,
            plain_http=args.plain_http,
        )
    except KeyboardInterrupt:
        logger.info("Operation cancelled by user")
        sys.exit(1)
    except (RegistryError, ValueError) as e:
        logger.error(f"Error: {e}")
        sys.exit(1)
    except (HTTPError, URLError, OSError) as e:
        logger.error(f"Network/IO Error: {e}")
        sys.exit(1)
    except Exception as e:  # noqa: BLE001
        logger.error(f"Unexpected error: {e}")
        import traceback

        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    main()