mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-02 10:52:22 +00:00
Merge remote-tracking branch 'origin/develop' into HEAD
# Conflicts: # .github/workflows/test-oci-registries.yml
This commit is contained in:
MacRimi
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user