fix(oci): preserve private access to stack frontends across lifecycle operations

oci/remote/install_generic_stack.py

@@ -26,6 +26,7 @@ import xml.etree.ElementTree as ET
 from allocate_private_network import allocate_network, existing_bridges, existing_networks
 import oci_instances
 import oci_nested_mounts
+import oci_stack_public_routes
 from oci_ui import translate, msg_info, msg_ok, msg_warn, msg_error, msg_info2, stop_spinner, log
 
 HERE = Path(__file__).resolve().parent
@@ -306,6 +307,45 @@ def prepare_public_environment(services):
         _set_public_value(service, key, binding['value'].replace('@STACK_LAN_IP@', str(address)))
 
 
+def install_public_routes(services, primary_id, bridge):
+    """Private members can reach only this stack's own static frontend IPs."""
+    frontends = [service for service in services if service.get('main') or service.get('frontend')]
+    targets = {}
+    for service in frontends:
+        address = service.get('frontend_ipv4')
+        if not address or address == 'dhcp':
+            continue
+        destination = str(ipaddress.IPv4Interface(address).ip)
+        if destination in targets:
+            raise RuntimeError('Duplicate stack frontend address')
+        targets[destination] = service
+    if not targets:
+        return
+    records = {service['vmid']: oci_instances.read(oci_instances.ROOT, service['vmid']) for service in services}
+    stack_id = records[primary_id]['installation_id']
+    for service in services:
+        if service.get('main') or service.get('frontend'):
+            continue
+        vmid = service['vmid']
+        identity = records[vmid]['installation_id']
+        spec = {'schema': 1, 'stack_id': stack_id, 'primary_vmid': primary_id,
+                'vmid': vmid, 'installation_id': identity, 'private_bridge': bridge,
+                'private_address': service['ip'] + '/24', 'routes': [
+                    {'destination': destination, 'via': target['ip'], 'target_vmid': target['vmid'],
+                     'target_installation_id': records[target['vmid']]['installation_id']}
+                    for destination, target in sorted(targets.items())]}
+        plan = service['deployment']
+        plan['stack_public_routes'] = spec
+        oci_stack_public_routes.validate(plan, vmid, identity)
+        config = Path(f'/etc/pve/lxc/{vmid}.conf')
+        text = config.read_text()
+        # Never replace another include or accept a stale route contract.
+        oci_stack_public_routes.check(text, {}, vmid, require_host_assets=False)
+        line = oci_stack_public_routes.install(plan, vmid, identity)
+        config.write_text(text.rstrip('\n') + '\n' + line + '\n')
+        oci_stack_public_routes.check(config.read_text(), plan, vmid, identity)
+
+
 def bind_completed_jobs(services):
     """All members and private aliases exist before a job can be authorized."""
     from oci_compose_jobs import job_spec_from_record
@@ -1022,6 +1062,7 @@ def main():
                 result.update(completion_notes=list(deployment.get('completion_notes',[])), log=LOG)
                 print('PROXMENUX_RESULT='+base64.b64encode(json.dumps(result).encode()).decode(),flush=True)
                 return
