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

# Conflicts:
#	.github/workflows/test-oci-registries.yml

oci/remote/install_oci.sh

@@ -359,13 +359,18 @@ cleanup_canonical_inputs() {
 # A derived check accepts any HTTP answer below 500: the application is up,
 # even when its first page is a redirect or asks for a login.
 healthcheck_probe() {
+  local code status=0
   if [[ ${HC_ANY_STATUS:-false} == true ]]; then
-    local code
-    code=$(curl "${CURL_ARGS[@]}" "$HC_URL" 2>/dev/null) || return 1
-    [[ $code =~ ^[1-4][0-9][0-9]$ ]]
+    code=$(curl "${CURL_ARGS[@]}" "$HC_URL" 2>/dev/null) || status=$?
+    (( status == 0 )) && [[ $code =~ ^[1-4][0-9][0-9]$ ]] && return 0
   else
-    curl "${CURL_ARGS[@]}" "$HC_URL" 2>/dev/null
+    curl "${CURL_ARGS[@]}" "$HC_URL" >/dev/null 2>&1 || status=$?
+    (( status == 0 )) && return 0
   fi
+  # A port taken from the compose file may serve a database or SSH: an answer
+  # in another protocol (HTTP/0.9, weird or empty reply, reset) means it is up.
+  [[ ${HC_ACCEPT_NON_HTTP:-false} == true ]] \
+    && (( status == 1 || status == 8 || status == 52 || status == 56 ))
 }
 
 cleanup_failed_install() {
@@ -1025,15 +1030,18 @@ apply_extra_hosts() {
 }
 
 # Settings an update gives back to the rebuilt container before it starts:
-# the LAN leg of a suite or stack member and its start order.
+# the LAN leg of a suite or stack member, its start order and current tags.
 apply_kept_settings() {
   local key value
   while IFS=$'\t' read -r key value; do
     [[ -n $key ]] || continue
-    [[ $key == net1 || $key == startup ]] \
+    [[ $key == net1 || $key == startup || $key == tags ]] \
       || die "$(translate "Unsupported kept setting:") $key"
-    oci_quiet pct set "$VMID" "--$key" "$value" \
-      || die "$(translate "Could not restore the container setting:") $key"
+    if [[ $key == tags && -z $value ]]; then
+      oci_quiet pct set "$VMID" --delete tags
+    else
+      oci_quiet pct set "$VMID" "--$key" "$value"
+    fi || die "$(translate "Could not restore the container setting:") $key"
   done < <(jq -r '.kept_proxmox_settings // {} | to_entries[] | [.key, .value] | @tsv' "$DEPLOYMENT_FILE")
 }
 
@@ -1091,7 +1099,7 @@ ensure_runtime_home() {
   passwd_file="${rootfs}/etc/passwd"
   mount_ct_rootfs
   if ! home=$(python3 "$OCI_RUNTIME_RESOLVER" --home "$passwd_file" "$uid" 2>>"${OCI_LOG:-/dev/stderr}"); then
-    oci_quiet pct unmount "$VMID" || true
+    oci_quiet pct unmount "$VMID" || die "$(translate "Could not unmount the container filesystem")"
     return 0
   fi
   oci_quiet pct unmount "$VMID" || die "$(translate "Could not unmount the container filesystem")"
@@ -1230,6 +1238,17 @@ apply_installer_profile() {
       fi
       [[ -d ${rootfs}${target} ]] \
         || { oci_log "The volume to prepare does not exist: $target"; failed=1; break; }
+      # A reused volume may intentionally be mounted read-only (for example,
+      # /run/secrets). Do not remove files or change ownership/mode on it.
+      # Use the most specific containing mount: an RW child can override RO.
+      if [[ $(jq -r --arg target "$target" '
+        [.mounts[]? | select(.container_path as $path |
+          $target == $path or ($target | startswith($path + "/")))]
+        | sort_by(.container_path | length) | last | .read_only // false' \
+        "$DEPLOYMENT_FILE") == true ]]; then
+        oci_log "Skipping preparation of read-only volume: $target"
+        continue
+      fi
       if [[ $remove_lost_found == true ]]; then
         rm -rf "${rootfs}${target}/lost+found" || { failed=1; break; }
       fi
@@ -2253,6 +2272,7 @@ if [[ $START_AFTER == 1 && $HAS_STARTUP_HEALTHCHECK == 1 ]]; then
   [[ -n $IP ]] || die "$(translate "The healthcheck cannot run without an IP address")"
   HC_URL="${HC_SCHEME}://${IP}:${HC_PORT}${HC_PATH}"
   HC_ANY_STATUS=$(jq -r '.accept_any_status // false' <<<"$STARTUP_HEALTHCHECK")
+  HC_ACCEPT_NON_HTTP=$(jq -r '.accept_non_http // false' <<<"$STARTUP_HEALTHCHECK")
   if [[ $HC_ANY_STATUS == true ]]; then
     CURL_ARGS=(-sS --noproxy '*' -o /dev/null -w '%{http_code}' --max-time "$HC_REQUEST_TIMEOUT")
   else

oci/remote/oci_carried_record.py

@@ -38,6 +38,8 @@ DIRECTORY = '.proxmenux'
 NAME = 'oci-record.json'
 KIND = 'proxmenux.oci-carried-record'
 STAMP = 'carried.sha256'
+CONFIG_CACHE = 'carried-config.json'
+LOCAL_CONFIGS = Path('/etc/pve/lxc')
 LIMIT = 16 * 1024 * 1024
 STACK_CONTRACT = '/etc/pve/priv/proxmenux-stack-{}.json'
 NVIDIA_HOOK = re.compile(r'/usr/local/lib/proxmenux/oci/nvidia-mount-([a-f0-9]{64})\.sh')
@@ -279,6 +281,47 @@ def read_stamp(path):
     return value if isinstance(value, dict) else {}
 
 
+def configuration(root, vmid, verify=False):
+    """Reuse pct's formatted output only while its local source is unchanged.
+
+    Opening Manage checks every carried copy. Starting a fresh pct process
+    for each unchanged guest dominates that check. The private host cache is
+    keyed by the complete native configuration, including notes, pending
+    changes and snapshot sections. Lifecycle verification always calls pct.
+    An unavailable source, invalid cache or concurrent edit falls back to pct.
+    """
+    source = LOCAL_CONFIGS / f'{int(vmid)}.conf'
+    cache = instances.location(root, vmid).parent / CONFIG_CACHE
+
+    def fingerprint():
+        try:
+            return sha(source.read_bytes())
+        except OSError:
+            return None
+
+    before = fingerprint()
+    if before is not None and not verify:
+        try:
+            if not cache.is_symlink() and cache.stat().st_size <= LIMIT:
+                value = json.loads(cache.read_text())
+                if (isinstance(value, dict) and value.get('schema_version') == 1
+                        and value.get('source_sha256') == before
+                        and isinstance(value.get('config'), str) and value['config']
+                        and fingerprint() == before):
+                    return value['config']
+        except (OSError, ValueError):
+            pass
+    result = _run('pct', 'config', str(vmid))
+    if result.returncode != 0:
+        return None
+    # Do not associate output from one revision with another revision's hash.
+    if before is not None and fingerprint() == before:
+        with contextlib.suppress(OSError, ValueError):
+            instances.write(cache, {'schema_version': 1, 'source_sha256': before,
+                                    'config': result.stdout})
+    return result.stdout
+
+
 def superseded(record, existing, stamp):
     """Whether the container carries a copy this host did not write, with a
     record that differs from the one of this host. That is a container that
@@ -308,10 +351,10 @@ def carry(root, vmid, mount_stopped=True, verify=False, adopted=False):
     if record.get('status') != 'installed' or record.get('pending_transaction') \
             or record.get('pending_stack_transaction'):
         return 'skipped'
-    result = _run('pct', 'config', str(vmid))
-    if result.returncode != 0 or instances.identity(result.stdout.encode()) != record['installation_id']:
+    config = configuration(root, vmid, verify)
+    if config is None or instances.identity(config.encode()) != record['installation_id']:
         return 'skipped'
-    value = bundle(record, result.stdout)
+    value = bundle(record, config)
     expected = digest(value)
     record_path = instances.location(root, vmid)
     path = record_path.parent / STAMP
@@ -340,7 +383,7 @@ def carry(root, vmid, mount_stopped=True, verify=False, adopted=False):
             else:
                 value['saved_at'] = datetime.datetime.now(datetime.timezone.utc).isoformat()
                 value['generation'] = uuid.uuid4().hex
-                write_copy(rootfs, value, mapped_root(result.stdout))
+                write_copy(rootfs, value, mapped_root(config))
                 stamp = {'generation': value['generation']}
                 outcome = 'carried'
         if outcome == 'stale':

oci/remote/oci_instance_transaction.py

@@ -352,6 +352,22 @@ def profile_keys(deployment):
     return keys
 
 
+GUESSED_ENDPOINT = 'compose-first-tcp-port-fallback'
+# curl exit codes for a port that accepted the connection and answered in a
+# protocol other than HTTP: HTTP/0.9 refused, weird reply, empty reply, reset.
+NON_HTTP_ANSWER = (1, 8, 52, 56)
+
+
+def probe_http(args):
+    """Run one curl probe and return its output and exit code."""
+    log('$ ' + shlex.join(str(arg) for arg in args))
+    try:
+        result = subprocess.run(args, capture_output=True, timeout=60, close_fds=True)
+    except subprocess.TimeoutExpired:
+        return b'', -1
+    return result.stdout, result.returncode
+
+
 def effective_healthcheck(template):
     """The template's own health check or, when it has none, one derived from
     its web address: any HTTP answer below 500 means the application is up."""
@@ -367,9 +383,14 @@ def effective_healthcheck(template):
         path = str(endpoint.get('path') or '/')
         if (endpoint.get('scheme') in ('http', 'https') and str(endpoint.get('port', '')).isdigit()
                 and path.startswith('/') and not any(c.isspace() for c in path)):
-            return {'scheme': endpoint['scheme'], 'port': int(endpoint['port']), 'path': path,
-                    'timeout_seconds': 300, 'request_timeout_seconds': 10, 'stability_seconds': 4,
-                    'verify_tls': False, 'accept_any_status': True}
+            check = {'scheme': endpoint['scheme'], 'port': int(endpoint['port']), 'path': path,
+                     'timeout_seconds': 300, 'request_timeout_seconds': 10, 'stability_seconds': 4,
+                     'verify_tls': False, 'accept_any_status': True}
+            # A port taken from the compose file because the recipe names no
+            # web address may serve a database or SSH: any answer counts.
+            if endpoint.get('source') == GUESSED_ENDPOINT:
+                check['accept_non_http'] = True
+            return check
     # Without a web address, the new image must at least keep the container running.
     return {'type': 'running', 'timeout_seconds': 90, 'stability_seconds': 20}
 
@@ -415,10 +436,9 @@ def candidate_contract(record, operation, proposal=None):
     return with_default_healthcheck(result)
 
 
-# Resource settings edited in Proxmox that the new container keeps.
 # Proxmox settings the rebuilt container gets back exactly as they are: the
-# LAN leg ProxMenux adds to a suite or stack member, and the start order.
-KEPT_AS_IS = ('net1', 'startup')
+# LAN leg ProxMenux adds to a suite or stack member, the start order and tags.
+KEPT_AS_IS = ('net1', 'startup', 'tags')
 ADOPTABLE = {'memory': ('resources', 'memory_mb'), 'swap': ('resources', 'swap_mb'),
              'cores': ('resources', 'cores'), 'cpulimit': ('resources', 'cores'),
              'cpuunits': ('resources', 'cpu_units'), 'onboot': (None, 'onboot')}
@@ -734,11 +754,10 @@ def healthcheck(vmid, template):
                 args += ['-w', '%{http_code}']
             if not check.get('verify_tls', False):
                 args.append('-k')
-            try:
-                answer = run(*args, f'{scheme}://{ip}:{port}{path}')
-                ok = not any_status or re.fullmatch(rb'[1-4][0-9][0-9]', answer.strip()) is not None
-            except RuntimeError:
-                pass
+            answer, status = probe_http([*args, f'{scheme}://{ip}:{port}{path}'])
+            ok = status == 0 and (not any_status or re.fullmatch(rb'[1-4][0-9][0-9]', answer.strip()) is not None)
+            if not ok and check.get('accept_non_http') and status in NON_HTTP_ANSWER:
+                ok = True
         if ok:
             stable_since = stable_since or time.monotonic()
             if time.monotonic() - stable_since >= stability:
@@ -1051,6 +1070,9 @@ def apply(root, vmid, archive, operation, proposal=None, registry_digest=None, i
              'original_gpu_devices': original_gpu, 'desired_gpu_devices': desired_gpu,
              'was_running': run('pct', 'status', str(vmid)).strip() == b'status: running'}
     state['kept_config'] = {key: cfg[key] for key in KEPT_AS_IS if key in cfg}
+    # The installer adds default tags. Preserve an explicit absence too, so
+    # deleting all tags in Proxmox does not bring them back on the next update.
+    state['kept_config']['tags'] = cfg.get('tags', '')
     if not coordinated and state['kept_config']:
         # Applied by the installer before the new container starts, so the
         # application has its LAN leg from the first second.
@@ -1058,9 +1080,10 @@ def apply(root, vmid, archive, operation, proposal=None, registry_digest=None, i
     if coordinated:
         state['coordinated'] = coordinated
         state['preserved_stack_config'] = {key: cfg[key] for key in ('net1', 'startup', 'hookscript') if key in cfg}
+        state['preserved_stack_config']['tags'] = cfg.get('tags', '')
         if operation == 'update':
             state['preserved_stack_config'].update({key: cfg[key] for key in
-                ('net0', 'features', 'cmode', 'console', 'tty', 'cpuunits', 'tags') if key in cfg})
+                ('net0', 'features', 'cmode', 'console', 'tty', 'cpuunits') if key in cfg})
     checkpoint(journal, state, 'prepared')
     pending = copy.deepcopy(record)
     pending.update(status='updating', pending_transaction=str(journal), transaction_id=state['id'])
@@ -1141,7 +1164,10 @@ def apply(root, vmid, archive, operation, proposal=None, registry_digest=None, i
     restore_firewall(vmid, state)
     if coordinated:
         for key, value in state['preserved_stack_config'].items():
-            run('pct', 'set', str(vmid), '--' + key, value)
+            if key == 'tags' and not value:
+                run('pct', 'set', str(vmid), '--delete', 'tags')
+            else:
+                run('pct', 'set', str(vmid), '--' + key, value)
         check_runtime_mounts(run('pct', 'config', str(vmid)), candidate['deployment'])
         gpu_devices.check(run('pct', 'config', str(vmid)), candidate['deployment'])
         state['staged_config_sha256'] = sha(owned(vmid, record['installation_id']))

oci/remote/oci_instances.py

@@ -230,8 +230,9 @@ def identity(config):
     return match.group(1) if match else None
 
 
-def same_config_except_notes(record, config):
-    """Accept a presentation-only note edit, never a changed OCI identity or LXC setting."""
+def same_config_except_notes(record, config, ignore_tags=False):
+    """Accept a presentation-only note edit, never a changed OCI identity or LXC setting.
+    With `ignore_tags`, Proxmox tags may differ too: the operation keeps them as they are."""
     previous = record['observed']['config'].encode()
     expected = record['installation_id']
     if identity(previous) != expected or identity(config) != expected:
@@ -242,7 +243,8 @@ def same_config_except_notes(record, config):
         notes = [line for line in lines if line.startswith(b'description: ')]
         if len(notes) != 1:
             return None
-        return b''.join(line for line in lines if not line.startswith(b'description: '))
+        dropped = (b'description: ', b'tags: ') if ignore_tags else (b'description: ',)
+        return b''.join(line for line in lines if not line.startswith(dropped))
 
     original = without_notes(previous)
     return original is not None and original == without_notes(config)

oci/remote/oci_runtime.py

@@ -4,7 +4,9 @@
 from __future__ import annotations
 
 import json
+import os
 import shlex
+import stat
 import sys
 import tarfile
 from pathlib import Path
@@ -90,14 +92,39 @@ def effective_entrypoint(
 
 
 def home_directory(passwd_path: Path, uid: int) -> str | None:
-    """Return the safe home directory recorded for a container UID."""
-    for line in passwd_path.read_text(encoding="utf-8").splitlines():
+    """Read a regular image passwd file; only an absent UID defaults to /."""
+    if uid < 0:
+        return None
+    # The caller supplies <mounted rootfs>/etc/passwd. Pin etc so an image
+    # symlink or a rename during lookup cannot redirect the host-side read.
+    directory = os.open(passwd_path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
+    try:
+        if not stat.S_ISREG(os.stat(passwd_path.name, dir_fd=directory, follow_symlinks=False).st_mode):
+            return None
+        descriptor = os.open(
+            passwd_path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory,
+        )
+        try:
+            # Recheck the opened object, not its name; a FIFO replacement
+            # must neither block open() nor reach a read operation.
+            if not stat.S_ISREG(os.fstat(descriptor).st_mode):
+                return None
+            with os.fdopen(descriptor, encoding="utf-8", closefd=False) as stream:
+                content = stream.read()
+        finally:
+            os.close(descriptor)
+    finally:
+        os.close(directory)
+    for line in content.splitlines():
         fields = line.split(":")
-        if len(fields) < 7 or fields[2] != str(uid):
+        if len(fields) < 3 or fields[2] != str(uid):
             continue
+        if len(fields) != 7:
+            return None
         home = fields[5]
-        return home if home.startswith("/") else None
-    return None
+        safe = home.startswith("/") and not any(ord(char) < 32 or ord(char) == 127 for char in home)
+        return home if safe else None
+    return "/"
 
 
 def main() -> int:

oci/remote/oci_stack_native.py

@@ -192,7 +192,7 @@ class NativeAdapter:
                 config = instances.command('pct', 'config', str(vmid))
                 if instances.identity(config) != record['installation_id']:
                     raise ValueError(translate('The identity of a member was replaced'))
-                if (not self.journal.exists() or not self.state().get('replacement_intent')) and not instances.same_config_except_notes(record, config):
+                if (not self.journal.exists() or not self.state().get('replacement_intent')) and not instances.same_config_except_notes(record, config, ignore_tags=True):
                     raise ValueError(translate('A member configuration changed during the preparation'))
             else:
                 if Path('/etc/pve/lxc/%s.conf' % vmid).exists():

oci/tests/test_carried_config_cache.py

@@ -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_external_tags.py

@@ -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

@@ -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_readonly_volume_preparation.py

@@ -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

@@ -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

@@ -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()