Mirrors/ProxMenux
1
0
Fork 0
mirror of https://github.com/MacRimi/ProxMenux.git synced 2026-11-01 10:22:08 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/develop' into HEAD

Browse Source 
# Conflicts:
#	.github/workflows/test-oci-registries.yml
This commit is contained in:
MacRimi
2026-10-18 10:16:34 +02:00
parent ac42bea70f 41ef07ac55
commit bef2f837c2
76 changed files with 4239 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files
oci/tests/test_carried_config_cache.py
+134
View File
@@ -0,0 +1,134 @@
"""A fast Manage entry still detects config edits and verifies lifecycle work."""
import contextlib
import copy
import json
import os
from pathlib import Path
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 / 'remote'))
import oci_carried_record as carried
class CarriedConfigCacheTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.root = Path(self.tmp.name) / 'instances'
self.source = Path(self.tmp.name) / 'lxc'
self.source.mkdir()
self.config = self.source / '120.conf'
self.config.write_text('# proxmenux-instance=example\nunprivileged: 1\n')
(self.root / '120').mkdir(parents=True)
self.cache = self.root / '120' / carried.CONFIG_CACHE
self.patch = patch.object(carried, 'LOCAL_CONFIGS', self.source)
self.patch.start()
self.addCleanup(self.patch.stop)
def response(self, text='description: proxmenux-instance=example\nunprivileged: 1\n', status=0):
return SimpleNamespace(returncode=status, stdout=text)
def test_warm_reads_avoid_pct_and_cache_is_private(self):
with patch.object(carried, '_run', return_value=self.response()) as run:
first = carried.configuration(self.root, 120)
self.assertEqual(carried.configuration(self.root, 120), first)
run.assert_called_once_with('pct', 'config', '120')
self.assertEqual(self.cache.stat().st_mode & 0o777, 0o600)
def test_native_edits_invalidate_even_when_size_and_mtime_are_restored(self):
with patch.object(carried, '_run', side_effect=[self.response(), self.response('new')]) as run:
carried.configuration(self.root, 120)
info = self.config.stat()
self.config.write_text(self.config.read_text().replace('example', 'changed'))
os.utime(self.config, ns=(info.st_atime_ns, info.st_mtime_ns))
self.assertEqual(carried.configuration(self.root, 120), 'new')
self.assertEqual(run.call_count, 2)
def test_verify_forces_a_fresh_pct_read_with_a_warm_cache(self):
with patch.object(carried, '_run', side_effect=[self.response(), self.response('verified')]) as run:
carried.configuration(self.root, 120)
self.assertEqual(carried.configuration(self.root, 120, verify=True), 'verified')
self.assertEqual(run.call_count, 2)
def test_missing_native_source_never_uses_or_populates_a_cache(self):
self.config.unlink()
with patch.object(carried, '_run', return_value=self.response()) as run:
carried.configuration(self.root, 120)
carried.configuration(self.root, 120)
self.assertEqual(run.call_count, 2)
self.assertFalse(self.cache.exists())
def test_concurrent_edit_does_not_cache_output_for_the_wrong_revision(self):
def command(*args):
self.config.write_text('changed while pct was reading\n')
return self.response()
with patch.object(carried, '_run', side_effect=command):
carried.configuration(self.root, 120)
self.assertFalse(self.cache.exists())
def test_edit_during_cache_lookup_falls_back_to_pct(self):
with patch.object(carried, '_run', return_value=self.response()):
carried.configuration(self.root, 120)
read_text = Path.read_text
def edit_during_read(path, *args, **kwargs):
text = read_text(path, *args, **kwargs)
if path == self.cache:
self.config.write_text('changed while reading the cache\n')
return text
with patch.object(Path, 'read_text', edit_during_read), \
patch.object(carried, '_run', return_value=self.response('fresh')) as run:
self.assertEqual(carried.configuration(self.root, 120), 'fresh')
run.assert_called_once_with('pct', 'config', '120')
def test_corrupt_cache_and_pct_failure_are_not_accepted(self):
for text in ('not json', '[]', '{"schema_version": 999}'):
with self.subTest(cache=text):
self.cache.write_text(text)
with patch.object(carried, '_run', return_value=self.response(status=1)) as run:
self.assertIsNone(carried.configuration(self.root, 120))
run.assert_called_once()
self.assertEqual(self.cache.read_text(), text)
def test_symlink_cache_does_not_read_or_replace_its_target(self):
target = Path(self.tmp.name) / 'outside'
target.write_text('private original')
self.cache.symlink_to(target)
with patch.object(carried, '_run', return_value=self.response()) as run:
self.assertEqual(carried.configuration(self.root, 120), self.response().stdout)
run.assert_called_once()
self.assertTrue(self.cache.is_symlink())
self.assertEqual(target.read_text(), 'private original')
def test_cached_config_still_detects_a_restored_container_record(self):
installation = '488ed3cb-1145-477a-b90f-c46777d69fb2'
record = {'schema_version': 1, 'vmid': 120, 'installation_id': installation,
'status': 'installed', 'deployment': {'hostname': 'current'}}
carried.instances.write(carried.instances.location(self.root, 120), record)
native = f'description: proxmenux-instance={installation}\n'
with patch.object(carried, '_run', return_value=self.response(native)):
carried.configuration(self.root, 120)
rootfs = Path(self.tmp.name) / 'rootfs'
rootfs.mkdir()
previous = copy.deepcopy(record)
previous['deployment']['hostname'] = 'restored-snapshot'
carried.write_copy(rootfs, {'kind': carried.KIND, 'schema_version': 1,
'record': previous, 'generation': 'restored-generation'},
(os.getuid(), os.getgid()))
stamp = self.root / '120' / carried.STAMP
stamp.write_text(json.dumps({'generation': 'host-generation'}))
with patch.object(carried, '_run', side_effect=AssertionError('cached config expected')), \
patch.object(carried, 'read_cluster', return_value=None), \
patch.object(carried, 'running_pid', return_value=123), \
patch.object(carried, 'container_root', return_value=contextlib.nullcontext(rootfs)):
self.assertEqual(carried.carry(self.root, 120, mount_stopped=False), 'stale')
self.assertFalse(carried.instances.location(self.root, 120).exists())
self.assertTrue((self.root / '120' / f'retired-{installation}.json').exists())
if __name__ == '__main__':
unittest.main()
oci/tests/test_dialog_escape.py
+34
View File
@@ -0,0 +1,34 @@
"""Escape delay is scoped to dialog and preserves terminal-specific overrides."""
import os
from pathlib import Path
import subprocess
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'src'))
from proxmenux_oci.ui import DialogUI, UserCancelled
class DialogEscapeTests(unittest.TestCase):
def test_escape_uses_short_delay_without_changing_parent_environment(self):
with patch.dict(os.environ, {'TERM': 'xterm'}, clear=True), \
patch('proxmenux_oci.ui.subprocess.run', return_value=subprocess.CompletedProcess([], 255, stderr='')) as run:
self.assertIsNone(DialogUI().choose('Menu', [('1', 'First')]))
self.assertEqual(run.call_args.kwargs['env']['ESCDELAY'], '150')
self.assertNotIn('ESCDELAY', os.environ)
def test_explicit_delay_is_preserved(self):
with patch.dict(os.environ, {'ESCDELAY': '900'}), \
patch('proxmenux_oci.ui.subprocess.run', return_value=subprocess.CompletedProcess([], 0, stderr='1')) as run:
self.assertEqual(DialogUI().choose('Menu', [('1', 'First')]), '1')
self.assertEqual(run.call_args.kwargs['env']['ESCDELAY'], '900')
def test_escape_in_search_input_keeps_cancellation_behavior(self):
with patch('proxmenux_oci.ui.subprocess.run', return_value=subprocess.CompletedProcess([], 255, stderr='')):
with self.assertRaises(UserCancelled):
DialogUI().ask('Search')
if __name__ == '__main__':
unittest.main()
oci/tests/test_external_tags.py
+203
View File
@@ -0,0 +1,203 @@
"""Proxmox tags can change without preventing OCI management or being lost."""
import copy
import io
import json
import os
from pathlib import Path
import re
import shutil
import subprocess
import sys
import tempfile
import unittest
from contextlib import redirect_stdout
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'remote'))
import oci_instance_reconcile as reconcile
import oci_instance_transaction as transaction
from oci_installation_state import sha
MARKER = '11111111-1111-4111-8111-111111111111'
BASE = (f'description: proxmenux-instance={MARKER}\n'
'rootfs: local-lvm:vm-200-disk-0,size=8G\n'
'unprivileged: 1\ncores: 2\nmemory: 512\n'
'net0: name=eth0,bridge=vmbr0,hwaddr=02:00:00:00:00:01,ip=dhcp\n'
'net1: name=eth1,bridge=vmbr1,ip=dhcp\nstartup: order=20\n').encode()
def config(tags):
return BASE + (f'tags: {tags}\n'.encode() if tags is not None else b'')
def record(tags='oci;proxmenux'):
before = config(tags)
return {
'schema_version': 1, 'vmid': 200, 'status': 'installed',
'installation_id': MARKER,
'template': {'id': 'test-app', 'container_contract': {'volumes': []}},
'deployment': {
'vmid': 200, 'rootfs': {'storage': 'local-lvm', 'size_gb': 8},
'network': {}, 'resources': {'cores': 2, 'memory_mb': 512},
'mounts': [], 'devices': [],
},
'observed': {
'config': before.decode(), 'config_sha256': sha(before),
'image': {'architecture': 'amd64', 'manifest_digest': 'sha256:old',
'defaults': {'Volumes': {}}},
},
}
class ExternalTagsTests(unittest.TestCase):
def test_adding_editing_reordering_and_removing_tags_need_no_adoption(self):
for old, new in ((None, 'homelab'), ('oci;proxmenux', 'homelab;oci;proxmenux'),
('oci;proxmenux', 'proxmenux;oci'), ('oci;proxmenux', None),
('oci;proxmenux', '')):
with self.subTest(old=old, new=new):
saved, current = record(old), config(new)
self.assertIsNone(reconcile.propose(saved, current))
self.assertEqual(transaction.external_changes(saved, current), {})
def test_tags_can_change_together_with_cpu_and_memory(self):
current = config('homelab').replace(b'cores: 2', b'cores: 4').replace(b'memory: 512', b'memory: 2048')
self.assertIsNone(reconcile.propose(record(), current))
self.assertEqual(transaction.external_changes(record(), current), {'cores': 4, 'memory': 2048})
def test_tag_changes_do_not_bypass_identity_or_network_checks(self):
for current in (config('homelab').replace(MARKER.encode(), b'22222222-2222-4222-8222-222222222222'),
config('homelab').replace(b'bridge=vmbr0', b'bridge=vmbr2')):
with self.subTest(config=current):
with self.assertRaises(ValueError):
reconcile.propose(record(), current)
with self.assertRaises(ValueError):
transaction.external_changes(record(), current)
def test_new_volume_still_requires_adoption_when_tags_also_change(self):
saved = record()
current = config('homelab') + b'mp0: local-lvm:vm-200-disk-1,mp=/data,size=8G,backup=1\n'
with tempfile.TemporaryDirectory() as folder:
root = Path(folder)
path = root / '200/oci-compose.json'
path.parent.mkdir()
path.write_text(json.dumps(saved))
with patch.object(reconcile.instances, 'ROOT', root):
proposal = reconcile.propose(saved, current)
self.assertEqual(proposal['candidate']['deployment']['mounts'][0]['container_path'], '/data')
self.assertEqual(len(proposal['details']), 1)
self.assertEqual(json.loads(path.read_text()), saved)
class Prepared(Exception):
"""Stop a mocked transaction before it can stop or modify any container."""
class TagReplayTests(unittest.TestCase):
def prepare(self, folder, tags, operation, coordinated=None):
current, saved = config(tags), record()
archive = folder / 'image.tar'
archive.write_bytes(b'fixture')
states = []
def checkpoint(_path, state, phase):
self.assertEqual(phase, 'prepared')
states.append(copy.deepcopy(state))
raise Prepared()
def command(*args):
if args[:2] == ('pct', 'config'):
return current
if args[:3] == ('pvesh', 'get', '/cluster/ha/resources'):
return b'[]'
if args[:2] == ('pct', 'status'):
return b'status: running'
self.fail(f'Unexpected host command: {args}')
proposal = {'operation': 'recreate', 'candidate': copy.deepcopy(saved)} if operation == 'recreate' else None
image = {'architecture': 'amd64', 'os': 'linux', 'manifest_digest': 'sha256:new', 'defaults': {}}
with (patch.object(transaction.instances, 'read', return_value=saved),
patch.object(transaction, 'run', side_effect=command),
patch.object(transaction, 'check_archive'),
patch.object(transaction, 'image_from_archive', return_value=image),
patch.object(transaction.work_backup, 'locate', side_effect=lambda directory, *_: directory),
patch.object(transaction, 'open_log'),
patch.object(transaction, 'checkpoint', side_effect=checkpoint),
redirect_stdout(io.StringIO()), self.assertRaises(Prepared)):
transaction.apply(folder, 200, archive, operation, proposal=proposal, coordinated=coordinated)
return states[0]
@unittest.skipUnless(shutil.which('bash') and shutil.which('jq'), 'bash and jq are needed')
def test_update_and_modify_replay_current_tags_including_no_tags(self):
installer = (ROOT / 'remote/install_oci.sh').read_text()
function = re.search(r'^apply_kept_settings\(\) \{\n.*?^\}\n', installer, re.M | re.S)
self.assertIsNotNone(function)
for operation in ('update', 'recreate'):
for tags in ('homelab;oci;proxmenux', 'oci', None, ''):
with self.subTest(operation=operation, tags=tags), tempfile.TemporaryDirectory() as name:
folder = Path(name)
state = self.prepare(folder, tags, operation)
kept = state['runtime_deployment']['kept_proxmox_settings']
deployment = folder / 'deployment.json'
deployment.write_text(json.dumps({'kept_proxmox_settings': kept}))
calls = folder / 'calls'
script = '''
pct() { printf '%s\\0' "$@" >> "$CALLS"; printf '\\n' >> "$CALLS"; }
oci_quiet() { "$@"; }
translate() { printf '%s' "$1"; }
die() { printf '%s\\n' "$*" >&2; exit 1; }
'''
done = subprocess.run(['bash', '-euc', script + function.group(0) + '\napply_kept_settings'],
env=dict(os.environ, VMID='200', DEPLOYMENT_FILE=str(deployment), CALLS=str(calls)),
capture_output=True, text=True)
self.assertEqual(done.returncode, 0, done.stderr)
commands = [line.split(b'\0')[:-1] for line in calls.read_bytes().splitlines()]
expected = [b'set', b'200', b'--tags', tags.encode()] if tags else [b'set', b'200', b'--delete', b'tags']
self.assertIn(expected, commands)
self.assertIn([b'set', b'200', b'--net1', b'name=eth1,bridge=vmbr1,ip=dhcp'], commands)
self.assertIn([b'set', b'200', b'--startup', b'order=20'], commands)
def test_stack_operations_also_preserve_tags_and_their_absence(self):
for operation in ('update', 'recreate'):
for tags in ('homelab;oci', None, ''):
with self.subTest(operation=operation, tags=tags), tempfile.TemporaryDirectory() as name:
state = self.prepare(Path(name), tags, operation, coordinated={'stack': 'test'})
self.assertEqual(state['preserved_stack_config']['tags'], tags or '')
@unittest.skipUnless(shutil.which('bash') and shutil.which('jq'), 'bash and jq are needed')
def test_native_stack_check_accepts_tag_edits_and_nothing_else(self):
saved = record()
instances = transaction.instances
for new in ('homelab;oci;proxmenux', 'proxmenux;oci', '', None):
with self.subTest(new=new):
self.assertTrue(instances.same_config_except_notes(saved, config(new), ignore_tags=True))
self.assertFalse(instances.same_config_except_notes(saved, config(new)))
changed = config('homelab').replace(b'cores: 2', b'cores: 4')
self.assertFalse(instances.same_config_except_notes(saved, changed, ignore_tags=True))
other = config('homelab').replace(MARKER.encode(), b'22222222-2222-4222-8222-222222222222')
self.assertFalse(instances.same_config_except_notes(saved, other, ignore_tags=True))
source = (ROOT / 'remote/oci_stack_native.py').read_text()
self.assertIn('same_config_except_notes(record, config, ignore_tags=True)', source)
def test_installer_still_refuses_unsupported_kept_settings(self):
installer = (ROOT / 'remote/install_oci.sh').read_text()
function = re.search(r'^apply_kept_settings\(\) \{\n.*?^\}\n', installer, re.M | re.S).group(0)
with tempfile.TemporaryDirectory() as name:
deployment = Path(name) / 'deployment.json'
deployment.write_text(json.dumps({'kept_proxmox_settings': {'unprivileged': '0'}}))
script = '''
oci_quiet() { echo 'must not run' >&2; exit 99; }
translate() { printf '%s' "$1"; }
die() { printf '%s\\n' "$*" >&2; exit 1; }
'''
done = subprocess.run(['bash', '-euc', script + function + '\napply_kept_settings'],
env=dict(os.environ, VMID='200', DEPLOYMENT_FILE=str(deployment)),
capture_output=True, text=True)
self.assertEqual(done.returncode, 1)
self.assertIn('Unsupported kept setting: unprivileged', done.stderr)
self.assertNotIn('must not run', done.stderr)
if __name__ == '__main__':
unittest.main()
oci/tests/test_guessed_endpoint_healthcheck.py
+87
View File
@@ -0,0 +1,87 @@
"""A web address guessed from the compose file may belong to a service that does
not speak HTTP. Updates, Modify and recovery must accept its answer instead of
waiting for an HTTP reply that never comes."""
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / 'remote'))
import oci_instance_transaction as transaction
INSTALLER = ROOT / 'remote/install_oci.sh'
GUESSED = {'first_run': {'endpoints': [{'label': 'Web UI', 'scheme': 'http', 'port': 3306, 'path': '/',
'source': 'compose-first-tcp-port-fallback'}]}}
DECLARED = {'first_run': {'endpoints': [{'label': 'Web UI', 'scheme': 'http', 'port': 8080, 'path': '/'}]}}
def probe_function():
source = INSTALLER.read_text()
start = source.index('healthcheck_probe() {')
end = source.index('\n}\n', start) + 3
return source[start:end]
class GuessedEndpointCheckTests(unittest.TestCase):
def test_only_a_guessed_endpoint_accepts_other_protocols(self):
self.assertTrue(transaction.effective_healthcheck(GUESSED)['accept_non_http'])
self.assertNotIn('accept_non_http', transaction.effective_healthcheck(DECLARED))
def run_probe(self, exit_code, code='000', accept_non_http='true', any_status='true'):
with tempfile.TemporaryDirectory() as directory:
fake = Path(directory) / 'curl'
fake.write_text(f'#!/bin/sh\nprintf %s {code}\nexit {exit_code}\n')
fake.chmod(0o755)
script = (probe_function() + '\nCURL_ARGS=()\nHC_URL=http://192.0.2.1:3306/\n'
f'HC_ANY_STATUS={any_status}\nHC_ACCEPT_NON_HTTP={accept_non_http}\nhealthcheck_probe\n')
env = {**os.environ, 'PATH': f'{directory}:{os.environ["PATH"]}'}
return subprocess.run(['bash', '-c', script], env=env, capture_output=True).returncode
def test_installer_probe_accepts_a_non_http_answer_only_when_guessed(self):
for status in transaction.NON_HTTP_ANSWER:
with self.subTest(status=status):
self.assertEqual(self.run_probe(status), 0)
self.assertNotEqual(self.run_probe(status, accept_non_http='false'), 0)
# A closed port, a timeout or a server error still fail.
for status, code in ((7, '000'), (28, '000'), (0, '503')):
with self.subTest(status=status, code=code):
self.assertNotEqual(self.run_probe(status, code), 0)
self.assertEqual(self.run_probe(0, '200'), 0)
self.assertEqual(self.run_probe(0, any_status='false', accept_non_http='false'), 0)
def test_recovery_check_accepts_a_database_answer(self):
check = transaction.effective_healthcheck(GUESSED)
check.update(timeout_seconds=6, stability_seconds=0)
template = {'proxmox': {'installer_profile': {'startup_healthcheck': check}}}
def run(*args):
if args[:2] == ('pct', 'status'):
return b'status: running\n'
if args[:2] == ('pct', 'config'):
return b''
if args[0] == 'lxc-info':
return b'192.0.2.10\n'
raise AssertionError(args)
with patch.object(transaction, 'run', side_effect=run), \
patch.object(transaction, 'probe_http', return_value=(b'000', 1)), \
patch.object(transaction, 'log'):
self.assertEqual(transaction.healthcheck(113, template), 'http://192.0.2.10:3306/')
declared = transaction.effective_healthcheck(DECLARED)
declared.update(timeout_seconds=4, stability_seconds=0)
template = {'proxmox': {'installer_profile': {'startup_healthcheck': declared}}}
with patch.object(transaction, 'run', side_effect=run), \
patch.object(transaction, 'probe_http', return_value=(b'000', 1)), \
patch.object(transaction, 'log'), patch.object(transaction.time, 'sleep'):
with self.assertRaises(ValueError):
transaction.healthcheck(113, template)
if __name__ == '__main__':
unittest.main()
oci/tests/test_manage_sync_notice.py
+34
View File
@@ -0,0 +1,34 @@
"""Opening Manage shows the sync notice only while some configuration is still unread."""
from pathlib import Path
import sys
import tempfile
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 management
import oci_carried_record as carried
class ManageSyncNoticeTests(unittest.TestCase):
def test_notice_only_while_an_installed_application_has_no_cache(self):
rows = [{'vmid': 101, 'status': 'installed', 'pending': False},
{'vmid': 102, 'status': 'installed', 'pending': True},
{'vmid': 103, 'status': 'failed', 'pending': False}]
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
with patch.object(carried.instances, 'ROOT', root), \
patch.object(management, 'saved_inventory', return_value=rows):
self.assertTrue(management._records_unread(ROOT))
cache = carried.instances.location(root, 101).parent / carried.CONFIG_CACHE
cache.parent.mkdir(parents=True)
cache.write_text('{}')
# The pending and failed applications are never cached by the sync.
self.assertFalse(management._records_unread(ROOT))
if __name__ == '__main__':
unittest.main()
oci/tests/test_menu_performance.py
+109
View File
@@ -0,0 +1,109 @@
"""Menu speedups avoid unused probes, retaining confirmation and capacity checks."""
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import Mock,patch
ROOT=Path(__file__).resolve().parents[1]
sys.path[:0]=[str(ROOT/'src'),str(ROOT/'remote')]
from proxmenux_oci import management as menu
import oci_instances as instances
import oci_instance_reconcile as reconcile
import oci_remove as remover
import oci_work_backup as backup
class UpdateEntryTests(unittest.TestCase):
def run_update(self,approved,ready):
events=[]
record={'status':'installed','vmid':110,'deployment':{},'template':{}}
row={'vmid':110,'reason':'matched','stack':False,'pending':False,'status':'installed'}
ui=Mock();ui.unattended=True
ui.review.side_effect=lambda *a,**kw:events.append('review') or approved
with patch.object(menu,'carry_records',return_value={}) as carry, \
patch.object(menu,'check_selected',return_value=row), \
patch.object(menu,'_selected_configuration',return_value=b'current config'), \
patch.object(instances,'read',return_value=record),patch.object(reconcile,'propose',return_value=None), \
patch.object(menu,'_work_backup_ready',side_effect=lambda *a:events.append('backup') or ready), \
patch.object(menu,'_run_lifecycle',side_effect=lambda *a:events.append('lifecycle') or True):
result=menu.manage_instance(ROOT,ui,row,action='update')
carry.assert_called_once_with(ROOT,vmids=[110],verify=True)
return result,events
def test_declining_update_does_not_probe_disks_or_storages(self):
result,events=self.run_update(False,False)
self.assertFalse(result);self.assertEqual(events,['review'])
def test_accepted_update_checks_backup_space_before_starting(self):
result,events=self.run_update(True,True)
self.assertTrue(result);self.assertEqual(events,['review','backup','lifecycle'])
def test_insufficient_space_still_prevents_the_update(self):
result,events=self.run_update(True,False)
self.assertFalse(result);self.assertEqual(events,['review','backup'])
class BackupDestinationTests(unittest.TestCase):
def status(self,local_free,saved_free=None,only_needed=True):
with tempfile.TemporaryDirectory() as directory:
root=Path(directory);saved= root/'nas';saved.mkdir()
record={'deployment':{backup.KEY:'nas'}}
with patch.object(backup,'application',return_value=(110,[110,111])), \
patch.object(instances,'read',return_value=record), \
patch.object(backup,'required',return_value=100) as required, \
patch.object(backup,'available',side_effect=lambda p:saved_free if Path(p)==saved else local_free), \
patch.object(backup,'storage_path',return_value=saved if saved_free is not None else None) as path, \
patch.object(backup,'file_storages',return_value=[{'storage':'enough','free':200},{'storage':'small','free':50}]) as alternatives:
result=backup.status(root,110,only_needed=only_needed)
required.assert_called_once_with([110,111])
return result,path.call_count,alternatives.call_count
def test_local_room_avoids_saved_storage_and_alternative_queries(self):
result,saved,alternatives=self.status(200,saved_free=1000)
self.assertTrue(result['fits']);self.assertEqual((saved,alternatives),(0,0))
def test_saved_destination_that_fits_avoids_alternative_queries(self):
result,saved,alternatives=self.status(50,saved_free=200)
self.assertTrue(result['saved_fits']);self.assertEqual((saved,alternatives),(1,0))
def test_when_neither_fits_all_available_destinations_are_checked(self):
result,saved,alternatives=self.status(50,saved_free=40)
self.assertFalse(result['fits']);self.assertFalse(result['saved_fits'])
self.assertEqual(result['candidates'],[{'storage':'enough','free':200}])
self.assertEqual(alternatives,1)
def test_missing_saved_storage_still_offers_alternatives(self):
result,saved,alternatives=self.status(50)
self.assertEqual(result['candidates'],[{'storage':'enough','free':200}])
self.assertEqual(alternatives,1)
def test_complete_status_keeps_legacy_fields_and_all_candidates(self):
result,saved,alternatives=self.status(200,saved_free=300,only_needed=False)
self.assertTrue(result['fits']);self.assertTrue(result['saved_fits'])
self.assertEqual((saved,alternatives),(1,1))
self.assertEqual(result['candidates'],[{'storage':'enough','free':200}])
class RemovalPreviewTests(unittest.TestCase):
def preview(self,reader):
record={'vmid':110,'deployment':{},'template':{'catalog_ui':{'title':'Example'}}}
with patch.object(remover,'members_of',return_value=(110,record,[110])), \
patch.object(instances,'read',return_value=record), \
patch.object(menu,'_selected_configuration',side_effect=reader), \
patch.object(remover,'guest_config',side_effect=AssertionError('Duplicate pct process')), \
patch.object(remover,'host_directories',return_value=[]), \
patch.object(remover,'private_bridge',return_value=None):
return menu._removal_summary(ROOT,110)
def test_preview_reuses_verified_config_and_keeps_volume_and_host_path_details(self):
config=b'hostname: example\nrootfs: local:subvol-110-disk-0,size=8G\nmp0: /mnt/media,mp=/media,backup=0\n'
text=self.preview(lambda v:config)
self.assertIn('8G',text);self.assertIn('/mnt/media',text)
self.assertIn('not targeted for removal',text)
def test_missing_guest_can_still_be_reviewed_without_a_second_probe(self):
text=self.preview(lambda v:(_ for _ in ()).throw(RuntimeError('guest missing')))
self.assertIn('CT 110',text);self.assertIn('rootfs -',text)
if __name__=='__main__':unittest.main()
oci/tests/test_modify_latency.py
+103
View File
@@ -0,0 +1,103 @@
"""Opening Modify avoids backup probes while preserving operation safeguards."""
from pathlib import Path
import sys
import unittest
from unittest.mock import Mock,patch
ROOT=Path(__file__).resolve().parents[1]
sys.path[:0]=[str(ROOT/'src'),str(ROOT/'remote')]
from proxmenux_oci import management as menu,recreation
import oci_carried_record as carried
import oci_instances as instances
import oci_instance_reconcile as reconcile
from proxmenux_oci.ui import UserCancelled
def record():
return {'vmid': 115, 'installation_id': 'test-installation', 'status': 'installed',
'observed': {'config_sha256': 'native-config-fingerprint'},
'template': {'id': 'example-app', 'catalog_ui': {'title': 'Example'},
'container_contract': {'image': {'reference': 'example/app:latest'}},
'proxmox': {'installer_profile': {}}},
'deployment': {'vmid': 115, 'hostname': 'example', 'resources': {'cores': 2, 'memory_mb': 512},
'network': {'bridge': 'vmbr0', 'ipv4': 'dhcp', 'gateway': None},
'rootfs': {'storage': 'local-zfs', 'size_gb': 8}, 'mounts': [
{'container_path': '/data', 'source': 'local-zfs',
'type': 'managed-volume', 'size_gb': 8, 'backup': True}],
'devices': [], 'security': {'unprivileged': True},
'environment': [{'name': 'PASSWORD', 'value': 'private-test-secret',
'sensitive': True}]}}
class SelectedConfigTests(unittest.TestCase):
def test_valid_source_keyed_cache_avoids_another_pct_process(self):
with patch.object(carried,'configuration',create=True,return_value='description: current\n') as reader, \
patch.object(instances,'command') as command:
self.assertEqual(menu._selected_configuration(115),b'description: current\n')
reader.assert_called_once_with(instances.ROOT,115)
command.assert_not_called()
def test_old_or_unavailable_cache_falls_back_to_pct(self):
for reader in (None,Mock(return_value=None),Mock(return_value={}),Mock(side_effect=OSError('cache unavailable'))):
with self.subTest(reader=reader),patch.object(carried,'configuration',reader,create=True), \
patch.object(instances,'command',return_value=b'live config') as command:
self.assertEqual(menu._selected_configuration(115),b'live config')
command.assert_called_once_with('pct','config','115')
def test_identity_check_is_still_performed_on_the_selected_config(self):
r=record();r['installation_id']='11111111-1111-4111-8111-111111111111'
import contextlib
with patch.object(instances,'locked',return_value=contextlib.nullcontext()), \
patch.object(instances,'read',return_value=r), \
patch.object(menu,'_selected_configuration',return_value=b'description: another container\n'):
self.assertEqual(menu.check_selected(ROOT,{'vmid':115})['reason'],'identity-unconfirmed')
class EditorLatencyTests(unittest.TestCase):
def run_editor(self,proposal=None,backup_ready=True,stale=False,adoption_error=False,approved=True,declined=False):
events=[];saved=record()
row={'vmid':115,'reason':'matched','stack':False,'pending':False,'status':'installed'}
def edit(*a,**kw):
events.append('editor')
if declined:return None
if proposal is None:raise UserCancelled()
return proposal
def backup(*a,**kw):events.append('backup');return backup_ready
ui=Mock();ui.unattended=True;ui.review.return_value=approved
with patch.object(menu,'carry_records',return_value={115:'stale'} if stale else {}) as verify, \
patch.object(menu,'check_selected',return_value=row), \
patch.object(menu,'_selected_configuration',return_value=b'current config'), \
patch.object(instances,'read',return_value=saved), \
patch.object(reconcile,'propose',side_effect=ValueError('external changes') if adoption_error else None,return_value=None), \
patch.object(recreation,'edit_recreation',side_effect=edit), \
patch.object(menu,'_work_backup_ready',side_effect=backup), \
patch.object(menu,'_run_lifecycle',side_effect=lambda *a:events.append('lifecycle') or True):
try:result=menu.manage_instance(ROOT,ui,row,action='modify')
except UserCancelled:result=False
verify.assert_called_once_with(ROOT,vmids=[115],verify=True)
return result,events
def test_open_and_cancel_has_no_backup_or_storage_probes(self):
result,events=self.run_editor()
self.assertFalse(result);self.assertEqual(events,['editor'])
def test_declining_the_review_has_no_backup_or_storage_probes(self):
# The review is part of the Modify menu: declining it and leaving the
# menu returns no proposal, so no probe or operation follows.
result,events=self.run_editor(declined=True)
self.assertFalse(result);self.assertEqual(events,['editor'])
def test_confirmed_edit_keeps_backup_check_before_the_lifecycle(self):
result,events=self.run_editor({'operation':'recreate','candidate':record()})
self.assertTrue(result);self.assertEqual(events,['editor','backup','lifecycle'])
def test_missing_backup_space_still_prevents_the_lifecycle(self):
result,events=self.run_editor({'operation':'recreate','candidate':record()},backup_ready=False)
self.assertFalse(result);self.assertEqual(events,['editor','backup'])
def test_restored_or_conflicting_instance_cannot_enter_editor(self):
for options in ({'stale':True},{'adoption_error':True}):
result,events=self.run_editor(**options)
self.assertFalse(result);self.assertEqual(events,[])
if __name__=='__main__':unittest.main()
oci/tests/test_modify_menu.py
+208
View File
@@ -0,0 +1,208 @@
"""Selective OCI edits stay in a draft until reviewed, then use one transaction."""
import copy
import json
from pathlib import Path
import stat
import sys
import unittest
from unittest.mock import Mock, 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 management, recreation
from proxmenux_oci.ui import BackRequested, UserCancelled
def record():
return {'vmid': 115, 'installation_id': 'test-installation', 'status': 'installed',
'observed': {'config_sha256': 'native-config-fingerprint'},
'template': {'id': 'example-app', 'catalog_ui': {'title': 'Example'},
'container_contract': {'image': {'reference': 'example/app:latest'}},
'proxmox': {'installer_profile': {}}},
'deployment': {'vmid': 115, 'hostname': 'example', 'resources': {'cores': 2, 'memory_mb': 512},
'network': {'bridge': 'vmbr0', 'ipv4': 'dhcp', 'gateway': None},
'rootfs': {'storage': 'local-zfs', 'size_gb': 8}, 'mounts': [
{'container_path': '/data', 'source': 'local-zfs',
'type': 'managed-volume', 'size_gb': 8, 'backup': True}],
'devices': [], 'security': {'unprivileged': True},
'environment': [{'name': 'PASSWORD', 'value': 'private-test-secret',
'sensitive': True}]}}
class MenuUI:
def __init__(self, choices, answers=(), reviews=(True,), confirms=()):
self.choices, self.answers = iter(choices), iter(answers)
self.reviews, self.confirms = iter(reviews), iter(confirms)
self.events = []
self.back_enabled = False
def choose(self, text, options, default=None, **kwargs):
self.events.append(('menu', text, dict(options)))
value = next(self.choices)
if value is not None:
assert value in dict(options), value
return value
def ask(self, text, default=None, **kwargs):
self.events.append(('ask', text, default))
value = next(self.answers)
if isinstance(value, Exception):
raise value
return value
def confirm(self, text, default=False):
self.events.append(('confirm', text))
return next(self.confirms)
def review(self, text, *args, **kwargs):
self.events.append(('review', text))
return next(self.reviews)
def message(self, text, *args):
self.events.append(('message', text))
@patch('proxmenux_oci.i18n.language', return_value='en')
class ModifyMenuTests(unittest.TestCase):
def test_ram_only_does_not_ask_about_cpu_network_paths_devices_or_catalog(self, _language):
original = record()
saved = copy.deepcopy(original)
ui = MenuUI(['memory', 'apply'], ['768'])
with patch.object(recreation, 'refresh_template') as refresh:
result = recreation.edit_recreation(original, ui)
self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 2, 'memory_mb': 768})
self.assertEqual([e[1] for e in ui.events if e[0] == 'ask'], ['RAM in MiB'])
self.assertEqual(result['base_config_sha256'], 'native-config-fingerprint')
self.assertEqual(original, saved)
refresh.assert_not_called()
self.assertFalse(ui.back_enabled)
def test_several_sections_and_revisits_keep_the_current_draft(self, _language):
ui = MenuUI(['memory', 'cores', 'memory', 'apply'], ['768', '3', '1024'])
result = recreation.edit_recreation(record(), ui)
self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 1024})
self.assertEqual([e[2] for e in ui.events if e[0] == 'ask'], ['512', '2', '768'])
self.assertEqual(sum(e[0] == 'review' for e in ui.events), 1)
def test_declining_review_returns_to_menu_without_losing_edits(self, _language):
ui = MenuUI(['cores', 'apply', 'memory', 'apply'], ['3', '768'], reviews=[False, True])
result = recreation.edit_recreation(record(), ui)
self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 768})
self.assertEqual(sum(e[0] == 'review' for e in ui.events), 2)
def test_unchanged_apply_does_not_propose_recreating_the_guest(self, _language):
ui = MenuUI(['memory', 'apply'], ['512'])
self.assertIsNone(recreation.edit_recreation(record(), ui))
self.assertFalse(any(e[0] == 'review' for e in ui.events))
self.assertIn(('message', 'Nothing was changed.'), ui.events)
def test_discard_leaves_the_saved_record_untouched(self, _language):
original = record()
saved = copy.deepcopy(original)
ui = MenuUI(['memory', 'discard'], ['768'], confirms=[True])
self.assertIsNone(recreation.edit_recreation(original, ui))
self.assertEqual(original, saved)
self.assertFalse(any(e[0] == 'review' for e in ui.events))
def test_declining_discard_preserves_the_draft(self, _language):
ui = MenuUI(['memory', None, 'apply'], ['768'], confirms=[False])
result = recreation.edit_recreation(record(), ui)
self.assertEqual(result['candidate']['deployment']['resources']['memory_mb'], 768)
def test_cancel_without_changes_asks_nothing_else(self, _language):
ui = MenuUI([None])
self.assertIsNone(recreation.edit_recreation(record(), ui))
self.assertEqual(len(ui.events), 1)
def test_invalid_ram_is_retried_without_restarting_other_sections(self, _language):
ui = MenuUI(['cores', 'memory', 'apply'], ['3', 'invalid', '64', '768'])
result = recreation.edit_recreation(record(), ui)
self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 768})
self.assertEqual(sum(e[0] == 'message' for e in ui.events), 2)
self.assertEqual(sum(e[:2] == ('ask', 'Cores') for e in ui.events), 1)
def test_back_from_first_field_returns_to_menu_and_keeps_other_edits(self, _language):
ui = MenuUI(['cores', 'memory', 'apply'], ['3', BackRequested()])
result = recreation.edit_recreation(record(), ui)
self.assertEqual(result['candidate']['deployment']['resources'], {'cores': 3, 'memory_mb': 512})
self.assertFalse(ui.back_enabled)
def test_cancelling_a_partly_edited_section_rolls_back_only_that_section(self, _language):
real_edit = recreation.edit_setting
def edit(candidate, ui, setting):
if setting == 'environment':
candidate['deployment']['environment'][0]['value'] = 'partial-value'
raise UserCancelled()
real_edit(candidate, ui, setting)
ui = MenuUI(['cores', 'environment', 'apply'], ['3'])
with patch.object(recreation, 'edit_setting', side_effect=edit):
result = recreation.edit_recreation(record(), ui)
self.assertEqual(result['candidate']['deployment']['resources']['cores'], 3)
self.assertEqual(result['candidate']['deployment']['environment'], record()['deployment']['environment'])
def test_menu_and_review_do_not_expose_a_saved_secret(self, _language):
ui = MenuUI(['cores', 'apply'], ['3'])
recreation.edit_recreation(record(), ui)
self.assertNotIn('private-test-secret', json.dumps(ui.events))
def test_extra_paths_counts_only_paths_added_by_hand(self, _language):
candidate = record()
self.assertTrue(dict(recreation.modification_options(candidate))['paths'].rstrip().endswith(' 0'))
candidate['deployment']['mounts'].append({'container_path': '/media-extra', 'type': 'host-bind',
'source': '/mnt/media', 'custom': True})
self.assertTrue(dict(recreation.modification_options(candidate))['paths'].rstrip().endswith(' 1'))
def test_optional_categories_follow_the_application_profile(self, _language):
candidate = record()
options = dict(recreation.modification_options(candidate))
self.assertTrue({'cores', 'memory', 'network', 'paths', 'devices', 'environment', 'catalog', 'apply'} <= options.keys())
self.assertFalse({'shares', 'acceleration', 'detector'} & options.keys())
candidate['template']['proxmox']['installer_profile'] = {
'shared_directories': {'base_path': '/shares'}, 'object_detector': {'devices': []},
'hardware_acceleration': {'profiles': [{'id': 'intel', 'label': 'Intel GPU'}]}}
candidate['deployment']['hardware_profile'] = 'intel'
options = dict(recreation.modification_options(candidate))
self.assertTrue({'shares', 'acceleration', 'detector'} <= options.keys())
self.assertIn('Intel GPU', options['acceleration'])
class ModifyDispatchTests(unittest.TestCase):
def call_manager(self, proposal, lifecycle):
import oci_instances
import oci_instance_reconcile
saved = record()
row = {'vmid': 115, 'reason': 'matched', 'stack': False, 'pending': False, 'status': 'installed'}
with patch.object(management, 'carry_records', return_value={}), \
patch.object(management, 'check_selected', return_value=row), \
patch.object(management, '_work_backup_ready', return_value=True), \
patch.object(oci_instances, 'read', return_value=saved), \
patch.object(oci_instances, 'command', return_value='hostname: example'), \
patch.object(oci_instance_reconcile, 'propose', return_value=None), \
patch.object(recreation, 'edit_recreation', return_value=proposal), \
patch.object(management, '_run_lifecycle', side_effect=lifecycle):
return management.manage_instance(ROOT, Mock(), row, action='modify')
def test_cancel_does_not_start_a_lifecycle_transaction(self):
lifecycle = Mock()
self.assertFalse(self.call_manager(None, lifecycle))
lifecycle.assert_not_called()
def test_apply_sends_one_private_proposal_to_the_existing_backend(self):
proposal = {'operation': 'recreate', 'candidate': record(), 'base_config_sha256': 'fingerprint'}
seen = []
def lifecycle(command, title):
path = Path(command[command.index('--proposal') + 1])
seen.append(json.loads(path.read_text()))
self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600)
self.assertNotIn('private-test-secret', ' '.join(command))
self.assertIn('oci_update_current.py', command[1])
self.assertIn('--acknowledge-external-data', command)
return True
self.assertTrue(self.call_manager(proposal, lifecycle))
self.assertEqual(seen, [proposal])
if __name__ == '__main__':
unittest.main()
oci/tests/test_readonly_volume_preparation.py
+187
View File
@@ -0,0 +1,187 @@
"""Run the installer's volume preparation loop against filesystem fixtures.
Write doubles reject every mutation of the protected directory, including
lost+found removal. Writable paths still run real rm/chmod; chown is recorded
without requiring root. These tests never invoke Proxmox or touch a container.
"""
import json
import os
from pathlib import Path
import subprocess
import tempfile
import unittest
ROOT = Path(__file__).resolve().parents[1]
INSTALLER = ROOT / "remote/install_oci.sh"
def preparation_loop():
source = INSTALLER.read_text()
start = source.index(" if (( failed == 0 )) && [[ $(jq -r '.stack_managed")
end = source.index("\n if (( failed == 0 && tls_count == 1 )); then", start)
return source[start:end]
def preparation(target):
return {
"container_path": target,
"remove_lost_found": True,
"owner_strategy": "mapped-application-user",
"mode": "0750",
}
class ReadOnlyVolumePreparationTests(unittest.TestCase):
def run_case(self, mounts, *, target="/run/secrets", read_only=True,
preparations=None):
preparations = preparations if preparations is not None else [preparation(target)]
with tempfile.TemporaryDirectory() as directory:
fixture = Path(directory)
rootfs = fixture / "rootfs"
for item in preparations:
path = rootfs / item["container_path"].lstrip("/")
path.mkdir(parents=True, exist_ok=True)
(path / "lost+found").mkdir(exist_ok=True)
protected = rootfs / target.lstrip("/")
protected.mkdir(parents=True, exist_ok=True)
sentinel = protected / "sentinel"
sentinel.write_bytes(b"existing secret\n")
sentinel.chmod(0o600)
protected.chmod(0o700)
before = sentinel.stat()
template = {"proxmox": {"installer_profile": {
"volume_preparations": preparations,
}}}
(fixture / "template.json").write_text(json.dumps(template))
(fixture / "deployment.json").write_text(json.dumps({"mounts": mounts}))
script = r'''set -Eeuo pipefail
rootfs="$FIXTURE/rootfs"
TEMPLATE_FILE="$FIXTURE/template.json"
DEPLOYMENT_FILE="$FIXTURE/deployment.json"
HOST_BIND_UID=110001; HOST_BIND_GID=110001
HOST_ROOT_UID=100000; HOST_ROOT_GID=100000
failed=0; volumes_prepared=0
oci_log() { printf 'log %s\n' "$*"; }
write_op() {
local operation=$1
shift
local path=${!#}
printf 'WRITE %s %s\n' "$operation" "$path"
if [[ -n $PROTECTED && ( $path == "$PROTECTED" || $path == "$PROTECTED/"* ) ]]; then
printf 'Read-only file system\n' >&2
return 1
fi
case "$operation" in
rm|chmod) command "$operation" "$@" ;;
chown) return 0 ;;
esac
}
chown() { write_op chown "$@"; }
chmod() { write_op chmod "$@"; }
rm() { write_op rm "$@"; }
''' + preparation_loop() + r'''
printf 'failed=%s prepared=%s\n' "$failed" "$volumes_prepared"
test "$failed" = 0
'''
result = subprocess.run(
["bash", "-c", script], capture_output=True, text=True,
env=dict(os.environ, FIXTURE=str(fixture),
PROTECTED=str(protected) if read_only else ""),
timeout=15,
)
self.assertEqual(sentinel.read_bytes(), b"existing secret\n")
after = sentinel.stat()
self.assertEqual(
(before.st_ino, before.st_uid, before.st_gid, before.st_mode, before.st_mtime_ns),
(after.st_ino, after.st_uid, after.st_gid, after.st_mode, after.st_mtime_ns),
)
if read_only:
self.assertEqual(protected.stat().st_mode & 0o777, 0o700)
self.assertTrue((protected / "lost+found").is_dir())
elif "prepared=1" in result.stdout:
self.assertEqual(protected.stat().st_mode & 0o777, 0o750)
self.assertFalse((protected / "lost+found").exists())
return result
def assert_skipped(self, result):
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
self.assertNotIn("WRITE ", result.stdout)
self.assertIn("Skipping preparation of read-only volume", result.stdout)
self.assertIn("prepared=0", result.stdout)
def assert_prepared(self, result):
self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
for operation in ("rm", "chown", "chmod"):
self.assertIn(f"WRITE {operation} ", result.stdout)
self.assertIn("prepared=1", result.stdout)
def test_read_only_managed_and_bind_volumes_receive_no_writes(self):
for mount_type in ("managed-volume", "host-bind"):
with self.subTest(mount_type=mount_type):
self.assert_skipped(self.run_case([
{"container_path": "/run/secrets", "type": mount_type, "read_only": True},
]))
def test_read_only_parent_protects_nested_preparation(self):
self.assert_skipped(self.run_case([
{"container_path": "/run", "read_only": True},
]))
def test_writable_child_overrides_read_only_parent_regardless_of_order(self):
mounts = [
{"container_path": "/run", "read_only": True},
{"container_path": "/run/secrets", "read_only": False},
]
for ordered in (mounts, mounts[::-1]):
with self.subTest(mounts=ordered):
self.assert_prepared(self.run_case(ordered, read_only=False))
def test_child_without_read_only_flag_overrides_read_only_parent(self):
self.assert_prepared(self.run_case([
{"container_path": "/run", "read_only": True},
{"container_path": "/run/secrets"},
], read_only=False))
def test_read_only_child_overrides_writable_parent_regardless_of_order(self):
mounts = [
{"container_path": "/run/secrets", "read_only": True},
{"container_path": "/run", "read_only": False},
]
for ordered in (mounts, mounts[::-1]):
with self.subTest(mounts=ordered):
self.assert_skipped(self.run_case(ordered))
def test_writable_volume_is_still_prepared(self):
self.assert_prepared(self.run_case([
{"container_path": "/data", "read_only": False},
], target="/data", read_only=False))
def test_missing_read_only_flag_defaults_to_writable(self):
self.assert_prepared(self.run_case([
{"container_path": "/data"},
], target="/data", read_only=False))
def test_unrelated_mount_with_similar_prefix_does_not_skip_preparation(self):
self.assert_prepared(self.run_case([
{"container_path": "/run/secret", "read_only": True},
], read_only=False))
def test_directory_outside_mounts_is_still_prepared(self):
self.assert_prepared(self.run_case([], read_only=False))
def test_skipping_read_only_volume_does_not_skip_following_writable_volume(self):
result = self.run_case([
{"container_path": "/run/secrets", "read_only": True},
{"container_path": "/data", "read_only": False},
], preparations=[preparation("/run/secrets"), preparation("/data")])
self.assert_prepared(result)
self.assertIn("Skipping preparation of read-only volume: /run/secrets", result.stdout)
for line in result.stdout.splitlines():
if line.startswith("WRITE "):
self.assertTrue(line.endswith("/data") or line.endswith("/data/lost+found"), line)
if __name__ == "__main__":
unittest.main()
oci/tests/test_runtime_home_contract.py
+2 -2
View File
@@ -31,14 +31,14 @@ class RuntimeHomeResolutionTests(unittest.TestCase):
)
self.assertEqual(oci_runtime.home_directory(passwd, 1001), "/srv/app")
def test_unknown_or_unsafe_home_is_not_invented(self):
def test_unsafe_record_is_rejected_but_unknown_uid_defaults_to_root(self):
passwd = self.passwd(
"root:x:0:0:root:relative-home:/bin/sh\n"
"app:x:1001:1001:App user::/sbin/nologin\n"
)
self.assertIsNone(oci_runtime.home_directory(passwd, 0))
self.assertIsNone(oci_runtime.home_directory(passwd, 1001))
self.assertIsNone(oci_runtime.home_directory(passwd, 1234))
self.assertEqual(oci_runtime.home_directory(passwd, 1234), "/")
class RuntimeHomeInstallerContractTests(unittest.TestCase):
oci/tests/test_runtime_home_hardening.py
+260
View File
@@ -0,0 +1,260 @@
"""Exercise HOME resolution without reading image symlinks or changing guests."""
from contextlib import redirect_stderr, redirect_stdout
import io
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "remote"))
import oci_runtime
class HomeHardeningTests(unittest.TestCase):
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
(self.root / "etc").mkdir()
self.passwd = self.root / "etc/passwd"
self.passwd.write_text("root:x:0:0:root:/root:/bin/sh\n", encoding="utf-8")
def cli(self, uid="1234"):
output, error = io.StringIO(), io.StringIO()
with patch.object(sys, "argv", ["oci_runtime.py", "--home", str(self.passwd), uid]), \
redirect_stdout(output), redirect_stderr(error):
status = oci_runtime.main()
return status, output.getvalue(), error.getvalue()
def assert_rejected_without_read(self):
with patch.object(os, "fdopen", side_effect=AssertionError("unsafe file read")):
status, output, _ = self.cli()
self.assertEqual(status, 1)
self.assertEqual(output, "")
def test_unknown_uid_uses_root_only_after_reading_a_regular_database(self):
self.assertEqual(self.cli(), (0, "/\n", ""))
def test_empty_regular_database_uses_root(self):
self.passwd.write_text("")
self.assertEqual(self.cli(), (0, "/\n", ""))
def test_final_symlink_is_rejected_without_reading_its_target(self):
target = self.root / "outside"
target.write_text("app:x:1234:1234::/host-secret:/bin/sh\n")
self.passwd.unlink()
self.passwd.symlink_to(target)
self.assert_rejected_without_read()
self.assertEqual(target.read_text(), "app:x:1234:1234::/host-secret:/bin/sh\n")
def test_symlinked_etc_is_rejected_without_reading_its_target(self):
directory = self.root / "etc"
directory.rename(self.root / "outside-etc")
directory.symlink_to(self.root / "outside-etc", target_is_directory=True)
self.assert_rejected_without_read()
self.assertEqual(self.cli("0")[:2], (1, ""))
def test_pinned_etc_does_not_follow_a_later_directory_replacement(self):
original_open = os.open
outside = self.root / "outside"
outside.mkdir()
(outside / "passwd").write_text("root:x:0:0::/outside:/bin/sh\n")
replaced = []
def replace(path, flags, *args, **kwargs):
if str(path) == "passwd":
(self.root / "etc").rename(self.root / "original-etc")
(self.root / "etc").symlink_to(outside, target_is_directory=True)
replaced.append(True)
return original_open(path, flags, *args, **kwargs)
with patch.object(os, "open", side_effect=replace):
self.assertEqual(self.cli("0"), (0, "/root\n", ""))
self.assertEqual(replaced, [True])
def test_directory_is_rejected_without_reading(self):
self.passwd.unlink()
self.passwd.mkdir()
self.assert_rejected_without_read()
def test_fifo_is_rejected_without_reading_or_waiting_for_a_writer(self):
self.passwd.unlink()
os.mkfifo(self.passwd)
done = subprocess.run(
[sys.executable, "-B", str(ROOT / "remote/oci_runtime.py"), "--home", str(self.passwd), "0"],
capture_output=True, text=True, timeout=3,
)
self.assertEqual(done.returncode, 1)
self.assertEqual(done.stdout, "")
self.assert_rejected_without_read()
def test_missing_database_does_not_masquerade_as_unknown_uid(self):
self.passwd.unlink()
self.assert_rejected_without_read()
def test_permission_failure_does_not_masquerade_as_unknown_uid(self):
with patch.object(os, "open", side_effect=PermissionError("denied")):
self.assert_rejected_without_read()
def test_invalid_encoding_does_not_produce_a_fallback(self):
self.passwd.write_bytes(b"\xff\xfe")
status, output, _ = self.cli()
self.assertEqual((status, output), (1, ""))
def test_partial_matching_record_does_not_produce_a_fallback(self):
self.passwd.write_text("app:x:1234:1234\n")
self.assertEqual(self.cli()[:2], (1, ""))
def test_read_error_closes_both_descriptors_without_a_fallback(self):
descriptors = []
original_open = os.open
def opened(*args, **kwargs):
fd = original_open(*args, **kwargs)
descriptors.append(fd)
return fd
with patch.object(os, "open", side_effect=opened), \
patch.object(os, "fdopen", side_effect=OSError("read failed")):
self.assertEqual(self.cli()[:2], (1, ""))
self.assertEqual(len(descriptors), 2)
for fd in descriptors:
with self.assertRaises(OSError):
os.fstat(fd)
def test_matching_unsafe_home_is_not_replaced_with_root(self):
for value in ("", "relative", "/bad\tvalue", "/bad\x00value", "/bad\x7fvalue"):
with self.subTest(value=repr(value)):
self.passwd.write_text(f"app:x:1234:1234::{value}:/bin/sh\n")
self.assertEqual(self.cli()[:2], (1, ""))
def test_invalid_uid_does_not_open_the_database(self):
for uid in ("-1", "nobody"):
with self.subTest(uid=uid), patch.object(os, "open") as opened:
self.assertEqual(self.cli(uid)[:2], (1, ""))
opened.assert_not_called()
def test_symlink_replacement_between_check_and_open_is_not_read(self):
original_open = os.open
outside = self.root / "outside"
outside.write_text("app:x:1234:1234::/outside:/bin/sh\n")
def replace(path, flags, *args, **kwargs):
if str(path) == "passwd":
self.passwd.unlink()
self.passwd.symlink_to(outside)
return original_open(path, flags, *args, **kwargs)
with patch.object(os, "open", side_effect=replace):
self.assert_rejected_without_read()
def test_fifo_replacement_between_check_and_open_is_not_read(self):
original_open = os.open
def replace(path, flags, *args, **kwargs):
if str(path) == "passwd":
self.passwd.unlink()
os.mkfifo(self.passwd)
return original_open(path, flags, *args, **kwargs)
with patch.object(os, "open", side_effect=replace):
self.assert_rejected_without_read()
class HomeInstallerExecutionTests(unittest.TestCase):
"""Run the real shell HOME function and Python CLI with fake mount boundaries."""
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.root = Path(temporary.name)
(self.root / "etc").mkdir()
self.passwd = self.root / "etc/passwd"
self.passwd.write_text("root:x:0:0:root:/root:/bin/sh\napp:x:1001:1001::/srv/app:/bin/sh\n")
self.conf = self.root / "guest.conf"
self.trace = self.root / "trace"
self.source = (ROOT / "remote/install_oci.sh").read_text()
def run_home(self, config="unrelated: keep\n", unmount_status=0, mount_status=0):
self.conf.write_text(config)
self.trace.write_text("")
functions = []
for name in ("ensure_runtime_home", "set_runtime_environment"):
found = re.search(rf"^{name}\(\) \{{\n.*?^\}}\n", self.source, re.M | re.S)
self.assertIsNotNone(found, name)
functions.append(found.group(0))
script = r'''
set -euo pipefail
CONF=$1
FIXTURE_ROOT=$2
TRACE=$3
OCI_RUNTIME_RESOLVER=$4
PYTHON=$5
UNMOUNT_STATUS=$6
MOUNT_STATUS=$7
OCI_LOG="$FIXTURE_ROOT/resolver.log"
VMID=123456789
translate() { printf '%s' "$1"; }
die() { printf '%s\n' "$*" >&2; exit 1; }
mount_ct_rootfs() {
printf 'mount\n' >> "$TRACE"
[ "$MOUNT_STATUS" = 0 ] || die 'mount failed'
rootfs=$FIXTURE_ROOT
passwd_file="$rootfs/etc/passwd"
}
oci_quiet() { printf '%s\n' "$*" >> "$TRACE"; return "$UNMOUNT_STATUS"; }
python3() { printf 'resolve\n' >> "$TRACE"; command "$PYTHON" -B "$@"; }
'''
done = subprocess.run(
["bash", "-c", script + "\n".join(functions) + "\nensure_runtime_home\nprintf 'continued\\n'\n",
"bash", str(self.conf), str(self.root), str(self.trace), str(ROOT / "remote/oci_runtime.py"),
sys.executable, str(unmount_status), str(mount_status)],
capture_output=True, text=True, timeout=5,
)
return done, self.conf.read_text(), self.trace.read_text().splitlines()
def test_explicit_home_including_empty_never_mounts_or_resolves(self):
for home in ("/custom", ""):
config = f"unrelated: keep\nlxc.environment.runtime: HOME={home}\n"
done, after, trace = self.run_home(config)
self.assertEqual((done.returncode, after, trace), (0, config, []))
def test_default_and_effective_uid_write_the_resolved_home_after_unmount(self):
for uid, expected in ((None, "/root"), (1001, "/srv/app"), (9999, "/")):
config = "unrelated: keep\n" + (f"lxc.init.uid: {uid}\n" if uid is not None else "")
done, after, trace = self.run_home(config)
self.assertEqual(done.returncode, 0, done.stderr)
self.assertEqual(after, config + f"lxc.environment.runtime: HOME={expected}\n")
self.assertEqual(trace, ["mount", "resolve", "pct unmount 123456789"])
def test_invalid_uid_leaves_config_unchanged_without_mount(self):
config = "lxc.init.uid: not-a-uid\n"
done, after, trace = self.run_home(config)
self.assertEqual((done.returncode, after, trace), (0, config, []))
def test_rejected_input_unmounts_and_preserves_config(self):
self.passwd.unlink()
self.passwd.symlink_to(self.root / "missing")
config = "unrelated: keep\n"
done, after, trace = self.run_home(config)
self.assertEqual((done.returncode, after), (0, config))
self.assertEqual(trace, ["mount", "resolve", "pct unmount 123456789"])
def test_failed_unmount_stops_before_config_write_for_success_and_failure(self):
for valid in (True, False):
if not valid:
self.passwd.unlink()
config = "unrelated: keep\n"
done, after, trace = self.run_home(config, unmount_status=1)
self.assertNotEqual(done.returncode, 0)
self.assertEqual(after, config)
self.assertNotIn("continued", done.stdout)
self.assertEqual(trace, ["mount", "resolve", "pct unmount 123456789"])
def test_failed_mount_never_reads_or_writes(self):
config = "unrelated: keep\n"
done, after, trace = self.run_home(config, mount_status=1)
self.assertNotEqual(done.returncode, 0)
self.assertEqual(after, config)
self.assertEqual(trace, ["mount"])
if __name__ == "__main__":
unittest.main()
oci/tests/test_shared_directories.py
+9 -5
View File
@@ -43,15 +43,18 @@ class ScriptedUI:
def password(self, text, required=True):
return self._answer(text, "")
def choose(self, text, options, default=None):
def choose(self, text, options, default=None, **_kwargs):
return self._answer(text, default)
def confirm(self, text, default=False):
return self._answer(text, default)
def message(self, text):
def message(self, text, *_args):
pass
def review(self, *_args, **_kwargs):
return True
@patch("proxmenux_oci.i18n.language", return_value="en")
@patch("proxmenux_oci.installer.host.storages", side_effect=storages)
@@ -223,14 +226,15 @@ class SharedDirectoriesTests(unittest.TestCase):
patch("proxmenux_oci.recreation.edit_acceleration"), \
patch("proxmenux_oci.recreation.edit_environment"):
ui = ScriptedUI({"Add a directory to share": [True], NAME: ["photos"],
"Choose a setting. Changes are applied only after the final review.": ["shares", "apply"],
"Where to store /shares/photos": ["host-bind"],
"Host path for /shares/photos": ["/tank/photos"]})
proposal = edit_recreation(record, ui)
mounts = [m["container_path"] for m in proposal["candidate"]["deployment"]["mounts"]]
self.assertEqual(mounts, ["/config", "/shares/media", "/shares/photos"])
untouched = edit_recreation(record, ScriptedUI())
self.assertEqual([m["container_path"] for m in untouched["candidate"]["deployment"]["mounts"]],
["/config", "/shares/media"])
untouched = edit_recreation(record, ScriptedUI({
"Choose a setting. Changes are applied only after the final review.": ["discard"]}))
self.assertIsNone(untouched)
self.assertEqual([m["container_path"] for m in plan["mounts"]], ["/config", "/shares/media"])