+            install_public_routes(services, primary_id, bridge)
             bind_completed_jobs(services)
             hook_spec = Path(tmp)/'lifecycle.json'
             write_json(hook_spec,{'schema':1,'stack':deployment['stack_name'],'dependencies':[

oci/remote/install_oci.sh

@@ -2107,6 +2107,16 @@ apply_rlimits
 apply_host_monitor
 apply_kept_settings
 
+# Recreated stack members retain their owned pre-init routes. Append rather
+# than replacing lxc.include: other managed runtime includes must survive.
+if jq -e '.stack_public_routes != null' "$DEPLOYMENT_FILE" >/dev/null; then
+  ROUTE_INCLUDE=$(python3 "${SCRIPT_DIR}/oci_stack_public_routes.py" install "$VMID" \
+    --deployment "$DEPLOYMENT_FILE" --installation-id "$INSTANCE_ID") \
+    || die "Managed public route installation failed"
+  [[ -n "$ROUTE_INCLUDE" ]] || die "Managed public route include is missing"
+  grep -Fxq "$ROUTE_INCLUDE" "$CONF" || printf '%s\n' "$ROUTE_INCLUDE" >>"$CONF"
+fi
+
 while IFS= read -r REPAIR_ENCODED; do
   [[ -n $REPAIR_ENCODED ]] || continue
   run_pre_start_repair "$(printf '%s' "$REPAIR_ENCODED" | base64 -d)"

oci/remote/oci_instance_transaction.py

@@ -33,6 +33,7 @@ import oci_host_mounts as host_mounts
 import oci_nested_mounts as nested_mounts
 import oci_accelerators as gpu_devices
 import oci_runtime_settings as runtime_settings
+import oci_stack_public_routes as public_routes
 import oci_image_cache as image_cache
 import oci_work_backup as work_backup
 import oci_ui
@@ -52,7 +53,7 @@ BASIC = {'arch', 'cmode', 'console', 'tty', 'cores', 'cpulimit', 'cpuunits', 'de
 # Their output is data (and may hold saved secrets); it is never logged.
 DATA_COMMANDS = {('pct', 'config'), ('pvesh', 'get'), ('pct', 'df')}
 # Kept across a modification: the watchdog and where the backups are made.
-HOST_PREFERENCES = ('watchdog', 'work_storage')
+HOST_PREFERENCES = ('watchdog', 'work_storage', 'stack_public_routes')
 LOG_DIR = Path(os.environ.get('OCI_LOG_DIR', '/var/log/proxmenux/oci'))
 ANSI = re.compile(r'\x1b\[[0-9;?]*[A-Za-z]')
 SPINNER_FRAME = re.compile('^ ?[' + ''.join(oci_ui.FRAMES) + ']')
@@ -476,7 +477,7 @@ def candidate_contract(record, operation, proposal=None):
     # of what is being edited: it stays as the record has it now.
     for key in HOST_PREFERENCES:
         if key in record['deployment']:
-            result['deployment'][key] = record['deployment'][key]
+            result['deployment'][key] = copy.deepcopy(record['deployment'][key])
         else:
             result['deployment'].pop(key, None)
     return with_default_healthcheck(result)
@@ -519,6 +520,11 @@ def external_changes(record, config, adopt=True):
     return values
 
 
+def without_public_routes(config, vmid):
+    """The managed include is validated separately from sysctl includes."""
+    return b'\n'.join(line for line in config.splitlines() if not public_routes.own(line, vmid)) + b'\n'
+
+
 def preflight(record, candidate, config, coordinated=None, in_place=False, baseline=True):
     deployment = record['deployment']
     desired = candidate['deployment']
@@ -555,7 +561,10 @@ def preflight(record, candidate, config, coordinated=None, in_place=False, basel
     files = runtime_settings.file_binds(config)
     if files:
         runtime_keys.add('lxc.mount.entry')
-    runtime_settings.check(config, deployment, record['vmid'])
+    public_routes.check(config, deployment, record['vmid'], record['installation_id'])
+    if deployment.get('stack_public_routes'):
+        runtime_keys.add('lxc.include')
+    runtime_settings.check(without_public_routes(config, record['vmid']), deployment, record['vmid'])
     coordinated_keys = {'hookscript', *KEPT_AS_IS} if coordinated else set(KEPT_AS_IS)
     preserved_keys = {'nameserver', 'searchdomain', 'timezone', 'protection'} if in_place else set()
     detached = unused_volumes(config)
@@ -680,7 +689,8 @@ def freeze_host_sources(record, candidate, acknowledge_external_data, files=()):
 
 
 def check_runtime_mounts(config, deployment):
-    runtime_settings.check(config, deployment, deployment['vmid'])
+    public_routes.check(config, deployment, deployment['vmid'])
+    runtime_settings.check(without_public_routes(config, deployment['vmid']), deployment, deployment['vmid'])
     nested_mounts.check(config)
     actual = mounts(config)
     files = runtime_settings.file_binds(config)
@@ -1303,6 +1313,7 @@ def complete_recovery(root, journal, state):
         raise ValueError(translate('The configuration changed after the backup was restored; the recovery is not confirmed'))
     cleanup_restored_format_dirs(vmid, state['record']['deployment'], state['record']['installation_id'])
     runtime_settings.restore(state['record']['deployment'], vmid)
+    public_routes.restore(state['record']['deployment'], vmid, state['record']['installation_id'])
     check_runtime_mounts(config, state['record']['deployment'])
     gpu_devices.check(config, state['record']['deployment'])
     if show:

oci/remote/oci_remove.py

@@ -20,6 +20,7 @@ import oci_image_cache as image_cache
 import oci_work_backup as work_backup
 import oci_instances as instances
 import oci_compose_jobs as compose_jobs
+import oci_stack_public_routes as public_routes
 from oci_installation_state import parse_config
 import oci_console
 import oci_runtime_settings as runtime_settings
@@ -198,7 +199,7 @@ def _leftovers(vmid):
     """Whether anything of the container is still on the host."""
     paths = [runtime_settings.include_path(vmid), runtime_settings.legacy_include_path(vmid),
              runtime_settings.seccomp_path(vmid), SNIPPETS / f'proxmenux-rclone-{int(vmid)}-fuse-hook.sh', MONITOR_APPS / f'{int(vmid)}.json',
-             CLUSTER_RECORDS / f'{int(vmid)}.json',
+             CLUSTER_RECORDS / f'{int(vmid)}.json', public_routes.include_path(vmid), public_routes.contract_path(vmid),
              *oci_console.LOG_DIR.glob(f'{int(vmid)}.console.log*')]
     return any(path.exists() for path in paths)
 
@@ -214,7 +215,8 @@ def sweep_orphans(root):
                                (runtime_settings.seccomp_path(0).parent, r'([0-9]+)\.proxmenux-seccomp'),
                                (oci_console.LOG_DIR, r'([0-9]+)\.console\.log.*'),
                                (SNIPPETS, r'proxmenux-rclone-([0-9]+)-fuse-hook\.sh'),
-                               (CLUSTER_RECORDS, r'([0-9]+)\.json')):
+                               (CLUSTER_RECORDS, r'([0-9]+)\.json'),
+                               (public_routes.DIRECTORY, r'([0-9]+)\.(?:conf|json)')):
         if directory.is_dir():
             found.update(int(m.group(1)) for m in (re.fullmatch(pattern, p.name) for p in directory.iterdir()) if m)
     # Only the App tab registrations of OCI installs; the other ones belong to
@@ -231,10 +233,17 @@ def sweep_orphans(root):
         if instances.guest_exists(vmid):
             continue
         record = instances.has_contract(root, vmid)
+        route_assets = public_routes.contract_path(vmid).exists() or public_routes.include_path(vmid).exists()
+        route_identity = instances.read(root, vmid)['installation_id'] if record and route_assets else None
         if record and not instances.release_orphan(root, vmid):
             continue
         if not (record or _leftovers(vmid)):
             continue
+        try:
+            public_routes.remove(vmid, route_identity)
+        except (ValueError, OSError):
+            # Changed or unowned assets remain for inspection, never guessed.
+            continue
         oci_console.remove_log(vmid)
         remove_host_state(vmid)
         cleaned.append(vmid)
@@ -340,6 +349,7 @@ def remove(root, vmid):
             msg_warn(f"{translate('The container no longer exists:')} CT {member}")
             incomplete = True
             oci_console.remove_log(member)
+            public_routes.remove(member, record['installation_id'])
             remove_host_state(member)
         elif instances.identity(config) != record['installation_id']:
             msg_warn(f"{translate('The VMID belongs to another container now and is not touched:')} CT {member}")
@@ -349,6 +359,7 @@ def remove(root, vmid):
             subprocess.run(['pct', 'stop', str(member), '--skiplock', '1'], check=False, capture_output=True)
             run('pct', 'destroy', str(member), '--purge', '1', '--destroy-unreferenced-disks', '1')
             oci_console.remove_log(member)
+            public_routes.remove(member, record['installation_id'])
             remove_host_state(member)
             msg_ok(f"{translate('Container removed:')} CT {member}")
     if bridge and not bridge_in_use(bridge, set(members)):

oci/remote/oci_restore_recovery.py

@@ -31,6 +31,7 @@ import tempfile
 
 import oci_carried_record as carried
 import oci_console
+import oci_stack_public_routes as public_routes
 import oci_instances as instances
 import oci_compose_jobs as compose_jobs
 import oci_stack_plan
@@ -59,7 +60,7 @@ HOST_MONITOR_CONTENT = 'lxc.namespace.share.pid = 1\nlxc.namespace.share.net = 1
 # them from, so they are not offered again.
 UNRECOVERABLE = '.unrecoverable.json'
 # The fields of a record that name a container of the application.
-VMID_KEYS = ('vmid', 'primary_vmid', 'base_vmid')
+VMID_KEYS = ('vmid', 'primary_vmid', 'base_vmid', 'target_vmid')
 
 
 def run(*args):
@@ -287,7 +288,7 @@ def renumbered_lines(text, old, new):
              f'lxc.include: {runtime_settings.legacy_include_path(old)}': f'lxc.include: {runtime_settings.include_path(new)}'}
     hook = re.compile(rf'(hookscript: \S+:snippets/proxmenux-rclone-){int(old)}(-fuse-hook\.sh)')
     lines = [hook.sub(rf'\g<1>{int(new)}\g<2>', swaps.get(line, line)) if line.startswith('hookscript: ')
-             else swaps.get(line, line) for line in text.split('\n')]
+             else swaps.get(line, public_routes.renumber_line(line, old, new)) for line in text.split('\n')]
     return '\n'.join(lines)
 
 
@@ -441,6 +442,11 @@ def check_host_resources(plan):
         vmid, label = entry['vmid'], f"CT {entry['vmid']}"
         values = parse_config(entry['config'].encode())
         deployment = entry['copy']['record']['deployment']
+        projected = renumbered_lines(entry['config'], entry.get('recorded', vmid), vmid)
+        try:
+            public_routes.check(projected, deployment, vmid, entry.get('installation_id'), require_host_assets=False)
+        except ValueError as error:
+            plan['blockers'].append(f"{label}: {error}")
         directories = [value.split(',', 1)[0] for key, value in values.items()
                        if re.fullmatch(r'mp[0-9]+', key) and value.startswith('/')]
         devices = [oci_stack_modify.device_path(value) for key, value in values.items()
@@ -482,6 +488,9 @@ def check_host_resources(plan):
             if key != 'lxc.include':
                 continue
             path = Path(value)
+            if any(public_routes.own(f'lxc.include: {path}', number) for number in {vmid, entry.get('recorded', vmid)}):
+                # Canonical assets are emitted from the validated deployment before start.
+                continue
             if path in HOST_MONITOR:
                 if path.exists() and path.read_text() != HOST_MONITOR_CONTENT:
                     plan['blockers'].append(f"{label}: {translate('a different host monitor include already exists:')} {path}")
@@ -708,12 +717,27 @@ def check_records(root, plan):
             plan['blockers'].append(f"CT {entry['vmid']}: {translate('this host keeps a record of another installation with an operation pending')}")
 
 
+def check_public_route_members(root, plan):
+    """Guest-carried routes can target only members validated in this recovery."""
+    restored = {entry['vmid']: entry['copy']['record'] for entry in plan['restored'] if entry.get('copy')}
+    try:
+        records = {member['vmid']: restored.get(member['vmid']) or instances.read(root, member['vmid'])
+                   for member in plan['members']}
+        if not any(record.get('deployment', {}).get('stack_public_routes') for record in records.values()):
+            return
+        oci_stack_plan.validate_public_route_members(plan['primary'], records)
+    except (ValueError, KeyError, OSError) as error:
+        plan['blockers'].append(translate('Invalid stack contract') + ': ' + str(error))
+
+
 def prepare(root):
     """Every restored application of this node, with what prevents its
     recovery and what the user has to know."""
     plans = applications(root, [examine(entry) for entry in pending(root)])
     for plan in plans:
         check_records(root, plan)
+        if not plan['blockers']:
+            check_public_route_members(root, plan)
         if not plan['blockers']:
             check_networks(plan)
             check_host_resources(plan)
@@ -740,6 +764,7 @@ def restore_host_files(entry):
     vmid = entry['vmid']
     deployment = entry['copy']['record']['deployment']
     renumber_config(entry)
+    public_routes.restore(deployment, vmid, entry.get('installation_id'))
     for line in entry['config'].splitlines():
         key, _, value = line.partition(': ')
         if key == 'lxc.include':

oci/remote/oci_stack_native.py

@@ -15,6 +15,7 @@ import uuid
 
 import oci_image_cache as image_cache
 import oci_carried_record as carried
+import oci_stack_public_routes as public_routes
 import oci_compose_jobs as compose_jobs
 import oci_instances as instances
 import oci_instance_transaction as member_tx
@@ -227,6 +228,7 @@ class NativeAdapter:
                 if row['type'] != 'lxc' or row.get('node') != socket.gethostname().split('.')[0]:
                     raise ValueError(translate('A member VMID is in use by another guest or is on another node'))
                 config = instances.command('pct', 'config', str(vmid))
+                public_routes.check(config, record['deployment'], vmid, record['installation_id'])
                 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, ignore_tags=True):

oci/remote/oci_stack_plan.py

@@ -1,5 +1,6 @@
 """Read-only validation of a coordinated stack operation, before host changes."""
 import copy
+import oci_stack_public_routes as public_routes
 import re
 import uuid
 
@@ -32,6 +33,27 @@ def validate_job_check(check, record=None):
     return job
 
 
+def validate_public_route_members(primary_id, records):
+    """A route may name only the frontends of this exact recorded stack."""
+    for vmid, record in records.items():
+        spec = public_routes.validate(record.get('deployment', {}), vmid, record['installation_id'])
+        if spec is None:
+            continue
+        primary = records.get(primary_id)
+        if (primary is None or spec['primary_vmid'] != primary_id
+                or spec['stack_id'] != primary['installation_id']):
+            raise ValueError(translate('Invalid stack contract'))
+        members = {member['vmid']: member['installation_id']
+                   for member in primary.get('stack', {}).get('members', [])}
+        if members.get(vmid) != record['installation_id']:
+            raise ValueError(translate('Invalid stack contract'))
+        for route in spec['routes']:
+            target = records.get(route['target_vmid'])
+            if (target is None or target['installation_id'] != route['target_installation_id']
+                    or members.get(route['target_vmid']) != route['target_installation_id']):
+                raise ValueError(translate('Invalid stack contract'))
+
+
 def build(primary, records, inventory, operation):
     """Inventory maps VMIDs to installation UUIDs, including unrelated guests.
 
@@ -43,6 +65,7 @@ def build(primary, records, inventory, operation):
     stack = primary.get('stack')
     if not stack or not stack.get('members'):
         raise ValueError(translate('This is not a coordinated stack'))
+    validate_public_route_members(primary['vmid'], records)
     snapshots = stack['members']
     ids = [member['vmid'] for member in snapshots]
     if any(type(vmid) is not int or vmid < 100 for vmid in ids) or len(set(ids)) != len(ids):

oci/remote/oci_stack_public_routes.py

@@ -0,0 +1,351 @@
+#!/usr/bin/env python3
+"""Owned /32 routes to a stack's own frontends, installed before OCI init."""
+import argparse
+import copy
+import fcntl
+import ipaddress
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+import tempfile
+import uuid
+
+DIRECTORY = Path('/etc/pve/priv/proxmenux-stack-public-routes')
+HELPER = Path('/usr/local/share/proxmenux/oci/engine/remote/oci_stack_public_routes.py')
+CONFIGS = Path('/etc/pve/lxc')
+ROOT_UID = 0
+PRIVATE = ipaddress.ip_network('10.77.0.0/16')
+IP = '/usr/sbin/ip'
+NSENTER = '/usr/bin/nsenter'
+NS_GET_NSTYPE = 0xb703
+CLONE_NEWNET = 0x40000000
+
+
+class RouteError(ValueError):
+    pass
+
+
+def _vmid(value):
+    if type(value) is not int or not 100 <= value < 1000000000:
+        raise RouteError('Invalid public route VMID')
+    return value
+
+
+def _identity(value):
+    try:
+        result = str(uuid.UUID(value))
+    except (ValueError, TypeError, AttributeError) as error:
+        raise RouteError('Invalid public route identity') from error
+    if result != value:
+        raise RouteError('Invalid public route identity')
+    return result
+
+
+def include_path(vmid):
+    return DIRECTORY / f'{_vmid(vmid)}.conf'
+
+
+def contract_path(vmid):
+    return DIRECTORY / f'{_vmid(vmid)}.json'
+
+
+def _address(value):
+    if not isinstance(value, str):
+        raise RouteError('Invalid public route address')
+    try:
+        address = ipaddress.IPv4Address(value)
+    except (ValueError, TypeError) as error:
+        raise RouteError('Invalid public route address') from error
+    if address.is_loopback or address.is_multicast or address.is_unspecified or address.is_link_local:
+        raise RouteError('Invalid public route address')
+    return address
+
+
+def validate(deployment, vmid, identity=None):
+    spec = deployment.get('stack_public_routes')
+    if spec is None:
+        return None
+    required = {'schema', 'stack_id', 'primary_vmid', 'vmid', 'installation_id',
+                'private_bridge', 'private_address', 'routes'}
+    if not isinstance(spec, dict) or set(spec) != required or type(spec['schema']) is not int or spec['schema'] != 1:
+        raise RouteError('Invalid public route contract')
+    if _vmid(spec['vmid']) != _vmid(vmid):
+        raise RouteError('Public route owner differs')
+    _vmid(spec['primary_vmid'])
+    _identity(spec['stack_id'])
+    if _identity(spec['installation_id']) != (identity or spec['installation_id']):
+        raise RouteError('Public route owner identity differs')
+    if not isinstance(spec['private_bridge'], str) or not re.fullmatch(r'vmbr[0-9]+', spec['private_bridge']):
+        raise RouteError('Invalid public route bridge')
+    if not isinstance(spec['private_address'], str):
+        raise RouteError('Invalid public route private address')
+    try:
+        source = ipaddress.IPv4Interface(spec['private_address'])
+    except (ValueError, TypeError) as error:
+        raise RouteError('Invalid public route private address') from error
+    if source.network.prefixlen != 24 or source.ip not in PRIVATE or source.ip in (source.network.network_address, source.network.broadcast_address):
+        raise RouteError('Invalid public route private address')
+    routes = spec['routes']
+    if not isinstance(routes, list) or not routes or len(routes) > 64:
+        raise RouteError('Invalid public route list')
+    destinations = set()
+    for route in routes:
+        if not isinstance(route, dict) or set(route) != {'destination', 'via', 'target_vmid', 'target_installation_id'}:
+            raise RouteError('Invalid public route entry')
+        destination, via = _address(route['destination']), _address(route['via'])
+        if (destination in source.network or destination in destinations or via not in source.network
+                or via in (source.ip, source.network.network_address, source.network.broadcast_address)
+                or _vmid(route['target_vmid']) == vmid):
+            raise RouteError('Invalid public route destination or next hop')
+        _identity(route['target_installation_id'])
+        destinations.add(destination)
+    return copy.deepcopy(spec)
+
+
+def include_content():
+    return f'lxc.hook.start-host = /usr/bin/python3 -I {HELPER} start\n'
+
+
+def own(line, vmid):
+    if isinstance(line, bytes):
+        line = line.decode()
+    return line.rstrip('\n') == f'lxc.include: {include_path(vmid)}'
+
+
+def renumber_line(line, old, new):
+    binary = isinstance(line, bytes)
+    text = line.decode() if binary else line
+    if own(text, old):
+        text = f'lxc.include: {include_path(new)}' + ('\n' if text.endswith('\n') else '')
+    return text.encode() if binary else text
+
+
+def _read(path):
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or info.st_uid != ROOT_UID or info.st_mode & 0o022:
+        raise RouteError('Unsafe managed public route file')
+    if info.st_size > 262144:
+        raise RouteError('Managed public route file is too large')
+    return path.read_bytes()
+
+
+def _managed_directory(path, create=False):
+    if create:
+        path.mkdir(parents=True, exist_ok=True)
+    info = path.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != ROOT_UID or info.st_mode & 0o022:
+        raise RouteError('Unsafe managed public route directory')
+
+
+def _write(path, data, mode=0o600):
+    _managed_directory(path.parent, create=True)
+    if path.is_symlink():
+        raise RouteError('Managed public route path is a symbolic link')
+    fd, temporary = tempfile.mkstemp(prefix='.public-routes-', dir=path.parent)
+    try:
+        with os.fdopen(fd, 'wb') as output:
+            output.write(data)
+            output.flush()
+            os.fsync(output.fileno())
+        # pmxcfs supplies the permissions of files below its private directory.
+        if Path('/etc/pve') not in path.parents:
+            os.chmod(temporary, mode)
+        os.replace(temporary, path)
+    finally:
+        if os.path.exists(temporary):
+            os.unlink(temporary)
+
+
+def install(deployment, vmid, identity=None):
+    spec = validate(deployment, vmid, identity)
+    if spec is None:
+        return
+    # The installed engine is the authority; upgrades replace it normally.
+    _managed_directory(HELPER.parent)
+    _read(HELPER)
+    _write(contract_path(vmid), (json.dumps(spec, sort_keys=True) + '\n').encode())
+    _write(include_path(vmid), include_content().encode())
+    return f'lxc.include: {include_path(vmid)}'
+
+
+restore = install
+publish = install
+
+
+def check(config, deployment, vmid, identity=None, require_host_assets=True):
+    spec = validate(deployment, vmid, identity)
+    text = config.decode() if isinstance(config, bytes) else config
+    includes = [line for line in text.splitlines() if line.startswith('lxc.include: ')
+                and str(DIRECTORY) + '/' in line]
+    expected = [f'lxc.include: {include_path(vmid)}'] if spec else []
+    if includes != expected:
+        raise RouteError('Managed public route include differs from the contract')
+    if spec and require_host_assets:
+        _managed_directory(DIRECTORY)
+        _managed_directory(HELPER.parent)
+        if _read(include_path(vmid)) != include_content().encode():
+            raise RouteError('Managed public route include was changed')
+        if json.loads(_read(contract_path(vmid))) != spec:
+            raise RouteError('Managed public route contract was changed')
+        _read(HELPER)
+    return spec
+
+
+def remove(vmid, identity=None):
+    path = contract_path(vmid)
+    if not path.exists() and not path.is_symlink():
+        if include_path(vmid).exists() or include_path(vmid).is_symlink():
+            raise RouteError('Public route include has no ownership contract')
+        return False
+    _managed_directory(DIRECTORY)
+    spec = json.loads(_read(path))
+    validate({'stack_public_routes': spec}, vmid, identity)
+    include = include_path(vmid)
+    if include.exists() or include.is_symlink():
+        if _read(include) != include_content().encode():
+            raise RouteError('Managed public route include was changed')
+        include.unlink()
+    path.unlink()
+    return True
+
+
+def _config(vmid):
+    text = _read(CONFIGS / f'{_vmid(vmid)}.conf').decode()
+    matches = re.findall(r'proxmenux-instance=([0-9a-f-]{36})(?![0-9a-f-])', text)
+    if len(set(matches)) != 1:
+        raise RouteError('Public route container identity is unknown')
+    values = {}
+    for line in text.splitlines():
+        if line.startswith('#') or ': ' not in line:
+            continue
+        key, value = line.split(': ', 1)
+        if key in ('net0', 'net1'):
+            if key in values:
+                raise RouteError('Ambiguous public route network')
+            parts = [part.split('=', 1) for part in value.split(',') if '=' in part]
+            if len({key for key, _ in parts}) != len(parts):
+                raise RouteError('Ambiguous public route network')
+            values[key] = dict(parts)
+    return matches[0], values, text
+
+
+def live_contract(spec):
+    """Validate actual PVE ownership and both ends of every generated route."""
+    vmid = spec['vmid']
+    identity, networks, text = _config(vmid)
+    check(text, {'stack_public_routes': spec}, vmid, identity)
+    source = networks.get('net0', {})
+    if source.get('bridge') != spec['private_bridge'] or source.get('ip') != spec['private_address'] or source.get('name') != 'eth0':
+        raise RouteError('Public route private interface differs')
+    primary_identity, primary_networks, _ = _config(spec['primary_vmid'])
+    if primary_identity != spec['stack_id'] or primary_networks.get('net0', {}).get('bridge') != spec['private_bridge']:
+        raise RouteError('Public route stack ownership differs')
+    for route in spec['routes']:
+        target_identity, target_networks, _ = _config(route['target_vmid'])
+        private, frontend = target_networks.get('net0', {}), target_networks.get('net1', {})
+        try:
+            private_ip = ipaddress.IPv4Interface(private.get('ip'))
+            public_ip = ipaddress.IPv4Interface(frontend.get('ip'))
+        except (ValueError, TypeError) as error:
+            raise RouteError('Public route target is not a static frontend') from error
+        if (target_identity != route['target_installation_id'] or private.get('bridge') != spec['private_bridge']
+                or private_ip.network != ipaddress.IPv4Interface(spec['private_address']).network
+                or str(private_ip.ip) != route['via'] or str(public_ip.ip) != route['destination']):
+            raise RouteError('Public route target ownership or addresses differ')
+    mac = source.get('hwaddr', '').lower()
+    if not re.fullmatch(r'(?:[0-9a-f]{2}:){5}[0-9a-f]{2}', mac):
+        raise RouteError('Public route private hardware address is unknown')
+    return mac
+
+
+def namespace_fd(path):
+    if not isinstance(path, str) or not re.fullmatch(r'/proc/[1-9][0-9]*/(?:fd/[0-9]+|ns/net)', path):
+        raise RouteError('LXC did not provide a network namespace')
+    fd = os.open(path, os.O_RDONLY)
+    try:
+        if fcntl.ioctl(fd, NS_GET_NSTYPE) != CLONE_NEWNET:
+            raise RouteError('LXC namespace is not a network namespace')
+        info, host = os.fstat(fd), os.stat('/proc/self/ns/net')
+        if (info.st_dev, info.st_ino) == (host.st_dev, host.st_ino):
+            raise RouteError('Refusing to configure the host network namespace')
+        return fd
+    except BaseException:
+        os.close(fd)
+        raise
+
+
+def _namespace_command(fd, *args):
+    process = subprocess.run([NSENTER, f'--net=/proc/self/fd/{fd}', '--', IP, *args],
+                             pass_fds=(fd,), capture_output=True, check=True, timeout=15)
+    return process.stdout
+
+
+def apply(spec, fd, mac):
+    source = ipaddress.IPv4Interface(spec['private_address'])
+    interfaces = json.loads(_namespace_command(fd, '-j', 'address', 'show', 'dev', 'eth0'))
+    if (len(interfaces) != 1 or interfaces[0].get('address', '').lower() != mac
+            or not any(item.get('family') == 'inet' and item.get('local') == str(source.ip)
+                       and item.get('prefixlen') == source.network.prefixlen
+                       for item in interfaces[0].get('addr_info', []))):
+        raise RouteError('Network namespace does not belong to the public route owner')
+    for route in spec['routes']:
+        _namespace_command(fd, '-4', 'route', 'replace', route['destination'] + '/32',
+                           'via', route['via'], 'dev', 'eth0', 'src', str(source.ip))
+
+
+def start():
+    if os.geteuid() != 0:
+        raise RouteError('Public routes require root')
+    name = os.environ.get('LXC_NAME', '')
+    if not re.fullmatch(r'[1-9][0-9]*', name):
+        raise RouteError('LXC did not provide its container identity')
+    vmid = _vmid(int(name))
+    spec = json.loads(_read(contract_path(vmid)))
+    validate({'stack_public_routes': spec}, vmid)
+    mac = live_contract(spec)
+    pid = os.environ.get('LXC_PID', '')
+    if not re.fullmatch(r'[1-9][0-9]*', pid):
+        raise RouteError('LXC did not provide its init process identity')
+    fd = namespace_fd(f'/proc/{pid}/ns/net')
+    try:
+        apply(spec, fd, mac)
+    finally:
+        os.close(fd)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    commands = parser.add_subparsers(dest='operation', required=True)
+    start_parser = commands.add_parser('start')
+    start_parser.add_argument('hook_arguments', nargs='*')
+    install_parser = commands.add_parser('install')
+    install_parser.add_argument('vmid', type=int)
+    install_parser.add_argument('--deployment', required=True)
+    install_parser.add_argument('--installation-id', required=True)
+    args = parser.parse_args()
+    if os.geteuid() != 0:
+        raise RouteError('Public routes require root')
+    if args.operation == 'start':
+        # Proxmox's network scripts use the default LXC version-0 hook ABI.
+        expected = [os.environ.get('LXC_NAME', ''), 'lxc', 'start-host']
+        if args.hook_arguments != expected:
+            raise RouteError('Unexpected public route hook invocation')
+        start()
+    else:
+        line = install(json.loads(Path(args.deployment).read_text()), args.vmid, args.installation_id)
+        if line:
+            print(line)
+
+
+if __name__ == '__main__':
+    try:
+        main()
+    except (ValueError, OSError, subprocess.SubprocessError) as error:
+        detail = str(error)
+        if isinstance(error, subprocess.CalledProcessError) and error.stderr:
+            output = error.stderr.decode(errors='replace') if isinstance(error.stderr, bytes) else error.stderr
+            detail += ': ' + output.strip()[:1000]
+        raise SystemExit('Managed public routes failed: ' + detail)