#!/usr/bin/env python3 """Install only this project's admission hook; never restart application containers.""" import argparse import base64 import fcntl import hashlib import json import os from pathlib import Path import subprocess import sys import tempfile import types BASE = Path('/mnt/PoolOne/apps/youyidian-3d') POLICY = BASE / 'release-policy/desktop-admission' GUARD = POLICY / 'guard.py' UNIT = 'youyidian-3d-material-local-proxy.service' UNIT_FILE = Path('/etc/systemd/system') / UNIT DROPIN = Path(str(UNIT_FILE) + '.d/30-desktop-admission.conf') PROXY = BASE / 'local-proxy/youyidian-3d-material' def dropin_text(): # An independent drop-in survives legacy writers replacing the base unit. return ('[Service]\n' f'ExecStartPre=/usr/bin/python3 {GUARD}\n' 'ExecReload=\n' f'ExecReload=/usr/bin/python3 {GUARD}\n' f'ExecReload=/usr/sbin/nginx -p {PROXY}/ -c {PROXY}/nginx.conf -s reload\n') def run(args): value = subprocess.run(args, capture_output=True, text=True, timeout=25) if value.returncode != 0: raise RuntimeError('system_command_failed') return value.stdout.strip() def atomic_write(path, data, mode): fd, temp = tempfile.mkstemp(prefix='.admission-install-', dir=path.parent) try: with os.fdopen(fd, 'wb') as stream: os.fchown(stream.fileno(), 0, 0) os.fchmod(stream.fileno(), mode) stream.write(data) stream.flush() os.fsync(stream.fileno()) os.replace(temp, path) finally: if os.path.exists(temp): os.unlink(temp) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--apply', action='store_true') parser.add_argument('--verify-reload', action='store_true') parser.add_argument('--expect-container', required=True) parser.add_argument('--expect-image', required=True) parser.add_argument('--expect-commit', required=True) parser.add_argument('--entry-sha256', required=True) args = parser.parse_args() handles = [] try: if os.geteuid() != 0: raise RuntimeError('root_required') if 'STAGED_POLICY_BASE64' in globals(): source = base64.b64decode(STAGED_POLICY_BASE64, validate=True) else: source = Path(__file__).with_name('desktop-release-admission.py').read_bytes() source = source.replace(b'\r\n', b'\n').replace(b'\r', b'\n') digest = hashlib.sha256(source).hexdigest() module = types.ModuleType('desktop_admission_install_probe') module.__file__ = str(GUARD) exec(compile(source, str(GUARD), 'exec'), module.__dict__) require = module.require # Same lock order as the canonical deployment, scoped to this project. for name in ['youyidian-3d-material-production-deploy.lock', 'youyidian-3d-material-local-proxy-reconcile.lock']: flags = os.O_WRONLY | os.O_NOFOLLOW if args.apply: flags |= os.O_CREAT fd = os.open('/run/' + name, flags, 0o600) handles.append(fd) fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) snapshot = module.current_snapshot() require(snapshot['containerId'] == args.expect_container and snapshot['imageId'] == args.expect_image and snapshot['revision'] == args.expect_commit, 'active_identity_changed') module.physical(UNIT_FILE, owner=(0, 0)) require(run(['systemctl', 'show', UNIT, '-p', 'FragmentPath', '--value']) == str(UNIT_FILE), 'unit_identity_changed') require(run(['systemctl', 'is-active', UNIT]) == 'active', 'proxy_not_active') before_pid = run(['systemctl', 'show', UNIT, '-p', 'MainPID', '--value']) require(before_pid.isdecimal() and before_pid != '0', 'proxy_pid_invalid') if GUARD.exists(): module.physical(GUARD, owner=(0, 0), mode=0o555) require(hashlib.sha256(GUARD.read_bytes()).hexdigest() == digest, 'existing_policy_requires_reviewed_upgrade') for path in [POLICY.parent, POLICY, POLICY / 'receipts', DROPIN.parent]: require(not path.is_symlink(), 'symlink_rejected') if path.exists(): module.physical(path, directory=True, owner=(0, 0)) if DROPIN.exists(): module.physical(DROPIN, owner=(0, 0), mode=0o644) require(DROPIN.read_text() == dropin_text(), 'existing_dropin_differs') result = {'mode': 'apply' if args.apply else 'plan', 'policySha256': digest, 'unit': UNIT, 'dropIn': str(DROPIN), 'applicationContainerId': snapshot['containerId'], 'imageId': snapshot['imageId'], 'revision': snapshot['revision'], 'beforeProxyPid': before_pid} if not args.apply: print(json.dumps(result)) return 0 for path in [POLICY.parent, POLICY, POLICY / 'receipts']: created = not path.exists() path.mkdir(mode=0o700, exist_ok=True) if created: # NAS root's primary group need not be gid 0. os.chown(path, 0, 0) os.chmod(path, 0o700) module.physical(POLICY, directory=True, owner=(0, 0), mode=0o700) module.physical(POLICY / 'receipts', directory=True, owner=(0, 0), mode=0o700) if not GUARD.exists(): atomic_write(GUARD, source, 0o555) module.attest(snapshot, args.entry_sha256) module.validate_receipt(snapshot) created_dropin_dir = not DROPIN.parent.exists() DROPIN.parent.mkdir(mode=0o755, exist_ok=True) if created_dropin_dir: os.chown(DROPIN.parent, 0, 0) os.chmod(DROPIN.parent, 0o755) if not DROPIN.exists(): atomic_write(DROPIN, dropin_text().encode(), 0o644) run(['systemctl', 'daemon-reload']) reload_commands = run(['systemctl', 'show', UNIT, '-p', 'ExecReload', '--value']) start_commands = run(['systemctl', 'show', UNIT, '-p', 'ExecStartPre', '--value']) require(str(GUARD) in reload_commands and reload_commands.index(str(GUARD)) < reload_commands.index('/usr/sbin/nginx'), 'reload_hook_not_active') require(str(GUARD) in start_commands, 'start_hook_not_active') if args.verify_reload: run(['systemctl', 'reload', UNIT]) require(run(['systemctl', 'show', UNIT, '-p', 'MainPID', '--value']) == before_pid, 'proxy_process_restarted') require(module.current_snapshot() == snapshot, 'application_changed_during_install') result.update({'installed': True, 'reloadVerified': args.verify_reload, 'applicationUnchanged': True, 'proxyPidUnchanged': True, 'startHookVerified': True, 'reloadHookVerified': True}) print(json.dumps(result)) return 0 except Exception as error: # Do not emit arbitrary subprocess output, HTTP bodies or env values. allowed = {'root_required', 'system_command_failed'} reason = str(error) if type(error).__name__ == 'Rejected' or str(error) in allowed else 'installation_stopped' print(json.dumps({'installed': False, 'reason': reason}), file=sys.stderr) return 1 finally: for fd in handles: os.close(fd) if __name__ == '__main__': sys.exit(main())