From 7d2d43032035f058fec2d7b60f239625f3863b43 Mon Sep 17 00:00:00 2001 From: Jonatan <190036+jcastro@users.noreply.github.com> Date: Fri, 16 Oct 2026 19:58:40 +0200 Subject: [PATCH 1/2] Add private registry accounts to OCI manager Apps --- .github/workflows/test-oci-registries.yml | 34 +++ oci/README.md | 34 +++ oci/src/proxmenux_oci/cli.py | 4 + oci/src/proxmenux_oci/custom.py | 18 +- oci/src/proxmenux_oci/registry_accounts.py | 282 +++++++++++++++++ oci/src/proxmenux_oci/registry_auth.py | 334 +++++++++++++++++++++ oci/tests/test_registry_accounts.py | 168 +++++++++++ oci/tests/test_registry_auth.py | 224 ++++++++++++++ oci/tests/test_registry_failures.py | 160 ++++++++++ oci/tests/test_registry_references.py | 45 +++ 10 files changed, 1298 insertions(+), 5 deletions(-) create mode 100644 .github/workflows/test-oci-registries.yml create mode 100644 oci/src/proxmenux_oci/registry_accounts.py create mode 100644 oci/src/proxmenux_oci/registry_auth.py create mode 100644 oci/tests/test_registry_accounts.py create mode 100644 oci/tests/test_registry_auth.py create mode 100644 oci/tests/test_registry_failures.py create mode 100644 oci/tests/test_registry_references.py diff --git a/.github/workflows/test-oci-registries.yml b/.github/workflows/test-oci-registries.yml new file mode 100644 index 00000000..9612c0d3 --- /dev/null +++ b/.github/workflows/test-oci-registries.yml @@ -0,0 +1,34 @@ +name: Offline OCI registry tests + +on: + pull_request: + paths: + - 'oci/**' + - '.github/workflows/test-oci-registries.yml' + push: + branches: [main, develop] + paths: + - 'oci/**' + - '.github/workflows/test-oci-registries.yml' + workflow_dispatch: + +permissions: + contents: read + +jobs: + registry-accounts: + runs-on: ubuntu-latest + timeout-minutes: 5 + steps: + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: actions/setup-python@v5 + with: + python-version: '3.13' + - name: Install OCI test dependencies + run: python3 -m pip install -r oci/requirements.txt + - name: Run offline registry account tests + env: + PYTHONDONTWRITEBYTECODE: '1' + run: python3 -m unittest discover -s oci/tests -p 'test_registry_*.py' -v diff --git a/oci/README.md b/oci/README.md index 6c2c84fa..45b05c2c 100644 --- a/oci/README.md +++ b/oci/README.md @@ -10,6 +10,7 @@ This document describes the engine in this directory: how it is organised, how a - [Structure](#structure) - [How a container is created](#how-a-container-is-created) - [Update, modify, recreate and remove](#update-modify-recreate-and-remove) +- [Private image registries](#private-image-registries) - [From a Compose file or a docker run command to an oci-compose](#from-a-compose-file-or-a-docker-run-command-to-an-oci-compose) - [The oci-compose, block by block](#the-oci-compose-block-by-block) - [Writing and testing an oci-compose](#writing-and-testing-an-oci-compose) @@ -60,6 +61,39 @@ There are three places an oci-compose comes from: On a node, the engine is installed in `/usr/local/share/proxmenux/oci/engine/`. Every installation keeps its record in `/usr/local/share/proxmenux/oci/instances//oci-compose.json`: the oci-compose it was installed from, under `template`, together with the answers given during the installation and the configuration Proxmox VE ended with. Updates, changes and removals read that record. +## Private image registries + +Open **OCI manager Apps → Registry accounts** on the Proxmox node. Sign in +with a registry hostname (for example `ghcr.io` or `registry.example.com:5443`), +a username and a password or token with image pull permission. Passwords are +entered in hidden fields and confirmed before the registry is contacted. + +Select a saved account to check access to an image, download it to template +storage, install it with the existing LXC wizard, update its credentials or +remove them. Image references retain their tag or full `sha256` digest. +The normal custom-image importer can also use the saved credentials, including +Compose files whose services come from private registries. + +Skopeo stores credentials in `/root/.config/containers/auth.json` by default. +`REGISTRY_AUTH_FILE` and `XDG_CONFIG_HOME` overrides are respected. The +credential directory must be owned by root with mode `700`, and the file with +mode `600`. Basic credentials are base64 encoded, not encrypted. Configured +Skopeo credential helpers use their own stores. Failed sign-in attempts leave +the previous file credentials intact; removing an account leaves running +applications unchanged. Other Skopeo/Docker authentication files are managed +separately and may still grant access. + +TLS verification remains enabled. Install a private registry CA in +`/etc/containers/certs.d/REGISTRY[:PORT]/ca.crt` when required. Configure accounts +on each node that pulls images; registry credentials are not written into the +application template or copied into its LXC. + +Image access checks have a 30-second timeout per request; sign-in has a 45-second +timeout. Errors identify rejected credentials, missing images, incompatible +architectures, network/TLS failures, rate limits and registry outages. Downloads +reuse the native update cache and layer integrity checks. Damaged cached images +are downloaded again, and storage failures leave existing applications alone. + ## How a container is created Selecting an application, or giving a definition, opens the same sequence. diff --git a/oci/src/proxmenux_oci/cli.py b/oci/src/proxmenux_oci/cli.py index 4e9025d9..9266f283 100644 --- a/oci/src/proxmenux_oci/cli.py +++ b/oci/src/proxmenux_oci/cli.py @@ -559,6 +559,7 @@ def interactive(catalog: Catalog) -> int: ("search", translate("Search applications")), ("all", f"{translate('All applications'):<35} ({len(applications):>3})"), ("manage", translate("Manage installed OCI applications")), + ("registries", translate("Registry accounts")), ("custom", translate("Install an image that is not in the catalog")), ("", ""), ] + [(number, f"{translate(labels[key]):<35} ({len(categories[key]):>3})") @@ -580,6 +581,9 @@ def interactive(catalog: Catalog) -> int: console.show_logo() console.msg_title(translate("Manage installed OCI applications")) interactive_management(PROJECT_ROOT, ui) + elif selection == "registries": + from .registry_accounts import open_registry_accounts + open_registry_accounts(ui) elif selection == "custom": from .custom import explore explore(ui) diff --git a/oci/src/proxmenux_oci/custom.py b/oci/src/proxmenux_oci/custom.py index f15f3152..6364265c 100644 --- a/oci/src/proxmenux_oci/custom.py +++ b/oci/src/proxmenux_oci/custom.py @@ -22,9 +22,17 @@ from .i18n import source_text from .installer import ADVANCED_MODE, DEFAULT_MODE from .converter import ConversionError from .i18n import translate +from .registry_auth import RegistryError, validate_image from .ui import UserCancelled -IMAGE_REFERENCE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9._/-]*(?::[A-Za-z0-9._-]+)?(?:@sha256:[a-f0-9]{64})?$') +def _valid_image_reference(reference: str) -> bool: + try: + validate_image(reference) + return True + except RegistryError: + return False + + MAX_COMPOSE_BYTES = 256 * 1024 # An oci-compose: the file ProxMenux installs an application from. The catalog # keeps one for each of its applications and every installation records its own. @@ -176,7 +184,7 @@ def template_from_compose(text: str, title: str | None = None) -> dict[str, Any] def compose_from_image(reference: str, title: str | None = None) -> str: """A Compose file written from what the image declares: its ports, its persistent paths and its variables.""" - if not IMAGE_REFERENCE.fullmatch(reference): + if not _valid_image_reference(reference): raise ConversionError(f"{translate('This is not a valid image reference:')} {reference}") result = subprocess.run(['skopeo', 'inspect', '--config', f'docker://{reference}'], capture_output=True, text=True, check=False, timeout=120) @@ -283,7 +291,7 @@ def compose_from_docker_run(command: str, title: str | None = None) -> str: f"{', '.join(sorted(set(unsupported)))}") if not image: raise ConversionError(translate('The command does not name an image')) - if not IMAGE_REFERENCE.fullmatch(image): + if not _valid_image_reference(image): raise ConversionError(f"{translate('This is not a valid image reference:')} {image}") service['image'] = image name = normalize_app_id(title or service.get('container_name') @@ -402,7 +410,7 @@ def template_from_oci_compose(text: str) -> dict[str, Any]: reference = image.get('reference') if isinstance(image, dict) else None if not isinstance(reference, str) or not reference.strip(): raise ConversionError(translate('The oci-compose is incomplete: it does not say which image to install.')) - if not IMAGE_REFERENCE.fullmatch(reference): + if not _valid_image_reference(reference): raise ConversionError(f"{translate('This is not a valid image reference:')} {reference}") faulty = translate('The oci-compose has errors:') @@ -726,7 +734,7 @@ def explore(ui) -> None: summary += '\n\n' + '\n'.join(notes) if not available: advice = translate('Check the image reference and registry access. If the registry is unavailable, ' - 'try again later. A private registry needs credentials, which are not supported yet.') + 'try again later. For a private image, sign in from Registry accounts in the OCI menu.') ui.message(f'{summary}\n\n{advice}', title) return if not template.get('compatibility', {}).get('automatic_install_candidate'): diff --git a/oci/src/proxmenux_oci/registry_accounts.py b/oci/src/proxmenux_oci/registry_accounts.py new file mode 100644 index 00000000..2e67a855 --- /dev/null +++ b/oci/src/proxmenux_oci/registry_accounts.py @@ -0,0 +1,282 @@ +# ========================================================== +# ProxMenux - Private OCI registry accounts +# ========================================================== +# Author : Jonatan +# GitHub : github.com/jcastro +# Maintainer : MacRimi +# Copyright : (c) 2026 MacRimi & contributors +# License : (GPL-3.0) (https://github.com/MacRimi/ProxMenux/blob/main/LICENSE) +# Version : 1.0 +# Last Updated: 16/10/2026 +# ========================================================== +# Description: +# Manage private registry accounts from the native OCI menu. +# Check image access, download verified archives and hand image +# installation to the existing ProxMenux LXC wizard. +# ========================================================== + +"""Private image registry accounts in the native ProxMenux OCI menu.""" +from __future__ import annotations + +from contextlib import contextmanager, redirect_stdout +import errno +import importlib +import io +import os +from pathlib import Path +import re +import subprocess +import sys + +from .i18n import translate +from .registry_auth import (RegistryAuth, RegistryError, clean, require_root, + validate_registry, validate_image, registry_of, + registry_error, registry_environment) + +ENGINE_ROOT = Path(__file__).resolve().parents[2] +REGISTRY_QUERY_TIMEOUT = 30 + + +@contextmanager +def image_queries(reference): + """Keep the native resolver while retaining safe, useful transport errors.""" + images = importlib.import_module('oci_installation_state') + original = images.command + registry = registry_of(reference) + + def query(*args): + if args[:2] != ('skopeo', 'inspect'): + return original(*args) + try: + result = subprocess.run(args, capture_output=True, timeout=REGISTRY_QUERY_TIMEOUT) + except subprocess.TimeoutExpired: + raise RegistryError(translate('{registry} did not respond within {timeout} seconds. Try again later.').format(registry=registry, timeout=REGISTRY_QUERY_TIMEOUT)) from None + except FileNotFoundError: + raise RegistryError(translate('Skopeo is not available. Restore the ProxMenux OCI dependencies and try again.')) from None + if result.returncode: + raise RegistryError(registry_error(registry, result.stderr.decode(errors='replace'))) + return result.stdout + + images.command = query + try: + yield images + finally: + images.command = original + + +def bytes_label(size): + value = float(size) + for unit in ('B', 'KiB', 'MiB', 'GiB', 'TiB'): + if value < 1024 or unit == 'TiB': + return f'{value:.0f} {unit}' if unit == 'B' else f'{value:.1f} {unit}' + value /= 1024 + + +def host_architecture(): + architecture = {'x86_64': 'amd64', 'aarch64': 'arm64'}.get(os.uname().machine) + if not architecture: + raise RegistryError(translate('This host architecture is not supported by the OCI image importer.')) + return architecture + + +def image_template(reference): + reference = validate_image(reference) + custom = importlib.import_module('proxmenux_oci.custom') + try: + compose = custom.compose_from_image(reference) + except (custom.ConversionError, subprocess.TimeoutExpired, ValueError) as error: + raise RegistryError(registry_error(registry_of(reference), error)) from None + try: + template = custom.template_from_compose(compose) + custom._ask_secrets(template) + except custom.ConversionError: + raise RegistryError(translate('ProxMenux could not translate this image. Try importing its Compose file from OCI manager Apps.')) from None + return template + + +def inspect_registry_image(reference): + reference = validate_image(reference) + try: + with registry_environment(), image_queries(reference) as images: + candidate = images.resolve_candidate(reference, host_architecture()) + except RegistryError: + raise + except ValueError as error: + detail = str(error).casefold() + if 'platform' in detail or 'architecture' in detail: + raise RegistryError(translate('This image has no compatible Linux image for this host ({architecture}).').format(architecture=host_architecture())) from None + if 'manifest does not match' in detail or 'integrity' in detail: + raise RegistryError(translate('The image manifest failed integrity verification. Try again or contact the image publisher.')) from None + raise RegistryError(translate('The registry returned invalid or unsupported image metadata.')) from None + except (KeyError, AttributeError, TypeError): + raise RegistryError(translate('The registry returned incomplete or unsupported image metadata.')) from None + except (RuntimeError, subprocess.TimeoutExpired) as error: + raise RegistryError(registry_error(registry_of(reference), error)) from None + return {'reference': reference, 'architecture': host_architecture(), 'version': clean(candidate.get('version')) or None, + 'digest': candidate['manifest_digest'], 'size': candidate.get('size') or 0} + + +def download_registry_image(reference, storage): + """Download and verify with the exact native updater cache implementation.""" + reference = validate_image(reference) + if not re.fullmatch(r'[A-Za-z0-9_-]+', storage): + raise RegistryError(translate('Invalid template storage name.')) + inspect_registry_image(reference) + native = importlib.import_module('oci_update_current') + desired = {'template': {'container_contract': {'image': {'reference': reference}}}, + 'deployment': {'template_storage': storage}} + try: + with native.instances.locked(native.instances.ROOT), registry_environment(), redirect_stdout(io.StringIO()): + path, digest = native.resolve_archive(desired, f'arch: {host_architecture()}\n'.encode()) + except BlockingIOError: + raise RegistryError(translate('Another ProxMenux OCI operation is running. Try again when it finishes.')) from None + except RegistryError: + raise + except (RuntimeError, ValueError, OSError, subprocess.TimeoutExpired) as error: + detail = str(error).casefold() + if getattr(error, 'errno', None) == errno.ENOSPC or 'does not fit' in detail or 'no space left' in detail: + raise RegistryError(translate('Not enough free space on {storage}. Free some space or select another storage.').format(storage=storage)) from None + if isinstance(error, PermissionError) or getattr(error, 'errno', None) == errno.EROFS: + raise RegistryError(translate('Cannot write the image archive to {storage}. Check storage access and mount status.').format(storage=storage)) from None + if 'unsafe' in detail or 'symbolic link' in detail: + raise RegistryError(translate('The image cache path is unsafe. Check the template storage before trying again.')) from None + if 'integrity' in detail or 'intact image' in detail: + raise RegistryError(translate('Image integrity verification failed. The partial download was discarded; try again.')) from None + raise RegistryError(translate('The image could not be downloaded or verified. Check registry and storage access, then try again.')) from None + return {'reference': reference, 'archive': str(path), 'digest': digest, 'storage': storage, + 'size': Path(path).stat().st_size, 'architecture': host_architecture()} + + +def registry_accounts_menu(ui, engine=ENGINE_ROOT, auth=None): + """Manage private registry accounts and images in ProxMenux.""" + require_root() + engine = Path(engine) + if str(engine / 'remote') not in sys.path: + sys.path.insert(0, str(engine / 'remote')) + auth = auth or RegistryAuth() + cancelled = importlib.import_module('proxmenux_oci.ui').UserCancelled + title = translate('Registry accounts') + while True: + # A broken authfile must let the user return to ProxMenux instead of + # showing the same error indefinitely before a menu can be displayed. + try: + rows = auth.accounts() + except (RegistryError, OSError, ValueError) as error: + ui.message(clean(error, 600), title) + return + try: + options = [('add', translate('Sign in to a registry'))] + options += [(f'account-{index}', f"{row['registry']:<28} {row['account'] or row['method']} [{translate('saved')}]") + for index, row in enumerate(rows)] + options += [('storage', translate('Credential storage and access'))] + count = translate('{count} saved account.') if len(rows) == 1 else translate('{count} saved accounts.') + choice = ui.choose(translate('Manage private image registries.\n' + 'Saved credentials let ProxMenux access your private images.') + + '\n\n' + count.format(count=len(rows)), + options, 'account-0' if rows else 'add', title=title, show_tags=False) + if choice is None: + return + row = None + if choice.startswith('account-'): + try: + row = rows[int(choice.partition('-')[2])] + except (ValueError, IndexError): + continue + choice = ui.choose(translate('Registry: {registry}\nAccount: {account}\n' + 'Authentication: {method}\n\n' + 'Saved credentials are verified when an image is accessed.').format( + registry=row['registry'], account=row['account'] or translate('Credential helper'), + method=row['method']), + [('inspect', translate('Check access to an image')), + ('download', translate('Download an image to template storage')), + ('install', translate('Install an image as a native LXC')), + ('login', translate('Update credentials')), + ('logout', translate('Remove saved credentials'))], 'inspect', title=title, show_tags=False) + if choice is None: + continue + if choice == 'storage': + ui.message(translate('Credentials file:\n{path}\n\n' + 'Owned by root; directory mode 700 and file mode 600.\n' + 'Basic credentials are base64 encoded, not encrypted.\n' + 'Credential helpers use their own storage.\n\n' + 'Use an access token with permission to pull images.\n' + 'Configure credentials on each node that downloads images.').format(path=auth.path), title) + elif choice in {'add', 'login'}: + registry = row['registry'].split('/', 1)[0] if row else ui.ask(translate('Registry hostname'), 'ghcr.io') + try: + registry = validate_registry(registry) + except RegistryError as error: + raise RegistryError(str(error)) from None + username = ui.ask(translate('Username'), row['account'] if row else None) + password = ui.password(translate('Password / access token (image pull permission)')) + ui.info(translate('Verifying registry access. Please wait...')) + auth.login(registry, username, password) + del password + ui.message(translate('Signed in to {registry} as {username}.\n\n' + 'Credentials are saved for ProxMenux.\n' + 'Select the account to check, download or install an image.').format( + registry=registry, username=clean(username)), title) + elif choice == 'logout' and row: + registry = row['registry'].split('/', 1)[0] + if ui.confirm(translate('Remove the saved credentials for {registry}?\nRunning apps will keep working.').format(registry=registry), + False, title=title): + auth.logout(registry) + ui.message(translate('Saved credentials for {registry} were removed.\n' + 'Other Skopeo/Docker credential files are managed separately.').format(registry=registry), title) + elif choice in {'inspect', 'download', 'install'} and row: + registry = row['registry'].split('/', 1)[0] + owner = row['account'] or 'owner' + reference = validate_image(ui.ask(translate('Image reference'), f'{registry}/{owner}/app:latest')) + if registry_of(reference) != registry: + raise RegistryError(translate('This account is for {registry}. Enter an image from that registry.').format(registry=registry)) + if choice == 'inspect': + ui.info(translate('Checking image access and metadata. Please wait...')) + result = inspect_registry_image(reference) + ui.message(translate('Image access confirmed.\n\n{reference}\n\n' + 'Architecture: {architecture}\nVersion: {version}\n' + 'Download size: {size}\nDigest: {digest}').format( + reference=reference, architecture=result['architecture'], + version=result['version'] or translate('Not declared'), + size=bytes_label(result['size']), digest=result['digest']), title) + elif choice == 'download': + host = importlib.import_module('proxmenux_oci.host') + storages = host.storages('vztmpl') + if not storages: + raise RegistryError(translate('No active Proxmox storage accepts OCI template archives.')) + names = [item['storage'] for item in storages] + storage = ui.choose(translate('Choose storage for the downloaded image.'), + [(item['storage'], f"{item['storage']} ({bytes_label(item.get('avail', 0))} {translate('available')})") + for item in storages], 'local' if 'local' in names else names[0], title=title) + if storage is None: + continue + ui.info(translate('Downloading and verifying the image.\nThis can take a few minutes. Please wait...')) + result = download_registry_image(reference, storage) + ui.message(translate('Image downloaded and verified.\n\n{reference}\n\n' + 'Storage: {storage}\nArchive size: {size}\n' + 'Archive: {archive}\n\nNo container was created.\n' + 'ProxMenux removes unused cached images after successful updates.').format( + reference=reference, storage=storage, size=bytes_label(result['size']), + archive=result['archive']), title) + else: + ui.info(translate('Reading image metadata. Please wait...')) + with registry_environment(): + template = image_template(reference) + if not template.get('compatibility', {}).get('automatic_install_candidate'): + raise RegistryError(translate('ProxMenux cannot install this image automatically. Import its Compose file instead.')) + installer = importlib.import_module('proxmenux_oci.installer') + native_cli = importlib.import_module('proxmenux_oci.cli') + mode = ui.choose(translate('Installation settings'), [(installer.DEFAULT_MODE, translate('Default settings')), + (installer.ADVANCED_MODE, translate('Advanced settings'))], installer.DEFAULT_MODE, title=title) + if mode is not None: + native_cli.install_template(ui, template, template['id'], mode) + except cancelled: + continue + except (RegistryError, OSError, ValueError) as error: + ui.message(clean(error, 600), title) + + +def open_registry_accounts(ui): + try: + registry_accounts_menu(ui) + except (RegistryError, OSError, ValueError) as error: + ui.message(clean(error, 600), translate('Registry accounts')) diff --git a/oci/src/proxmenux_oci/registry_auth.py b/oci/src/proxmenux_oci/registry_auth.py new file mode 100644 index 00000000..8cfac7d5 --- /dev/null +++ b/oci/src/proxmenux_oci/registry_auth.py @@ -0,0 +1,334 @@ +# ========================================================== +# ProxMenux - Private OCI registry accounts +# ========================================================== +# Author : Jonatan +# GitHub : github.com/jcastro +# Maintainer : MacRimi +# Copyright : (c) 2026 MacRimi & contributors +# License : (GPL-3.0) (https://github.com/MacRimi/ProxMenux/blob/main/LICENSE) +# Version : 1.0 +# Last Updated: 16/10/2026 +# ========================================================== +# Description: +# Manage persistent Skopeo credentials for private registries. +# Validate accounts without exposing tokens and preserve saved +# file credentials when authentication fails. +# ========================================================== + +"""Persistent Skopeo authentication for private ProxMenux OCI images.""" +from __future__ import annotations + +import base64 +from contextlib import contextmanager +import fcntl +import ipaddress +import json +import os +from pathlib import Path +import re +import stat +import subprocess +import tempfile + +from .i18n import translate + +ANSI = re.compile(r'\x1b\][^\x07]*(?:\x07|\x1b\\)|\x1b\[[0-?]*[ -/]*[@-~]') + + +class RegistryError(RuntimeError): + pass + + +def clean(value, limit=180): + text = ANSI.sub('', str(value or '')) + return ''.join(c for c in text if ord(c) >= 32 and ord(c) != 127)[:limit] + + +def image_repository(reference): + repository = reference.split('@', 1)[0] + if ':' in repository.rsplit('/', 1)[-1]: + repository = repository.rsplit(':', 1)[0] + return repository + + +def require_root(): + if os.geteuid() != 0: + raise RegistryError(translate('Manage registry accounts as root on the Proxmox node.')) + + +def validate_registry(value): + """Accept a registry authority, never a URL, image path or credentials.""" + value = str(value).strip().lower() + invalid = translate('Use a registry hostname, for example ghcr.io or registry.example.com:5000.') + if not value or len(value) > 260 or any(c in value for c in '/@?#\\'): + raise RegistryError(invalid) + if value.startswith('['): + match = re.fullmatch(r'\[([^]]+)\](?::([0-9]+))?', value) + try: + if not match or ipaddress.ip_address(match[1]).version != 6: + raise ValueError() + except ValueError: + raise RegistryError(invalid) from None + port = match[2] + else: + match = re.fullmatch(r'([a-z0-9](?:[a-z0-9.-]*[a-z0-9])?)(?::([0-9]+))?', value) + if not match or any(not re.fullmatch(r'[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?', part) + for part in match[1].split('.')): + raise RegistryError(invalid) + port = match[2] + if port and not 1 <= int(port) <= 65535: + raise RegistryError(invalid) + return {'index.docker.io': 'docker.io', 'registry-1.docker.io': 'docker.io'}.get(value, value) + + +def credential_key(value): + """Compare existing Docker aliases without exposing credential contents.""" + value = str(value).removeprefix('https://').removeprefix('http://').rstrip('/') + return {'index.docker.io/v1': 'docker.io', 'index.docker.io': 'docker.io', + 'registry-1.docker.io': 'docker.io'}.get(value, value) + + +def registry_of(reference): + first = image_repository(reference).split('/', 1)[0] + if '/' in reference and ('.' in first or ':' in first or first == 'localhost'): + return first + return 'docker.io' + + +def validate_image(value): + """Validate before the native importer, including registry ports and IPv6.""" + invalid = translate('Use an image reference, for example ghcr.io/owner/app:latest.') + if not value or len(value) > 512 or any(c.isspace() or ord(c) < 32 for c in value): + raise RegistryError(invalid) + name, separator, digest = value.partition('@') + if separator and not re.fullmatch(r'sha256:[a-f0-9]{64}', digest): + raise RegistryError(translate('Only a full sha256 image digest is supported.')) + last = name.rsplit('/', 1)[-1] + if ':' in last: + name, tag = name.rsplit(':', 1) + if not re.fullmatch(r'[A-Za-z0-9_][A-Za-z0-9._-]{0,127}', tag): + raise RegistryError(translate('Invalid image tag.')) + parts = name.split('/') + if len(parts) > 1 and ('.' in parts[0] or ':' in parts[0] or parts[0] == 'localhost'): + try: + validate_registry(parts.pop(0)) + except RegistryError: + raise RegistryError(invalid) from None + component = r'[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*' + if not parts or len(name) > 255 or any(not re.fullmatch(component, part) for part in parts): + raise RegistryError(invalid) + return value + + +def registry_error(registry, error='', login=False): + """Classify errors without echoing registry-controlled text or a token.""" + detail = str(error).casefold() + if any(word in detail for word in ('credential helper', 'docker-credential-', 'error getting credentials', 'error storing credentials')): + return translate('The configured credential helper could not be used. Check its installation and credential store.') + if any(word in detail for word in ('too many requests', 'toomanyrequests', 'statuscode: 429', 'status code: 429', 'status: 429')): + return translate('{registry} is rate limiting requests. Wait a while and try again.').format(registry=registry) + if any(word in detail for word in ('statuscode: 500', 'status code: 500', 'status: 500', + 'statuscode: 502', 'status code: 502', 'status: 502', + 'statuscode: 503', 'status code: 503', 'status: 503', + 'statuscode: 504', 'status code: 504', 'status: 504', + 'internal server error', 'service unavailable', 'bad gateway')): + return translate('{registry} reported a server error. Try again later.').format(registry=registry) + if any(word in detail for word in ('unauthorized', 'authentication required', 'denied', 'forbidden', 'invalid username', + 'invalid password', 'invalid credentials', 'username/password', + 'statuscode: 401', 'statuscode: 403', 'status code from registry 401', + 'status code from registry 403')): + return (translate('Authentication failed for {registry}. Check the account and token permissions.').format(registry=registry) if login else + translate('Access denied by {registry}. Check the image name and sign in from Registry accounts.').format(registry=registry)) + if any(word in detail for word in ('http response to https client', + 'first record does not look like a tls handshake')): + return translate('{registry} answered with plain HTTP. Use a registry with HTTPS; TLS verification remains enabled.').format(registry=registry) + if any(word in detail for word in ('x509', 'certificate', 'tls handshake')): + return translate('TLS verification failed for {registry}. Install the registry CA in /etc/containers/certs.d/{registry}/ca.crt.').format(registry=registry) + if any(word in detail for word in ('no such host', 'connection refused', 'deadline exceeded', 'timeout', 'timed out', + 'network is unreachable', 'connection reset', 'unexpected eof')): + return translate('Could not reach {registry}. Check its address, DNS and network access.').format(registry=registry) + if any(word in detail for word in ('manifest unknown', 'name unknown', 'not found', 'statuscode: 404', 'status code: 404')): + return (translate('No registry API was found at {registry}. Check the hostname and port.').format(registry=registry) if login else + translate('The image or tag was not found on {registry}, or this account cannot access it. Check the reference.').format(registry=registry)) + if any(word in detail for word in ('invalid manifest', 'invalid character', 'unsupported manifest', 'unsupported schema')): + return translate('{registry} returned invalid or unsupported image metadata.').format(registry=registry) + return (translate('Could not sign in to {registry}. Check registry access and token permissions.').format(registry=registry) if login else + translate('Could not read the image from {registry}. Check the reference and sign in from Registry accounts.').format(registry=registry)) + + +class RegistryAuth: + """Persistent Skopeo credentials for the native ProxMenux menu. + + Only the selected authfile is managed. Login is staged so failed attempts + never replace saved file credentials. Skopeo still manages credential helpers. + """ + def __init__(self, path=None): + configured = os.environ.get('REGISTRY_AUTH_FILE') + config_home = Path(os.environ.get('XDG_CONFIG_HOME') or Path.home() / '.config') + self.path = Path(path if path is not None else configured or config_home / 'containers/auth.json') + if not self.path.is_absolute() or '..' in self.path.parts: + raise RegistryError(translate('The registry authentication file must have an absolute path.')) + + def _validate_path(self): + for path in [self.path, *self.path.parents]: + if path.is_symlink(): + raise RegistryError(translate('The registry authentication path cannot contain symbolic links.')) + for path in self.path.parents: + if not path.exists(): + continue + info = path.stat() + shared_temporary = info.st_uid == 0 and info.st_mode & stat.S_ISVTX + if (not stat.S_ISDIR(info.st_mode) or info.st_uid not in {0, os.geteuid()} + or info.st_mode & 0o022 and not shared_temporary): + raise RegistryError(translate('The registry authentication path has an unsafe parent directory.')) + parent = self.path.parent + if parent.exists(): + info = parent.stat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077: + raise RegistryError(translate('The registry credential directory must be owned by root with mode 700.')) + + def _read(self): + self._validate_path() + try: + fd = os.open(self.path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK) + except FileNotFoundError: + return {}, None + with os.fdopen(fd, 'rb') as source: + info = os.fstat(source.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077: + raise RegistryError(translate('The registry authentication file must be owned by root with mode 600.')) + raw = source.read(1024 * 1024 + 1) + try: + data = json.loads(raw) if len(raw) <= 1024 * 1024 else None + except ValueError: + data = None + if not isinstance(data, dict) or any(not isinstance(data.get(key, {}), dict) for key in ('auths', 'credHelpers')): + raise RegistryError(translate('The registry authentication file is invalid; it was not changed.')) + return data, raw + + def accounts(self): + data, _ = self._read() + rows = {} + for key, credential in data.get('auths', {}).items(): + if not isinstance(credential, dict): + continue + username = None + encoded = credential.get('auth') + if isinstance(encoded, str) and len(encoded) < 65536: + try: + decoded = base64.b64decode(encoded, validate=True).decode('utf-8') + if ':' in decoded: + username = clean(decoded.split(':', 1)[0], 120) + except (ValueError, UnicodeError): + pass + if encoded or credential.get('identitytoken') or credential.get('registrytoken'): + rows[credential_key(key)] = {'registry': clean(credential_key(key), 320), 'account': username, + 'method': translate('Token') if not encoded else translate('Username / token'), 'status': 'Saved'} + for key, helper in data.get('credHelpers', {}).items(): + rows[credential_key(key)] = {'registry': clean(credential_key(key), 320), 'account': None, + 'method': translate('Helper:') + ' ' + clean(helper, 80), 'status': 'Configured'} + return sorted(rows.values(), key=lambda row: row['registry']) + + @contextmanager + def _staged(self): + self._validate_path() + self.path.parent.mkdir(parents=True, mode=0o700, exist_ok=True) + self._validate_path() + lock = self.path.with_name('.' + self.path.name + '.proxmenux.lock') + fd = os.open(lock, os.O_CREAT | os.O_WRONLY | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600) + try: + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077: + raise RegistryError(translate('Unsafe registry credential lock.')) + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + raise RegistryError(translate('Another registry account operation is running. Try again when it finishes.')) from None + data, before = self._read() + stage_fd, name = tempfile.mkstemp(prefix='.proxmenux-auth-', dir=self.path.parent) + stage = Path(name) + try: + with os.fdopen(stage_fd, 'wb') as out: + out.write(before if before is not None else b'{"auths":{}}\n') + yield stage, data + self._read_stage(stage) + if self._read()[1] != before: + raise RegistryError(translate('The credential file changed during this operation. Try again; it was not overwritten.')) + stage.chmod(0o600) + with stage.open('rb') as saved: + os.fsync(saved.fileno()) + os.replace(stage, self.path) + finally: + stage.unlink(missing_ok=True) + finally: + os.close(fd) + + @staticmethod + def _read_stage(stage): + try: + data = json.loads(stage.read_text()) + if not isinstance(data, dict) or any(not isinstance(data.get(k, {}), dict) for k in ('auths', 'credHelpers')): + raise ValueError() + return data + except (OSError, ValueError): + raise RegistryError(translate('Skopeo returned an invalid credential file; saved credentials were not changed.')) from None + + def login(self, registry, username, password, timeout=45): + registry = validate_registry(registry) + if not username or not username.strip() or len(username) > 512 or any(ord(c) < 32 or ord(c) == 127 or c == ':' for c in username): + raise RegistryError(translate('Enter a valid registry username.')) + if not password or len(password) > 65536 or any(c in password for c in '\r\n\0'): + raise RegistryError(translate('Enter a non-empty password or access token on a single line.')) + with self._staged() as (stage, _): + try: + process = subprocess.run(['skopeo', 'login', '--authfile', str(stage), '--username', username, + '--password-stdin', registry], input=password + '\n', text=True, + capture_output=True, timeout=timeout) + except subprocess.TimeoutExpired: + raise RegistryError(translate('Login to {registry} timed out. Saved credentials were not changed.').format(registry=registry)) from None + except FileNotFoundError: + raise RegistryError(translate('Skopeo is not available. Restore the ProxMenux OCI dependencies and try again.')) from None + if process.returncode: + raise RegistryError(registry_error(registry, process.stderr, login=True)) + + def logout(self, registry): + registry = validate_registry(registry) + data, _ = self._read() + matches = lambda key: credential_key(key).split('/', 1)[0] == registry + if not any(matches(key) for group in ('auths', 'credHelpers') for key in data.get(group, {})): + return False + with self._staged() as (stage, data): + # Skopeo logs out the authority itself; entries scoped to a + # repository are also removed from this managed authfile below. + if any(credential_key(key) == registry for group in ('auths', 'credHelpers') for key in data.get(group, {})): + try: + process = subprocess.run(['skopeo', 'logout', '--authfile', str(stage), registry], + capture_output=True, text=True, timeout=30) + except subprocess.TimeoutExpired: + raise RegistryError(translate('Logout from {registry} timed out. Saved file credentials were not changed.').format(registry=registry)) from None + except FileNotFoundError: + raise RegistryError(translate('Skopeo is not available. Restore the ProxMenux OCI dependencies and try again.')) from None + if process.returncode: + raise RegistryError(translate('Could not remove credentials for {registry}; saved file credentials were not changed.').format(registry=registry)) + data = self._read_stage(stage) + for group in ('auths', 'credHelpers'): + data[group] = {key: value for key, value in data.get(group, {}).items() if not matches(key)} + stage.write_text(json.dumps(data)) + return True + + +@contextmanager +def registry_environment(): + """Use durable credentials for every descendant, including update probes.""" + previous = os.environ.get('REGISTRY_AUTH_FILE') + auth = RegistryAuth() + if auth.path.exists(): + auth._read() + os.environ['REGISTRY_AUTH_FILE'] = str(auth.path) + try: + yield + finally: + if previous is None: + os.environ.pop('REGISTRY_AUTH_FILE', None) + else: + os.environ['REGISTRY_AUTH_FILE'] = previous diff --git a/oci/tests/test_registry_accounts.py b/oci/tests/test_registry_accounts.py new file mode 100644 index 00000000..4e349c95 --- /dev/null +++ b/oci/tests/test_registry_accounts.py @@ -0,0 +1,168 @@ +import contextlib +import io +from pathlib import Path +import sys +from types import SimpleNamespace +import unittest +from unittest.mock import MagicMock, patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +sys.path.insert(0, str(ROOT / "remote")) +from proxmenux_oci import registry_accounts as accounts + + +class Cancelled(RuntimeError): + pass + + +class UI: + def __init__(self, choices, answers=(), password='MENU_TEST_TOKEN'): + self.choices = iter(choices) + self.answers = iter(answers) + self.password_value = password + self.seen = [] + + def choose(self, text, options, default=None, **kwargs): + self.seen.append((text, options, kwargs)) + return next(self.choices) + + def ask(self, text, default=None): + self.seen.append(text) + return next(self.answers) + + def password(self, text): + self.seen.append(text) + if isinstance(self.password_value, Exception): + raise self.password_value + return self.password_value + + def message(self, text, title=None): + self.seen.append((title, text)) + + def info(self, text, title=None): + self.seen.append((title, text)) + + def confirm(self, text, default=False, **kwargs): + self.seen.append(text) + return False + + +class RegistryMenuTests(unittest.TestCase): + def run_menu(self, ui, auth, extra_modules=None): + modules = {'proxmenux_oci.ui': SimpleNamespace(UserCancelled=Cancelled), + 'proxmenux_oci.host': SimpleNamespace(storages=lambda content: [{'storage': 'local', 'avail': 10**9}])} + modules.update(extra_modules or {}) + with patch.object(accounts, 'require_root'), patch.object(accounts.importlib, 'import_module', side_effect=lambda name: modules[name]): + accounts.registry_accounts_menu(ui, engine='/fake/engine', auth=auth) + + def account(self): + return {'registry': 'ghcr.io', 'account': 'example', 'method': 'Username / token', 'status': 'Saved'} + + def auth(self, rows=None): + return SimpleNamespace(accounts=MagicMock(return_value=rows or []), login=MagicMock(), logout=MagicMock()) + + def test_native_menu_can_add_an_account_without_exposing_the_token(self): + auth = self.auth() + ui = UI(['add', None], ['ghcr.io', 'example']) + self.run_menu(ui, auth) + auth.login.assert_called_once_with('ghcr.io', 'example', 'MENU_TEST_TOKEN') + self.assertNotIn('MENU_TEST_TOKEN', repr(ui.seen)) + + def test_cancelled_password_does_not_change_credentials(self): + auth = self.auth() + ui = UI(['add', None], ['ghcr.io', 'example'], password=Cancelled()) + self.run_menu(ui, auth) + auth.login.assert_not_called() + + def test_invalid_authority_cannot_reach_login(self): + auth = self.auth() + ui = UI(['add', None], ['https://ghcr.io']) + self.run_menu(ui, auth) + auth.login.assert_not_called() + self.assertIn('registry hostname', repr(ui.seen)) + + def test_account_can_download_through_the_shared_native_cache_helper(self): + auth = self.auth([self.account()]) + ui = UI(['account-0', 'download', 'local', None], ['ghcr.io/example/private:latest']) + result = {'archive': '/cache/proxmenux-update-amd64-fixture.tar', 'size': 1024} + with patch.object(accounts, 'download_registry_image', return_value=result) as download: + self.run_menu(ui, auth) + download.assert_called_once_with('ghcr.io/example/private:latest', 'local') + self.assertIn('Image downloaded and verified', repr(ui.seen)) + + def test_selected_account_does_not_pull_from_another_registry(self): + auth = self.auth([self.account()]) + ui = UI(['account-0', 'download', None], ['quay.io/example/private:latest']) + with patch.object(accounts, 'download_registry_image') as download: + self.run_menu(ui, auth) + download.assert_not_called() + + def test_logout_requires_confirmation_in_the_menu(self): + auth = self.auth([self.account()]) + ui = UI(['account-0', 'logout', None]) + self.run_menu(ui, auth) + auth.logout.assert_not_called() + + def test_menu_offers_install_download_and_credential_management(self): + auth = self.auth([self.account()]) + ui = UI(['account-0', None, None]) + self.run_menu(ui, auth) + actions = ui.seen[1][1] + self.assertEqual([tag for tag, label in actions], ['inspect', 'download', 'install', 'login', 'logout']) + + def test_menu_preserves_the_configured_language_when_an_operation_fails(self): + from proxmenux_oci import i18n + cache = {'Saved': 'Guardado'} + with patch.object(i18n, '_language', 'es'), patch.object(i18n, '_cache', cache), \ + patch.object(accounts, 'registry_accounts_menu', side_effect=RuntimeError('test')): + with self.assertRaises(RuntimeError): + accounts.open_registry_accounts(UI([])) + self.assertEqual(i18n._language, 'es') + self.assertIs(i18n._cache, cache) + + def test_native_main_menu_dispatches_to_registry_accounts(self): + from proxmenux_oci import cli + catalog = SimpleNamespace(index_path=SimpleNamespace(exists=lambda: True), + load_index=lambda: {'applications': []}) + ui = UI(['registries', None]) + with patch.object(cli, 'interactive_ui', return_value=ui), \ + patch.object(accounts, 'open_registry_accounts') as open_accounts: + self.assertEqual(cli.interactive(catalog), 0) + open_accounts.assert_called_once_with(ui) + + def test_terminal_fallback_can_sign_in_without_exposing_the_password(self): + from proxmenux_oci.ui import TerminalUI + auth = self.auth() + output = io.StringIO() + with patch.object(accounts, 'require_root'), \ + patch('builtins.input', side_effect=['1', 'registry.test', 'example', '0']), \ + patch('getpass.getpass', return_value='MENU_TEST_TOKEN'), \ + contextlib.redirect_stdout(output): + accounts.registry_accounts_menu(TerminalUI(), auth=auth) + auth.login.assert_called_once_with('registry.test', 'example', 'MENU_TEST_TOKEN') + self.assertIn('Signed in to registry.test', output.getvalue()) + self.assertNotIn('MENU_TEST_TOKEN', output.getvalue()) + + def test_startup_configuration_errors_return_a_message_to_the_main_menu(self): + from proxmenux_oci.registry_auth import RegistryError + ui = UI([]) + with patch.object(accounts, 'registry_accounts_menu', side_effect=RegistryError('Invalid auth path')): + accounts.open_registry_accounts(ui) + self.assertIn('Invalid auth path', repr(ui.seen)) + + def test_private_image_install_uses_the_existing_native_lxc_wizard(self): + auth = self.auth([self.account()]) + ui = UI(['account-0', 'install', 'default', None], ['ghcr.io/example/private:latest']) + template = {'id': 'private-app', 'compatibility': {'automatic_install_candidate': True}} + install = MagicMock() + modules = {'proxmenux_oci.installer': SimpleNamespace(DEFAULT_MODE='default', ADVANCED_MODE='advanced'), + 'proxmenux_oci.cli': SimpleNamespace(install_template=install)} + with patch.object(accounts, 'image_template', return_value=template), \ + patch.object(accounts, 'registry_environment', contextlib.nullcontext): + self.run_menu(ui, auth, modules) + install.assert_called_once_with(ui, template, 'private-app', 'default') + + +if __name__ == '__main__': + unittest.main() diff --git a/oci/tests/test_registry_auth.py b/oci/tests/test_registry_auth.py new file mode 100644 index 00000000..1e32d5cd --- /dev/null +++ b/oci/tests/test_registry_auth.py @@ -0,0 +1,224 @@ +import base64 +import contextlib +import errno +import io +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +sys.path.insert(0, str(ROOT / "remote")) +from proxmenux_oci import registry_auth as auth_module + + +class AuthenticationTests(unittest.TestCase): + + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.directory = Path(self.temp.name).resolve() / 'containers' + self.directory.mkdir(mode=0o700) + self.path = self.directory / 'auth.json' + self.auth = auth_module.RegistryAuth(self.path) + self.token = 'TEST_TOKEN_MUST_NEVER_APPEAR_IN_OUTPUT' + self.output = io.StringIO() + + def save(self, registry='ghcr.io', username='example', token='old-token'): + data = {'auths': {registry: {'auth': base64.b64encode(f'{username}:{token}'.encode()).decode()}}} + self.path.write_text(json.dumps(data)) + self.path.chmod(0o600) + return data + + def skopeo_login(self, argv, **kwargs): + self.assertEqual(argv[:2], ['skopeo', 'login']) + self.assertNotIn(self.token, ' '.join(argv)) + self.assertNotIn('env', kwargs) + self.assertEqual(kwargs['input'], self.token + '\n') + stage = Path(argv[argv.index('--authfile') + 1]) + self.assertEqual(stage.stat().st_mode & 0o777, 0o600) + data = json.loads(stage.read_text()) + data.setdefault('auths', {})[argv[-1]] = {'auth': base64.b64encode(f'new-user:{self.token}'.encode()).decode()} + stage.write_text(json.dumps(data)) + return SimpleNamespace(returncode=0, stderr='', stdout='Login Succeeded!') + + def test_login_is_atomic_private_and_preserves_other_accounts(self): + original = self.save('docker.io') + with patch.object(auth_module.subprocess, 'run', side_effect=self.skopeo_login): + self.auth.login('ghcr.io', 'new-user', self.token) + data = json.loads(self.path.read_text()) + self.assertEqual(data['auths']['docker.io'], original['auths']['docker.io']) + self.assertIn('ghcr.io', data['auths']) + self.assertEqual(self.path.stat().st_mode & 0o777, 0o600) + self.assertEqual(list(self.directory.glob('.proxmenux-auth-*')), []) + + def test_failed_login_retains_previous_file_and_never_echoes_registry_error(self): + self.save() + before = self.path.read_bytes() + result = SimpleNamespace(returncode=1, stderr='unauthorized: ' + self.token) + with patch.object(auth_module.subprocess, 'run', return_value=result), self.assertRaises(auth_module.RegistryError) as error: + self.auth.login('ghcr.io', 'new-user', self.token) + self.assertIn('Authentication failed', str(error.exception)) + self.assertNotIn(self.token, str(error.exception)) + self.assertEqual(self.path.read_bytes(), before) + self.assertEqual(list(self.directory.glob('.proxmenux-auth-*')), []) + + def test_timed_out_login_does_not_create_a_credential_file(self): + with patch.object(auth_module.subprocess, 'run', side_effect=subprocess.TimeoutExpired('skopeo', 5)), \ + self.assertRaises(auth_module.RegistryError): + self.auth.login('ghcr.io', 'new-user', self.token, 5) + self.assertFalse(self.path.exists()) + + def test_external_credential_edit_is_not_overwritten(self): + self.save() + def concurrent_edit(argv, **kwargs): + result = self.skopeo_login(argv, **kwargs) + self.save('quay.io', 'other-task') + return result + with patch.object(auth_module.subprocess, 'run', side_effect=concurrent_edit), self.assertRaises(auth_module.RegistryError): + self.auth.login('ghcr.io', 'new-user', self.token) + self.assertEqual(set(json.loads(self.path.read_text())['auths']), {'quay.io'}) + + def test_empty_listing_does_not_create_paths(self): + path = self.directory / 'new' / 'auth.json' + auth = auth_module.RegistryAuth(path) + self.assertEqual(auth.accounts(), []) + self.assertFalse(path.parent.exists()) + + def test_logout_preserves_unrelated_accounts(self): + self.save() + data = json.loads(self.path.read_text()) + data['auths']['docker.io'] = {'auth': base64.b64encode(b'user:other-token').decode()} + self.path.write_text(json.dumps(data)) + def logout(argv, **kwargs): + stage = Path(argv[argv.index('--authfile') + 1]) + saved = json.loads(stage.read_text()) + saved['auths'].pop('ghcr.io') + stage.write_text(json.dumps(saved)) + return SimpleNamespace(returncode=0) + with patch.object(auth_module.subprocess, 'run', side_effect=logout): + self.assertTrue(self.auth.logout('ghcr.io')) + self.assertEqual(set(json.loads(self.path.read_text())['auths']), {'docker.io'}) + + def test_logout_unknown_registry_does_not_modify_file(self): + self.save() + before = self.path.read_bytes() + with patch.object(auth_module.subprocess, 'run') as run: + self.assertFalse(self.auth.logout('quay.io')) + run.assert_not_called() + self.assertEqual(self.path.read_bytes(), before) + + def test_logout_removes_repository_scopes_without_affecting_other_registries(self): + self.save('ghcr.io/private/team') + with patch.object(auth_module.subprocess, 'run') as run: + self.assertTrue(self.auth.logout('ghcr.io')) + run.assert_not_called() + self.assertFalse(json.loads(self.path.read_text())['auths']) + + def test_symlink_authfile_is_rejected_without_reading_target(self): + target = self.directory / 'other' + target.write_text('sensitive target') + self.path.symlink_to(target) + with self.assertRaises(auth_module.RegistryError): + self.auth.accounts() + self.assertEqual(target.read_text(), 'sensitive target') + + def test_insecure_directory_or_file_permissions_are_rejected(self): + self.save() + self.path.chmod(0o644) + with self.assertRaises(auth_module.RegistryError): + self.auth.accounts() + self.path.chmod(0o600) + self.directory.chmod(0o755) + with self.assertRaises(auth_module.RegistryError): + self.auth.accounts() + + def test_non_regular_credential_file_cannot_block_the_command(self): + os.mkfifo(self.path, mode=0o600) + with self.assertRaises(auth_module.RegistryError): + self.auth.accounts() + + def test_writable_parent_directory_is_rejected(self): + self.directory.parent.chmod(0o777) + self.addCleanup(self.directory.parent.chmod, 0o700) + with self.assertRaises(auth_module.RegistryError): + self.auth.accounts() + + def test_corrupt_file_is_not_overwritten(self): + self.path.write_text('{invalid') + self.path.chmod(0o600) + with patch.object(auth_module.subprocess, 'run') as run, self.assertRaises(auth_module.RegistryError): + self.auth.login('ghcr.io', 'new-user', self.token) + run.assert_not_called() + self.assertEqual(self.path.read_text(), '{invalid') + + def test_durable_authfile_is_used_by_descendants_and_environment_is_restored(self): + self.save() + with patch.dict(os.environ, {'REGISTRY_AUTH_FILE': str(self.path)}): + with auth_module.registry_environment(): + self.assertEqual(os.environ['REGISTRY_AUTH_FILE'], str(self.path)) + self.assertEqual(os.environ['REGISTRY_AUTH_FILE'], str(self.path)) + with patch.dict(os.environ, clear=True), patch.object(auth_module.Path, 'home', return_value=self.directory.parent): + # Use the normal XDG_CONFIG_HOME fallback without touching /root. + os.environ['XDG_CONFIG_HOME'] = str(self.directory.parent) + with auth_module.registry_environment(): + self.assertEqual(os.environ['REGISTRY_AUTH_FILE'], str(self.path)) + self.assertNotIn('REGISTRY_AUTH_FILE', os.environ) + + def test_account_listing_never_returns_a_password_or_token(self): + self.save(token=self.token) + with patch.object(auth_module.subprocess, 'run') as run: + rows = self.auth.accounts() + run.assert_not_called() + text = json.dumps(rows) + self.assertEqual(rows[0]['account'], 'example') + self.assertNotIn(self.token, text) + self.assertNotIn(base64.b64encode(f'example:{self.token}'.encode()).decode(), text) + + def test_invalid_staged_credentials_cannot_replace_a_working_account(self): + self.save() + before = self.path.read_bytes() + def corrupt_login(argv, **kwargs): + Path(argv[argv.index('--authfile') + 1]).write_text('{invalid') + return SimpleNamespace(returncode=0, stderr='') + with patch.object(auth_module.subprocess, 'run', side_effect=corrupt_login), \ + self.assertRaises(auth_module.RegistryError): + self.auth.login('ghcr.io', 'new-user', self.token) + self.assertEqual(self.path.read_bytes(), before) + self.assertEqual(list(self.directory.glob('.proxmenux-auth-*')), []) + + def test_credential_commit_failure_preserves_saved_file_and_cleans_staging(self): + for failure in (OSError(errno.ENOSPC, 'No space left on device'), + PermissionError(errno.EACCES, 'Permission denied')): + self.save() + before = self.path.read_bytes() + with self.subTest(error=type(failure).__name__), \ + patch.object(auth_module.subprocess, 'run', side_effect=self.skopeo_login), \ + patch.object(auth_module.os, 'replace', side_effect=failure), \ + self.assertRaises(OSError): + self.auth.login('ghcr.io', 'new-user', self.token) + self.assertEqual(self.path.read_bytes(), before) + self.assertEqual(list(self.directory.glob('.proxmenux-auth-*')), []) + + def test_failed_or_timed_out_logout_keeps_saved_accounts(self): + self.save() + before = self.path.read_bytes() + for failure in (None, subprocess.TimeoutExpired('skopeo', 30), FileNotFoundError()): + response = SimpleNamespace(returncode=1, stderr=self.token) + with self.subTest(error=type(failure).__name__), \ + patch.object(auth_module.subprocess, 'run', side_effect=failure, return_value=response), \ + self.assertRaises(auth_module.RegistryError) as error: + self.auth.logout('ghcr.io') + self.assertNotIn(self.token, str(error.exception)) + self.assertEqual(self.path.read_bytes(), before) + self.assertEqual(list(self.directory.glob('.proxmenux-auth-*')), []) + + +if __name__ == '__main__': + unittest.main() diff --git a/oci/tests/test_registry_failures.py b/oci/tests/test_registry_failures.py new file mode 100644 index 00000000..b9713e04 --- /dev/null +++ b/oci/tests/test_registry_failures.py @@ -0,0 +1,160 @@ +from contextlib import nullcontext +import errno +import io +from pathlib import Path +import subprocess +import sys +import tempfile +from types import SimpleNamespace +import unittest +from unittest.mock import MagicMock, patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +sys.path.insert(0, str(ROOT / "remote")) +from proxmenux_oci import registry_auth as auth +from proxmenux_oci import registry_accounts as accounts + + +class FailureHandlingTests(unittest.TestCase): + def setUp(self): + self.platform = patch.object(accounts, 'host_architecture', return_value='amd64') + self.platform.start() + self.addCleanup(self.platform.stop) + + def test_transport_errors_are_specific_and_never_echo_server_text(self): + cases = [ + ('unauthorized', 'Access denied'), ('denied', 'Access denied'), + ('manifest unknown', 'image or tag was not found'), + ('received unexpected HTTP status: 404 Not Found', 'image or tag was not found'), + ('toomanyrequests', 'rate limiting'), ('StatusCode: 429', 'rate limiting'), + ('500 Internal Server Error', 'server error'), ('503 Service Unavailable', 'server error'), + ('x509: certificate has expired', 'TLS verification failed'), + ('x509: certificate signed by unknown authority', 'TLS verification failed'), + ('http: server gave HTTP response to HTTPS client', 'plain HTTP'), + ('tls: first record does not look like a TLS handshake', 'plain HTTP'), + ('dial: no such host', 'Could not reach'), ('connection refused', 'Could not reach'), + ('i/o timeout', 'Could not reach'), ('connection reset', 'Could not reach'), + ('invalid character in manifest', 'invalid or unsupported'), + ('docker-credential-missing: executable file not found', 'credential helper'), + ] + for error, expected in cases: + with self.subTest(error=error): + message = auth.registry_error('registry.test', error + ' SECRET_TOKEN\x1b[31m') + self.assertIn(expected, message) + self.assertNotIn('SECRET_TOKEN', message) + self.assertNotIn('\x1b', message) + + def test_bad_login_credentials_have_a_specific_safe_error(self): + for error in ('invalid username/password', 'invalid username', 'invalid password', + 'invalid credentials', 'unauthorized', + 'Requesting bearer token: invalid status code from registry 403 (Forbidden)'): + with self.subTest(error=error): + message = auth.registry_error('registry.test', error + ' SECRET_TOKEN', login=True) + self.assertIn('Authentication failed', message) + self.assertNotIn('SECRET_TOKEN', message) + + def test_corrupt_authfile_does_not_trap_the_user_in_a_message_loop(self): + class Cancelled(RuntimeError): + pass + ui = SimpleNamespace(message=MagicMock(), choose=MagicMock()) + saved = SimpleNamespace(accounts=MagicMock(side_effect=auth.RegistryError('Invalid auth file'))) + with patch.object(accounts, 'require_root'), \ + patch.object(accounts.importlib, 'import_module', return_value=SimpleNamespace(UserCancelled=Cancelled)): + accounts.registry_accounts_menu(ui, auth=saved) + ui.message.assert_called_once() + ui.choose.assert_not_called() + + def images(self): + image = SimpleNamespace(command=MagicMock()) + def resolve(ref, architecture): + image.command('skopeo', 'inspect', '--raw', 'docker://' + ref) + return {'manifest_digest': 'sha256:' + 'a' * 64, 'size': 10, 'version': '1.0'} + image.resolve_candidate = resolve + return image + + def test_native_resolver_transport_hook_is_scoped_and_restored(self): + image = self.images() + original = image.command + with patch.object(accounts.importlib, 'import_module', return_value=image), \ + patch.object(accounts, 'registry_environment', nullcontext), \ + patch.object(accounts.subprocess, 'run', return_value=SimpleNamespace(returncode=0, stdout=b'{}')): + self.assertEqual(accounts.inspect_registry_image('registry.test/team/app:latest')['version'], '1.0') + self.assertIs(image.command, original) + + def test_native_resolver_restores_transport_hook_after_access_denied(self): + image = self.images() + original = image.command + response = SimpleNamespace(returncode=1, stderr=b'unauthorized SECRET_TOKEN') + with patch.object(accounts.importlib, 'import_module', return_value=image), \ + patch.object(accounts, 'registry_environment', nullcontext), \ + patch.object(accounts.subprocess, 'run', return_value=response), \ + self.assertRaises(auth.RegistryError) as error: + accounts.inspect_registry_image('registry.test/team/app:latest') + self.assertIn('Access denied', str(error.exception)) + self.assertNotIn('SECRET_TOKEN', str(error.exception)) + self.assertIs(image.command, original) + + def test_image_query_timeout_does_not_fall_back_to_a_long_native_timeout(self): + image = self.images() + with patch.object(accounts.importlib, 'import_module', return_value=image), \ + patch.object(accounts, 'registry_environment', nullcontext), \ + patch.object(accounts.subprocess, 'run', side_effect=subprocess.TimeoutExpired('skopeo', 30)), \ + self.assertRaises(auth.RegistryError) as error: + accounts.inspect_registry_image('registry.test/team/app:latest') + self.assertIn('within 30 seconds', str(error.exception)) + + def test_wrong_architecture_and_malformed_metadata_do_not_recommend_relogin(self): + for native_error, expected in [(ValueError('A single matching image platform cannot be resolved'), 'compatible Linux'), + (ValueError('Invalid JSON'), 'invalid or unsupported'), + (KeyError('config'), 'incomplete or unsupported'), + (AttributeError('bad config'), 'incomplete or unsupported')]: + image = self.images() + image.resolve_candidate = MagicMock(side_effect=native_error) + with self.subTest(error=type(native_error).__name__), \ + patch.object(accounts.importlib, 'import_module', return_value=image), \ + patch.object(accounts, 'registry_environment', nullcontext), \ + self.assertRaises(auth.RegistryError) as error: + accounts.inspect_registry_image('registry.test/team/app:latest') + self.assertIn(expected, str(error.exception)) + + def test_storage_failures_have_storage_messages_and_keep_native_lock_scope(self): + for failure, expected in [(ValueError('The new image does not fit on the storage'), 'Not enough free space'), + (OSError(errno.ENOSPC, 'No space left on device'), 'Not enough free space'), + (PermissionError('denied'), 'Cannot write'), + (OSError(errno.EROFS, 'Read-only file system'), 'Cannot write'), + (ValueError('Unsafe OCI archive path'), 'unsafe'), + (RuntimeError('Could not obtain an intact image'), 'integrity verification')]: + native = SimpleNamespace(instances=SimpleNamespace(ROOT='/fixture', locked=lambda root: nullcontext()), + resolve_archive=MagicMock(side_effect=failure)) + with self.subTest(error=str(failure)), patch.object(accounts, 'inspect_registry_image'), \ + patch.object(accounts.importlib, 'import_module', return_value=native), \ + patch.object(accounts, 'registry_environment', nullcontext), \ + self.assertRaises(auth.RegistryError) as error: + accounts.download_registry_image('registry.test/team/app:latest', 'local') + self.assertIn(expected, str(error.exception)) + + def test_missing_skopeo_does_not_create_a_saved_account(self): + with tempfile.TemporaryDirectory() as temp: + path = Path(temp).resolve() / 'containers/auth.json' + manager = auth.RegistryAuth(path) + with patch.object(auth.subprocess, 'run', side_effect=FileNotFoundError()), \ + self.assertRaises(auth.RegistryError) as error: + manager.login('registry.test', 'example', 'example-token') + self.assertIn('Skopeo is not available', str(error.exception)) + self.assertFalse(path.exists()) + + def test_invalid_usernames_and_multiline_or_empty_tokens_never_reach_skopeo(self): + with tempfile.TemporaryDirectory() as temp: + manager = auth.RegistryAuth(Path(temp).resolve() / 'containers/auth.json') + for username, token in [('', 'token'), (' ', 'token'), ('user:colon', 'token'), + ('user\x7f', 'token'), ('name', ''), ('name', 'x\ny'), + ('name', 'x\0y'), ('name', 'x' * 65537)]: + with self.subTest(username=username, token_length=len(token)), \ + patch.object(auth.subprocess, 'run') as run, self.assertRaises(auth.RegistryError): + manager.login('registry.test', username, token) + run.assert_not_called() + + +if __name__ == '__main__': + unittest.main() diff --git a/oci/tests/test_registry_references.py b/oci/tests/test_registry_references.py new file mode 100644 index 00000000..bfb9f22f --- /dev/null +++ b/oci/tests/test_registry_references.py @@ -0,0 +1,45 @@ +from pathlib import Path +import json +import sys +import unittest +from types import SimpleNamespace +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "src")) +sys.path.insert(0, str(ROOT / "remote")) +from proxmenux_oci.registry_auth import RegistryError, validate_registry, validate_image + + +class ReferenceTests(unittest.TestCase): + def test_native_image_importer_preserves_registry_port_and_digest(self): + from proxmenux_oci import custom + import yaml + for reference in ('registry.test:5443/team/app:stable', + '[::1]:5443/team/app@sha256:' + 'a' * 64): + with self.subTest(reference=reference), \ + patch.object(custom.subprocess, 'run', return_value=SimpleNamespace( + returncode=0, stdout=json.dumps({'config': {}}))) as run: + compose = yaml.safe_load(custom.compose_from_image(reference)) + self.assertEqual(next(iter(compose['services'].values()))['image'], reference) + self.assertEqual(run.call_args.args[0][-1], 'docker://' + reference) + + def test_private_registries_with_ports_and_digest_references_are_valid(self): + for reference in ('ghcr.io/team/app:latest', 'registry.local:5000/team/app:stable', + '[::1]:5000/team/app:latest', 'ghcr.io/team/app@sha256:' + 'a' * 64): + self.assertEqual(validate_image(reference), reference) + + def test_credentials_and_urls_cannot_be_used_as_image_references(self): + for reference in ('https://ghcr.io/team/app', 'user:token@ghcr.io/app', 'ghcr.io/app;id', + 'ghcr.io/../app', 'ghcr.io/app@sha256:short'): + with self.assertRaises(RegistryError): + validate_image(reference) + + def test_registry_authorities_do_not_accept_image_paths_or_credentials(self): + for registry in ('https://ghcr.io', 'ghcr.io/team', 'user:token@ghcr.io', 'host:99999'): + with self.assertRaises(RegistryError): + validate_registry(registry) + + +if __name__ == '__main__': + unittest.main() From 6fc9f9003b40a14cc03a1aa9d0e83b4f52f84cb3 Mon Sep 17 00:00:00 2001 From: Jonatan <190036+jcastro@users.noreply.github.com> Date: Fri, 16 Oct 2026 20:03:58 +0200 Subject: [PATCH 2/2] Update OCI wording tests for private registry support --- .github/scripts/tests/test_oci_wording.py | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/.github/scripts/tests/test_oci_wording.py b/.github/scripts/tests/test_oci_wording.py index bae032fa..fd357378 100644 --- a/.github/scripts/tests/test_oci_wording.py +++ b/.github/scripts/tests/test_oci_wording.py @@ -17,13 +17,15 @@ MENU = ROOT / 'oci/src/proxmenux_oci/management.py' CHOICE = 'Recover the previous installation' BEFORE = 'The operation stopped halfway. Choose "{choice}" for this container in the OCI management menu.' AFTER = 'The recovery did not complete. Review the log and choose "{choice}" again for this container in the OCI management menu.' +REGISTRY_ADVICE = ('Check the image reference and registry access. If the registry is unavailable, ' + 'try again later. For a private image, sign in from Registry accounts in the OCI menu.') NEW_KEYS = ( 'The Compose file describes several services:', 'Only one service at a time can be installed this way.', 'No docker run command was given', 'Image selection was cancelled', 'Could not inspect the image in its registry:', - 'Check the image reference and registry access. If the registry is unavailable, try again later. A private registry needs credentials, which are not supported yet.', + REGISTRY_ADVICE, BEFORE, AFTER, ) @@ -117,16 +119,15 @@ class OciWordingTests(TestCase): self.assertEqual(report('example:latest'), (True, 'The image is in its registry: amd64')) self.assertEqual(run.call_count, 1) - def test_failed_registry_inspection_has_neutral_advice(self): + def test_failed_registry_inspection_points_to_registry_accounts(self): ui = SimpleNamespace(message=Mock()) fn = function(CUSTOM, 'explore', read_definition=lambda _: ('definition', 'source'), ignored_settings=lambda _: [], template_from_compose=lambda _: { 'container_contract': {'image': {'reference': 'example:latest'}}}, describe=lambda _: 'SUMMARY', registry_report=lambda _: (False, 'INSPECTION FAILED')) fn(ui) - self.assertEqual(ui.message.call_args.args[0], 'SUMMARY\n\nINSPECTION FAILED\n\n' - 'Check the image reference and registry access. If the registry is unavailable, ' - 'try again later. A private registry needs credentials, which are not supported yet.') + self.assertEqual(ui.message.call_args.args[0], 'SUMMARY\n\nINSPECTION FAILED\n\n' + REGISTRY_ADVICE) + self.assertNotIn('not supported yet', ui.message.call_args.args[0]) def test_recovery_hint_quotes_translated_choice_in_all_locales(self): # Pin the real translated management choice, not an English AST constant alone.