mirror of
https://github.com/MacRimi/ProxMenux.git
synced 2026-11-01 18:32:13 +00:00
480 lines
21 KiB
Python
480 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify native OCI one-shot completion using authenticated raw LXC events.
|
|
|
|
No image wrapper, shell injection or STOPPED-state inference is used. The
|
|
supported monitor ABI was verified against upstream LXC v7.0.0 monitor.h,
|
|
monitor.c and tools/lxc_monitor.c. The textual CLI loses signal information.
|
|
This standalone helper is copied beside the persistent dependency hook.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import datetime
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import select
|
|
import socket
|
|
import stat
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.parse
|
|
import uuid
|
|
|
|
STATE_ROOT = Path('/var/lib/proxmenux/oci/jobs')
|
|
INSTANCE_ROOT = Path('/usr/local/share/proxmenux/oci/instances')
|
|
PCT = '/usr/sbin/pct'
|
|
MONITOR = '/usr/bin/lxc-monitor'
|
|
LXC_PATH = '/var/lib/lxc'
|
|
FRAME = struct.Struct('@i256si')
|
|
SUPPORTED_VERSIONS = frozenset({'7.0.0'})
|
|
DIGEST = re.compile(r'sha256:[0-9a-f]{64}\Z')
|
|
|
|
|
|
class JobError(RuntimeError):
|
|
pass
|
|
|
|
|
|
def _vmid(value):
|
|
if type(value) is not int or not 100 <= value <= 999999999:
|
|
raise JobError('Invalid OCI job VMID')
|
|
return value
|
|
|
|
|
|
def _root():
|
|
if os.geteuid() != 0:
|
|
raise JobError('OCI job execution requires root')
|
|
|
|
|
|
def _trusted_read(path, *, owner=0):
|
|
"""Open a root-owned regular file without following its final symlink."""
|
|
fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC | os.O_NOFOLLOW)
|
|
try:
|
|
info = os.fstat(fd)
|
|
if not stat.S_ISREG(info.st_mode) or info.st_uid != owner or info.st_mode & 0o022:
|
|
raise JobError('Untrusted OCI job metadata file')
|
|
if info.st_size > 8 * 1024 * 1024:
|
|
raise JobError('OCI job metadata file is too large')
|
|
with os.fdopen(fd, 'r', encoding='utf-8') as source:
|
|
fd = -1
|
|
return source.read()
|
|
finally:
|
|
if fd >= 0:
|
|
os.close(fd)
|
|
|
|
|
|
def _private_directory(path):
|
|
path.mkdir(parents=True, mode=0o700, exist_ok=True)
|
|
info = path.lstat()
|
|
if (not stat.S_ISDIR(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077):
|
|
raise JobError('OCI job state directory is not private')
|
|
|
|
|
|
def config_fingerprint(text):
|
|
"""Presentation and resource changes do not re-run a completed migration."""
|
|
selected = []
|
|
for line in text.split('\n'):
|
|
if line.startswith('['):
|
|
break # Pending/snapshot sections are not the active definition.
|
|
key, separator, value = line.partition(': ')
|
|
if not separator:
|
|
continue
|
|
if (key in {'entrypoint', 'env', 'rootfs', 'unprivileged', 'features', 'ostype'}
|
|
or re.fullmatch(r'(?:mp|dev)[0-9]+', key)
|
|
or key.startswith(('lxc.init.', 'lxc.environment.', 'lxc.mount.',
|
|
'lxc.cgroup', 'lxc.apparmor', 'lxc.seccomp'))):
|
|
if key == 'rootfs' or re.fullmatch(r'mp[0-9]+', key):
|
|
value = ','.join(part for part in value.split(',') if not part.startswith('size='))
|
|
selected.append((key, value))
|
|
if not any(key == 'rootfs' for key, _ in selected):
|
|
raise JobError('OCI job has no root filesystem configuration')
|
|
selected.sort(key=lambda item: item[0])
|
|
return hashlib.sha256(json.dumps(selected, separators=(',', ':'), ensure_ascii=True).encode()).hexdigest()
|
|
|
|
|
|
def _config_text(vmid):
|
|
# Validate the actual PVE file's ownership; use the same API rendering
|
|
# whose exact hash the coordinated transaction authorizes.
|
|
_trusted_read(Path('/etc/pve/lxc') / f'{vmid}.conf')
|
|
result = subprocess.run([PCT, 'config', str(vmid)], capture_output=True, check=True, timeout=15)
|
|
return result.stdout.decode('utf-8')
|
|
|
|
|
|
def _record_digest(record, config_text, *, executing=False):
|
|
pending = record.get('pending_transaction')
|
|
if not pending:
|
|
return record.get('observed', {}).get('resolved_registry_digest')
|
|
vmid = _vmid(record.get('vmid'))
|
|
if record.get('status') != 'updating' or not record.get('pending_stack_transaction'):
|
|
raise JobError('OCI job has no authorized coordinated update context')
|
|
journal = Path(pending)
|
|
try:
|
|
relative = journal.relative_to(INSTANCE_ROOT / str(vmid) / 'transactions')
|
|
except ValueError as error:
|
|
raise JobError('OCI job transaction is outside its private instance state') from error
|
|
if (len(relative.parts) != 2 or relative.parts[1] != 'transaction.json'
|
|
or not re.fullmatch(r'[0-9a-f]{32}', relative.parts[0])):
|
|
raise JobError('Invalid OCI job transaction path')
|
|
child = json.loads(_trusted_read(journal))
|
|
coordinated = child.get('coordinated') or {}
|
|
if (child.get('schema_version') != 1 or child.get('id') != record.get('transaction_id')
|
|
or child.get('id') != relative.parts[0] or child.get('vmid') != vmid
|
|
or child.get('phase') not in {'candidate-installed', 'health-passed', 'committing'}
|
|
or child.get('record', {}).get('installation_id') != record.get('installation_id')
|
|
or coordinated.get('journal') != record.get('pending_stack_transaction')
|
|
or child.get('staged_config_sha256') != hashlib.sha256(config_text.encode()).hexdigest()):
|
|
raise JobError('OCI job staged image is not bound to this actual container configuration')
|
|
stack_journal = Path(record['pending_stack_transaction'])
|
|
try:
|
|
stack_path = stack_journal.relative_to(INSTANCE_ROOT)
|
|
except ValueError as error:
|
|
raise JobError('OCI job stack transaction is outside private instance state') from error
|
|
if (len(stack_path.parts) != 4 or not stack_path.parts[0].isdigit()
|
|
or stack_path.parts[1] != 'stack-transactions'
|
|
or not re.fullmatch(r'[0-9a-f]{32}', stack_path.parts[2])
|
|
or stack_path.parts[3] != 'transaction.json'):
|
|
raise JobError('Invalid OCI job stack transaction path')
|
|
stack = json.loads(_trusted_read(stack_journal))
|
|
phases = {'starting', 'checking', 'publishing', 'recovering'}
|
|
if not executing:
|
|
phases.add('replacing')
|
|
members = stack.get('plan', {}).get('members') or []
|
|
if (stack.get('schema_version') != 1 or stack.get('id') != coordinated.get('id')
|
|
or stack.get('phase') not in phases
|
|
or not any(member.get('vmid') == vmid
|
|
and member.get('installation_id') == record.get('installation_id')
|
|
for member in members)):
|
|
raise JobError('OCI job is not authorized by the active stack transaction')
|
|
return child.get('registry_digest')
|
|
|
|
|
|
def job_spec_from_record(record, config_text, timeout_seconds=300):
|
|
"""Bind a first installation, staged update or restored member for its hook."""
|
|
if isinstance(config_text, bytes):
|
|
config_text = config_text.decode('utf-8')
|
|
if not isinstance(config_text, str) or not isinstance(record, dict):
|
|
raise JobError('Invalid OCI job binding input')
|
|
if type(timeout_seconds) is not int or not 30 <= timeout_seconds <= 3600:
|
|
raise JobError('OCI job timeout must be between 30 and 3600 seconds')
|
|
_vmid(record.get('vmid'))
|
|
try:
|
|
installation_id = str(uuid.UUID(record['installation_id']))
|
|
except (KeyError, ValueError, TypeError, AttributeError) as error:
|
|
raise JobError('OCI job requires an installation identity') from error
|
|
digest = _record_digest(record, config_text)
|
|
if not isinstance(digest, str) or not DIGEST.fullmatch(digest):
|
|
raise JobError('OCI job requires a verified image digest')
|
|
return {'installation_id': installation_id, 'image_digest': digest,
|
|
'expected_config_fingerprint': config_fingerprint(config_text),
|
|
'timeout_seconds': timeout_seconds}
|
|
|
|
|
|
def current_identity(vmid, spec):
|
|
"""Bind the request to both a trusted instance record and the actual CT."""
|
|
_vmid(vmid)
|
|
if not isinstance(spec, dict):
|
|
raise JobError('Invalid OCI job specification')
|
|
try:
|
|
installation_id = str(uuid.UUID(spec['installation_id']))
|
|
except (KeyError, ValueError, TypeError, AttributeError) as error:
|
|
raise JobError('OCI job requires an installation identity') from error
|
|
digest = spec.get('image_digest')
|
|
if not isinstance(digest, str) or not DIGEST.fullmatch(digest):
|
|
raise JobError('OCI job requires a verified image digest')
|
|
record = json.loads(_trusted_read(INSTANCE_ROOT / str(vmid) / 'oci-compose.json'))
|
|
if record.get('vmid') != vmid or record.get('installation_id') != installation_id:
|
|
raise JobError('OCI job installation identity changed')
|
|
config = _config_text(vmid)
|
|
if _record_digest(record, config, executing=True) != digest:
|
|
raise JobError('OCI job image digest changed')
|
|
description = next((line.partition(': ')[2] for line in config.split('\n')
|
|
if line.startswith('description: ')), '')
|
|
markers = re.findall(r'proxmenux-instance=([0-9a-fA-F-]{36})', urllib.parse.unquote(description))
|
|
if installation_id not in markers:
|
|
raise JobError('The CT does not carry the OCI job installation identity')
|
|
fingerprint = config_fingerprint(config)
|
|
if spec.get('expected_config_fingerprint', fingerprint) != fingerprint:
|
|
raise JobError('OCI job runtime configuration changed')
|
|
recovery = record.get('recovered_at', '')
|
|
if not isinstance(recovery, str):
|
|
raise JobError('Invalid OCI job recovery generation')
|
|
return {'vmid': vmid, 'installation_id': installation_id, 'image_digest': digest,
|
|
'config_fingerprint': fingerprint, 'recovery_generation': recovery}
|
|
|
|
|
|
def _status(vmid):
|
|
result = subprocess.run([PCT, 'status', str(vmid)], capture_output=True, text=True, timeout=15)
|
|
value = result.stdout.strip()
|
|
if result.returncode or value not in {'status: stopped', 'status: running'}:
|
|
raise JobError('Cannot establish the OCI job container state')
|
|
return value.removeprefix('status: ')
|
|
|
|
|
|
def _socket_name(lxcpath=LXC_PATH):
|
|
value = 14695981039346656037
|
|
for byte in ('lxc/'+lxcpath+'/monitor-sock').encode():
|
|
value = ((value ^ byte) * 1099511628211) & ((1 << 64)-1)
|
|
return f'lxc/{value:016x}/{lxcpath}'
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _subscription():
|
|
version = subprocess.run([MONITOR, '--version'], capture_output=True, text=True,
|
|
check=True, timeout=10).stdout.strip()
|
|
if version not in SUPPORTED_VERSIONS or FRAME.size != 264:
|
|
raise JobError('Unsupported native LXC monitor ABI; refusing to guess job completion')
|
|
# This CLI only bootstraps monitord when no other container has started it.
|
|
# Its lossy textual exit status is NEVER consumed.
|
|
primer = subprocess.Popen([MONITOR, '-n', '^$'], stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
close_fds=True)
|
|
receiver = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
try:
|
|
receiver.settimeout(1)
|
|
deadline = time.monotonic()+5
|
|
name = _socket_name()
|
|
while True:
|
|
try:
|
|
receiver.connect(b'\0'+name.encode())
|
|
break
|
|
except (ConnectionRefusedError, FileNotFoundError):
|
|
if time.monotonic() >= deadline:
|
|
raise JobError('Native LXC monitor subscription could not be established')
|
|
time.sleep(.05)
|
|
peer = struct.unpack('3i', receiver.getsockopt(socket.SOL_SOCKET, socket.SO_PEERCRED,
|
|
struct.calcsize('3i')))
|
|
pid, uid, gid = peer
|
|
executable = str(Path(f'/proc/{pid}/exe').resolve())
|
|
if uid != 0 or gid != 0 or executable not in {
|
|
'/usr/libexec/lxc/lxc-monitord', '/usr/lib/lxc/lxc-monitord'}:
|
|
raise JobError('Native LXC monitor peer is not the trusted root daemon')
|
|
receiver.setblocking(False)
|
|
yield receiver, {'version': version, 'peer_uid': uid, 'peer_gid': gid,
|
|
'peer_executable': executable, 'protocol': 'lxc-msg-native264'}
|
|
finally:
|
|
receiver.close()
|
|
primer.terminate()
|
|
try:
|
|
primer.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
primer.kill()
|
|
primer.wait(timeout=3)
|
|
|
|
|
|
def _frames(buffer):
|
|
messages = []
|
|
while len(buffer) >= FRAME.size:
|
|
chunk, buffer = buffer[:FRAME.size], buffer[FRAME.size:]
|
|
kind, name, value = FRAME.unpack(chunk)
|
|
if kind not in {0, 1, 2}:
|
|
raise JobError('Unknown native LXC monitor event')
|
|
try:
|
|
name = name.split(b'\0', 1)[0].decode('ascii')
|
|
except UnicodeDecodeError as error:
|
|
raise JobError('Malformed native LXC monitor event') from error
|
|
messages.append((kind, name, value))
|
|
return messages, buffer
|
|
|
|
|
|
def _run_attempt(vmid, deadline):
|
|
with _subscription() as (receiver, proof):
|
|
# Drain traffic before our start; events from unrelated guests do not
|
|
# become evidence for this job. There is no historical event replay.
|
|
pending = b''
|
|
while select.select([receiver], [], [], 0)[0]:
|
|
data = receiver.recv(8192)
|
|
if not data:
|
|
raise JobError('Native LXC monitor closed before job start')
|
|
messages, pending = _frames(pending+data)
|
|
if any(name == str(vmid) for _, name, _ in messages):
|
|
raise JobError('OCI job changed state before its start attempt')
|
|
started = False
|
|
# Fast successful PID1 jobs can make pct start return 255 because
|
|
# Proxmox can no longer query the PID. Only raw exit evidence decides.
|
|
process = subprocess.Popen([PCT, 'start', str(vmid)], stdin=subprocess.DEVNULL,
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
close_fds=True)
|
|
try:
|
|
while time.monotonic() < deadline:
|
|
if not select.select([receiver], [], [], min(1, max(.01, deadline-time.monotonic())))[0]:
|
|
continue
|
|
data = receiver.recv(8192)
|
|
if not data:
|
|
raise JobError('Native LXC monitor closed without a job exit event')
|
|
messages, pending = _frames(pending+data)
|
|
for kind, name, value in messages:
|
|
if name != str(vmid):
|
|
continue
|
|
if kind == 0 and value in {1, 2}:
|
|
started = True
|
|
if kind != 2:
|
|
continue
|
|
if not started or not 0 <= value <= 65535:
|
|
raise JobError('No fresh, valid native OCI job exit proof')
|
|
if not os.WIFEXITED(value):
|
|
reason = f'signal {os.WTERMSIG(value)}' if os.WIFSIGNALED(value) else 'unknown wait status'
|
|
raise JobError(f'OCI job CT {vmid} did not complete normally ({reason})')
|
|
exitcode = os.WEXITSTATUS(value)
|
|
if exitcode != 0:
|
|
raise JobError(f'OCI job CT {vmid} exited with code {exitcode}')
|
|
while _status(vmid) != 'stopped':
|
|
if time.monotonic() >= deadline:
|
|
raise JobError('OCI job reported exit zero but did not stop')
|
|
time.sleep(.05)
|
|
process.wait(timeout=max(.1, min(5, deadline-time.monotonic())))
|
|
return {**proof, 'raw_waitstatus': value, 'exit_code': exitcode,
|
|
'fresh_start_observed': True, 'pct_start_returncode': process.returncode}
|
|
raise JobError('OCI job timed out without verified native exit zero')
|
|
finally:
|
|
if process.poll() is None:
|
|
process.terminate()
|
|
try:
|
|
process.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait(timeout=3)
|
|
|
|
|
|
def _cache_matches(cache, identity):
|
|
return (isinstance(cache, dict) and cache.get('schema') == 1
|
|
and cache.get('status') == 'completed' and cache.get('identity') == identity
|
|
and isinstance(cache.get('proof'), dict)
|
|
and type(cache['proof'].get('raw_waitstatus')) is int
|
|
and cache['proof']['raw_waitstatus'] == 0
|
|
and cache['proof'].get('fresh_start_observed') is True)
|
|
|
|
|
|
def _read_cache(path):
|
|
try:
|
|
return json.loads(_trusted_read(path))
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
|
|
def _sync_directory(path):
|
|
descriptor = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
|
|
try:
|
|
os.fsync(descriptor)
|
|
finally:
|
|
os.close(descriptor)
|
|
|
|
|
|
def _write_cache(path, value):
|
|
descriptor, name = tempfile.mkstemp(prefix='.'+path.name+'.', dir=path.parent)
|
|
try:
|
|
with os.fdopen(descriptor, 'w', encoding='utf-8') as output:
|
|
json.dump(value, output, sort_keys=True)
|
|
output.write('\n')
|
|
output.flush()
|
|
os.fsync(output.fileno())
|
|
os.replace(name, path)
|
|
_sync_directory(path.parent)
|
|
finally:
|
|
with contextlib.suppress(FileNotFoundError):
|
|
os.unlink(name)
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _job_lock(vmid, deadline):
|
|
_private_directory(STATE_ROOT)
|
|
fd = os.open(STATE_ROOT/f'{vmid}.lock', os.O_CREAT|os.O_RDWR|os.O_NOFOLLOW|os.O_CLOEXEC, 0o600)
|
|
try:
|
|
info = os.fstat(fd)
|
|
if not stat.S_ISREG(info.st_mode) or info.st_uid != 0 or info.st_mode & 0o077:
|
|
raise JobError('Untrusted OCI job lock file')
|
|
while True:
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_EX|fcntl.LOCK_NB)
|
|
break
|
|
except BlockingIOError:
|
|
if time.monotonic() >= deadline:
|
|
raise JobError('Timed out waiting for another OCI job attempt')
|
|
time.sleep(.05)
|
|
yield
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def ensure_completed(vmid, job_spec, force=False):
|
|
_root()
|
|
_vmid(vmid)
|
|
if type(force) is not bool:
|
|
raise JobError('Invalid OCI job force flag')
|
|
timeout = job_spec.get('timeout_seconds', 300) if isinstance(job_spec, dict) else None
|
|
if type(timeout) is not int or not 30 <= timeout <= 3600:
|
|
raise JobError('OCI job timeout must be between 30 and 3600 seconds')
|
|
deadline = time.monotonic()+timeout
|
|
with _job_lock(vmid, deadline):
|
|
identity = current_identity(vmid, job_spec)
|
|
if _status(vmid) != 'stopped':
|
|
raise JobError('OCI job is already running without this attempt\'s exit subscription')
|
|
path = STATE_ROOT/f'{vmid}.json'
|
|
cache = _read_cache(path)
|
|
if not force and _cache_matches(cache, identity):
|
|
return {'status': 'completed', 'cached': True, **identity, 'proof': cache['proof']}
|
|
# A forced failed retry must never leave an earlier success reusable.
|
|
if path.exists():
|
|
path.unlink()
|
|
_sync_directory(path.parent)
|
|
proof = _run_attempt(vmid, deadline)
|
|
if current_identity(vmid, job_spec) != identity:
|
|
raise JobError('OCI job identity or configuration changed during execution')
|
|
if _status(vmid) != 'stopped':
|
|
raise JobError('OCI job restarted before completion could be recorded')
|
|
value = {'schema': 1, 'status': 'completed', 'identity': identity, 'proof': proof,
|
|
'attempt_id': str(uuid.uuid4()),
|
|
'completed_at': datetime.datetime.now(datetime.timezone.utc).isoformat()}
|
|
_write_cache(path, value)
|
|
return {'status': 'completed', 'cached': False, **identity, 'proof': proof}
|
|
|
|
|
|
def forget_completed(vmid, installation_id):
|
|
"""Remove only this installation's completion, never a reused VMID's state."""
|
|
_vmid(vmid)
|
|
try:
|
|
installation_id = str(uuid.UUID(installation_id))
|
|
except (ValueError, TypeError, AttributeError):
|
|
return False # Older non-job records may use a legacy identity format.
|
|
if not STATE_ROOT.exists():
|
|
return False
|
|
_root()
|
|
with _job_lock(vmid, time.monotonic()+30):
|
|
path = STATE_ROOT/f'{vmid}.json'
|
|
cache = _read_cache(path)
|
|
if (not isinstance(cache, dict) or not isinstance(cache.get('identity'), dict)
|
|
or cache['identity'].get('installation_id') != installation_id):
|
|
return False
|
|
path.unlink()
|
|
_sync_directory(path.parent)
|
|
# The stable lock inode stays: unlinking it could split concurrent
|
|
# waiters across two locks when this numeric VMID is reused.
|
|
return True
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('operation', choices=['ensure'])
|
|
parser.add_argument('vmid', type=int)
|
|
parser.add_argument('--spec-json', required=True)
|
|
parser.add_argument('--force', action='store_true')
|
|
args = parser.parse_args()
|
|
try:
|
|
if len(args.spec_json) > 65536:
|
|
raise JobError('OCI job specification is too large')
|
|
result = ensure_completed(args.vmid, json.loads(args.spec_json), force=args.force)
|
|
except (JobError, OSError, ValueError, KeyError, subprocess.SubprocessError) as error:
|
|
print(str(error), file=__import__('sys').stderr)
|
|
return 1
|
|
print(json.dumps(result, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|