mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-02 19:02:31 +00:00
Add private registry accounts to OCI manager Apps
This commit is contained in:
Jonatan
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user