Resume active profiles safely after forward APT upgrades

source/expanded/framebridge_expanded/package_upgrade.py

@@ -0,0 +1,440 @@
+"""Resume only previously running installed profiles after an APT upgrade.
+
+The pre-install snapshot never stops the process carrying APT. The configure
+hook schedules an isolated system service; it waits for dpkg to release its
+lock before restarting the captured profiles as their original Linux account.
+No credentials, boot policy or unrelated services are changed.
+"""
+from __future__ import annotations
+
+import argparse
+from contextlib import contextmanager
+import fcntl
+import hashlib
+import json
+import os
+from pathlib import Path
+import pwd
+import re
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+
+ROOT = Path('/var/lib/framebridge-expanded/package-upgrade')
+MODULE = '/usr/lib/framebridge-expanded/expanded/framebridge_expanded/package_upgrade.py'
+PREFIX = '/usr/lib/framebridge-expanded'
+PYTHONS = {'/usr/bin/python3', '/usr/bin/python3.12', '/usr/bin/python3.13', '/usr/bin/python3.14'}
+ENV = {'PATH': '/usr/sbin:/usr/bin:/sbin:/bin', 'LANG': 'C.UTF-8', 'LC_ALL': 'C.UTF-8'}
+UNIT = re.compile(r'framebridge-(?:autostart-[0-9a-f]{16}|profile-[0-9a-f]{20}|qa-[A-Za-z0-9-]{1,100})\.service\Z')
+MAXIMUM = 2 * 1024 * 1024
+
+
+def _need(value, message):
+    if not value:
+        raise RuntimeError(message)
+
+
+def _read(path, uid, maximum=MAXIMUM):
+    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
+    try:
+        info = os.fstat(fd)
+        _need(stat.S_ISREG(info.st_mode) and info.st_uid == uid
+              and info.st_nlink == 1 and not info.st_mode & 0o077
+              and info.st_size <= maximum, 'UPGRADE_FILE_UNTRUSTED')
+        data = os.read(fd, maximum + 1)
+        _need(len(data) <= maximum, 'UPGRADE_FILE_TOO_LARGE')
+        return data
+    finally:
+        os.close(fd)
+
+
+def _private(path, uid):
+    info = path.lstat()
+    _need(stat.S_ISDIR(info.st_mode) and info.st_uid == uid
+          and stat.S_IMODE(info.st_mode) == 0o700, 'UPGRADE_DIRECTORY_UNTRUSTED')
+
+
+def _root():
+    _need(os.getuid() == 0 and os.geteuid() == 0, 'UPGRADE_ROOT_REQUIRED')
+    for parent in [Path('/var/lib'), Path('/var/lib/framebridge-expanded')]:
+        if not parent.exists():
+            parent.mkdir(mode=0o700)
+        info = parent.lstat()
+        _need(stat.S_ISDIR(info.st_mode) and info.st_uid == 0
+              and not info.st_mode & 0o022, 'UPGRADE_ROOT_PARENT_UNTRUSTED')
+    ROOT.mkdir(mode=0o700, exist_ok=True)
+    _private(ROOT, 0)
+
+
+def _write(path, value):
+    fd, name = tempfile.mkstemp(prefix='.upgrade-', dir=path.parent)
+    try:
+        with os.fdopen(fd, 'wb') as stream:
+            os.fchmod(stream.fileno(), 0o600)
+            stream.write((json.dumps(value, indent=2, sort_keys=True) + '\n').encode())
+            stream.flush()
+            os.fsync(stream.fileno())
+        os.replace(name, path)
+    finally:
+        if os.path.exists(name):
+            os.unlink(name)
+
+
+def _birth(proc):
+    fields = (proc / 'stat').read_text().rsplit(') ', 1)[1].split()
+    return fields[19] if fields[0] not in {'Z', 'X'} else None
+
+
+def identify(proc):
+    args = [part.decode() for part in (proc / 'cmdline').read_bytes().rstrip(b'\0').split(b'\0')]
+    if (len(args) != 6 or args[0] not in PYTHONS
+            or args[1:5] != ['-m', 'framebridge_expanded.runtime', '_manager', '--state']):
+        return None
+    env = dict(x.split(b'=', 1) for x in (proc / 'environ').read_bytes().split(b'\0') if b'=' in x)
+    if (env.get(b'FRAMEBRIDGE_EXPANDED_PREFIX') != PREFIX.encode()
+            or env.get(b'FRAMEBRIDGE_EXPANDED_STATE') != args[5].encode()
+            or env.get(b'PYTHONPATH', b'').split(b':')[0] != (PREFIX + '/expanded').encode()):
+        return None
+    uid = proc.stat().st_uid
+    _need(uid != 0, 'UPGRADE_ROOT_PROFILE_REFUSED')
+    state = Path(args[5])
+    _need(state.is_absolute() and str(state) == os.path.normpath(str(state))
+          and not any(ord(c) < 32 for c in str(state)), 'UPGRADE_PROFILE_PATH_INVALID')
+    for parent in [state, *state.parents]:
+        _need(not parent.is_symlink(), 'UPGRADE_PROFILE_SYMLINK_REFUSED')
+    _private(state, uid)
+    birth = _birth(proc)
+    _need(birth is not None, 'UPGRADE_MANAGER_NOT_ALIVE')
+    record = json.loads(_read(state / 'manager.json', uid))
+    _need(record.get('pid') == int(proc.name) and record.get('birth') == birth
+          and record.get('uid') == uid, 'UPGRADE_MANAGER_RECORD_MISMATCH')
+    account = pwd.getpwuid(uid)
+    config = json.loads(_read(state / 'config.json', uid))
+    _need(config.get('role') in {'node', 'central', 'standalone'}, 'UPGRADE_CONFIG_INVALID')
+    group = (proc / 'cgroup').read_text().splitlines()
+    units = [part for line in group for part in line.split('/') if UNIT.fullmatch(part)]
+    _need(len(set(units)) <= 1, 'UPGRADE_MANAGER_UNIT_AMBIGUOUS')
+    return {'uid': uid, 'user': account.pw_name, 'home': account.pw_dir,
+            'state': str(state), 'pid': int(proc.name), 'birth': birth,
+            'config_sha256': hashlib.sha256(_read(state / 'config.json', uid)).hexdigest(),
+            'node_id': config.get('node_id'), 'unit': next(iter(set(units)), None),
+            'phase': 'CAPTURED'}
+
+
+def snapshot(proc_root=Path('/proc')):
+    _root()
+    rows = []
+    for proc in proc_root.iterdir():
+        if not proc.name.isdigit():
+            continue
+        try:
+            row = identify(proc)
+        except (FileNotFoundError, ProcessLookupError, PermissionError, UnicodeError, IndexError):
+            continue
+        if row:
+            rows.append(row)
+    keys = [(r['uid'], r['state']) for r in rows]
+    _need(len(keys) == len(set(keys)), 'UPGRADE_DUPLICATE_PROFILE_MANAGERS')
+    value = {'schema': 1, 'phase': 'CAPTURED', 'profiles': rows,
+             'transaction': os.urandom(8).hex(),
+             'captured_utc': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
+             'boot_policy_changed': False, 'credentials_changed': False}
+    pending = ROOT / 'pending.json'
+    if pending.exists() or pending.is_symlink():
+        previous = json.loads(_read(pending, 0))
+        _need(previous.get('phase') not in {'SCHEDULED', 'RUNNING'}, 'UPGRADE_PREVIOUS_WORKER_PENDING')
+        _write(ROOT / 'previous.json', previous)
+    _write(pending, value)
+    return value
+
+
+def schedule():
+    _root()
+    pending = ROOT / 'pending.json'
+    if not pending.exists():
+        return {'phase': 'NO_UPGRADE_SNAPSHOT'}
+    value = json.loads(_read(pending, 0))
+    if value.get('phase') in {'COMPLETE', 'SCHEDULED', 'RUNNING'}:
+        return value
+    _need(value.get('phase') == 'CAPTURED', 'UPGRADE_REVIEW_PREVIOUS_FAILURE')
+    if not value['profiles']:
+        value['phase'] = 'COMPLETE'
+        _write(pending, value)
+        return value
+    transaction = value.get('transaction', '')
+    _need(re.fullmatch(r'[a-f0-9]{16}', transaction), 'UPGRADE_TRANSACTION_INVALID')
+    value['phase'] = 'SCHEDULED'
+    _write(pending, value)
+    try:
+        result = subprocess.run([
+            '/usr/bin/systemd-run', '--quiet', '--collect',
+            '--unit=framebridge-package-upgrade-' + transaction, '--on-active=2s',
+            '--timer-property=RemainAfterElapse=no',
+            '--property=Type=oneshot', '--property=TimeoutStartSec=600',
+            '/usr/bin/python3', '-I', '-S', MODULE, 'resume'],
+            env=ENV, capture_output=True, timeout=15)
+        _need(result.returncode == 0, 'UPGRADE_WORKER_SCHEDULE_FAILED')
+    except BaseException:
+        value['phase'] = 'CAPTURED'
+        _write(pending, value)
+        raise
+    return value
+
+
+@contextmanager
+def _wait_dpkg(timeout=180):
+    """Hold APT/dpkg locks throughout bounded runtime replacement."""
+    deadline = time.monotonic() + timeout
+    descriptors = []
+    held = []
+    try:
+        for name in ('lock-frontend', 'lock'):
+            descriptors.append(os.open('/var/lib/dpkg/' + name, os.O_RDWR | os.O_NOFOLLOW))
+        while True:
+            try:
+                for fd in descriptors:
+                    fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                    held.append(fd)
+                break
+            except BlockingIOError:
+                for fd in held:
+                    fcntl.lockf(fd, fcntl.LOCK_UN)
+                held = []
+                _need(time.monotonic() < deadline, 'UPGRADE_DPKG_LOCK_TIMEOUT')
+                time.sleep(0.25)
+        yield
+    finally:
+        for fd in held:
+            fcntl.lockf(fd, fcntl.LOCK_UN)
+        for fd in descriptors:
+            os.close(fd)
+
+
+def _owner(row, operation):
+    account = pwd.getpwuid(row['uid'])
+    _need(row['uid'] != 0 and account.pw_name == row['user']
+          and account.pw_dir == row['home'], 'UPGRADE_ACCOUNT_CHANGED')
+    args = ['/usr/sbin/runuser', '-u', account.pw_name, '--', '/usr/bin/env', '-i',
+            'HOME=' + account.pw_dir, 'USER=' + account.pw_name, 'LOGNAME=' + account.pw_name,
+            'PATH=/usr/bin:/bin', 'LANG=C.UTF-8', 'LC_ALL=C.UTF-8',
+            'XDG_RUNTIME_DIR=/run/user/' + str(row['uid']),
+            'DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/' + str(row['uid']) + '/bus',
+            '/usr/bin/python3', '-I', '-S', MODULE, operation,
+            '--state', row['state'], '--config-sha256', row['config_sha256'],
+            '--manager-pid', str(row['pid']), '--manager-birth', row['birth']]
+    background = row.get('background')
+    if background:
+        args += ['--connection-pid', str(background['pid']),
+                 '--connection-birth', str(background['start_ticks'])]
+    result = subprocess.run(args, env=ENV, capture_output=True, text=True, timeout=120)
+    _need(result.returncode == 0, 'UPGRADE_OWNER_OPERATION_FAILED:' + operation)
+    value = json.loads(result.stdout)
+    _need(value.get('ok') is True, 'UPGRADE_OWNER_NOT_READY:' + operation)
+    return value
+
+
+def _start_manual(row):
+    unit = 'framebridge-profile-' + hashlib.sha256(row['state'].encode()).hexdigest()[:20] + '.service'
+    base = _unit(row)[:-2]  # runuser/env for the owner's actual user bus
+    result = subprocess.run([*base, '/usr/bin/systemd-run', '--user', '--quiet', '--collect',
+        '--unit=' + unit, '--property=Type=exec', '--property=ExitType=cgroup',
+        '--property=KillMode=control-group',
+        '--setenv=FRAMEBRIDGE_EXPANDED_PREFIX=' + PREFIX,
+        '--setenv=FRAMEBRIDGE_EXPANDED_STATE=' + row['state'],
+        '--setenv=HOME=' + row['home'], '--setenv=PATH=/usr/bin:/bin', '/usr/bin/fb', '--state-dir', row['state'], '--json', 'start'],
+        env=ENV, capture_output=True, timeout=20)
+    _need(result.returncode == 0, 'UPGRADE_NATIVE_USER_UNIT_START_FAILED')
+    return unit
+
+
+def _unit(row):
+    account = pwd.getpwuid(row['uid'])
+    _need(row['uid'] != 0 and account.pw_name == row['user']
+          and account.pw_dir == row['home'], 'UPGRADE_ACCOUNT_CHANGED')
+    return ['/usr/sbin/runuser', '-u', account.pw_name, '--', '/usr/bin/env', '-i',
+            'HOME=' + account.pw_dir, 'PATH=/usr/bin:/bin',
+            'XDG_RUNTIME_DIR=/run/user/' + str(row['uid']),
+            'DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/' + str(row['uid']) + '/bus',
+            '/usr/bin/systemctl', '--user']
+
+
+def _resume():
+    _root()
+    pending = ROOT / 'pending.json'
+    value = json.loads(_read(pending, 0))
+    _need(value.get('schema') == 1 and value.get('phase') == 'SCHEDULED', 'UPGRADE_INTENT_INVALID')
+    value['phase'] = 'RUNNING'
+    _write(pending, value)
+    failures = []
+    for row in value['profiles']:
+        try:
+            current = identify(Path('/proc') / str(row['pid']))
+            _need(current and all(current[k] == row[k] for k in
+                  ('uid', 'user', 'home', 'state', 'pid', 'birth', 'config_sha256', 'unit')),
+                  'UPGRADE_MANAGER_IDENTITY_CHANGED')
+            # Migration runs as the profile owner and only recognizes the old
+            # FrameBridge desktop entry; no third-party launcher is overwritten.
+            try:
+                row['desktop'] = _owner(row, 'user-desktop')
+            except Exception as desktop_error:
+                row['desktop'] = {'ok': False, 'warning': str(desktop_error)[:160]}
+            if row['unit']:
+                current = identify(Path('/proc') / str(row['pid']))
+                _need(current and all(current[k] == row[k] for k in
+                      ('uid', 'state', 'pid', 'birth', 'config_sha256', 'unit')),
+                      'UPGRADE_MANAGER_IDENTITY_CHANGED')
+                _need(UNIT.fullmatch(row['unit']), 'UPGRADE_UNIT_INVALID')
+                result = subprocess.run([*_unit(row), 'restart', row['unit']],
+                                        env=ENV, capture_output=True, timeout=100)
+                _need(result.returncode == 0, 'UPGRADE_UNIT_RESTART_FAILED')
+                row['runtime'] = _owner(row, 'user-ready')
+            else:
+                stopped = _owner(row, 'user-stop')
+                row['background'] = stopped.get('background')
+                row['unit'] = _start_manual(row)
+                row['runtime'] = _owner(row, 'user-ready')
+            row['phase'] = 'READY'
+        except Exception as error:
+            row['phase'] = 'FAILED'
+            row['error'] = str(error)[:160]
+            failures.append(row['state'])
+        _write(pending, value)
+    value['phase'] = 'FAILED' if failures else 'COMPLETE'
+    value['completed_utc'] = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime())
+    _write(pending, value)
+    _need(not failures, 'UPGRADE_PROFILE_RESTART_FAILED: inspect /var/lib/framebridge-expanded/package-upgrade/pending.json')
+    return value
+
+
+@contextmanager
+def _resume_lock():
+    fd = os.open(ROOT / '.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600)
+    try:
+        info = os.fstat(fd)
+        _need(stat.S_ISREG(info.st_mode) and info.st_uid == 0
+              and info.st_nlink == 1 and stat.S_IMODE(info.st_mode) == 0o600,
+              'UPGRADE_LOCK_UNTRUSTED')
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        yield
+    finally:
+        os.close(fd)
+
+
+def resume():
+    _root()
+    with _resume_lock(), _wait_dpkg():
+        return _resume()
+
+
+def _background(state):
+    from framebridge_expanded import connection_background as background
+    directory = background._directory(state, False)
+    record = background._read(directory / 'status.json') if directory else {}
+    if not record:
+        return None
+    valid = background._valid_identity(record, state)
+    pid = record.get('pid')
+    if not valid:
+        _need(type(pid) is not int or not Path('/proc', str(pid)).exists(),
+              'UPGRADE_BACKGROUND_IDENTITY_UNVERIFIED')
+        return None
+    return {'pid': pid, 'start_ticks': record['start_ticks'],
+            'verified': background.status(state)['verified'],
+            'preserved_in_memory': True, 'process_reloaded': False}
+
+
+def user_operation(operation, state, config_sha256, *, manager_pid=None, manager_birth=None,
+                   connection_pid=None, connection_birth=None):
+    _need(os.getuid() != 0 and os.getuid() == os.geteuid(), 'UPGRADE_OWNER_REQUIRED')
+    state = Path(state)
+    _private(state, os.getuid())
+    data = _read(state / 'config.json', os.getuid())
+    _need(hashlib.sha256(data).hexdigest() == config_sha256, 'UPGRADE_CONFIG_CHANGED')
+    sys.path.insert(0, PREFIX + '/expanded')
+    if operation == 'user-desktop':
+        from framebridge_expanded.console_desktop import migrate_legacy
+        return migrate_legacy(state)
+    from framebridge_expanded import runtime
+    if operation == 'user-stop':
+        from framebridge_expanded import autostart
+
+        def restart(value):
+            # runtime.stop writes a manual-stop marker. Retain its former bytes
+            # even when stopping fails; upgrading must not change boot intent.
+            marker = state / 'autostart/manual-stop.json'
+            before = _read(marker, os.getuid()) if os.path.lexists(marker) else None
+            manager = json.loads(_read(state / 'manager.json', os.getuid()))
+            _need(manager.get('pid') == manager_pid and manager.get('birth') == manager_birth
+                  and _birth(Path('/proc') / str(manager_pid)) == manager_birth,
+                  'UPGRADE_MANAGER_IDENTITY_CHANGED')
+            background = _background(state)
+            try:
+                current = runtime.stop(state, preserve_connection=True)
+                deadline = time.monotonic() + 20
+                while current.get('running') or any(x.get('alive') for x in current.get('processes', [])):
+                    _need(time.monotonic() < deadline, 'UPGRADE_OLD_RUNTIME_STOP_TIMEOUT')
+                    time.sleep(0.1)
+                    current = runtime.status(state)
+            finally:
+                if before is None:
+                    marker.unlink(missing_ok=True)
+                else:
+                    fd, temporary = tempfile.mkstemp(prefix='.upgrade-stop-', dir=marker.parent)
+                    with os.fdopen(fd, 'wb') as stream:
+                        os.fchmod(stream.fileno(), 0o600)
+                        stream.write(before)
+                    os.replace(temporary, marker)
+            return {'ok': True, 'stopped': True, 'background': background,
+                    'boot_policy_changed': False}
+
+        return autostart._configuration_action(state, restart)
+    else:
+        deadline = time.monotonic() + 60
+        while True:
+            result = runtime.status(state)
+            manager = json.loads(_read(state / 'manager.json', os.getuid()))
+            if (result.get('running') and result.get('ready')
+                    and (manager.get('pid'), manager.get('birth')) != (manager_pid, manager_birth)):
+                break
+            _need(time.monotonic() < deadline, 'UPGRADE_NEW_RUNTIME_NOT_READY')
+            time.sleep(0.25)
+    _need(result.get('running') is True and result.get('ready') is True, 'UPGRADE_NEW_RUNTIME_NOT_READY')
+    background = _background(state)
+    if connection_pid:
+        _need(background and background['pid'] == connection_pid
+              and background['start_ticks'] == connection_birth, 'UPGRADE_BACKGROUND_IDENTITY_CHANGED')
+    return {'ok': True, 'ready': True, 'running': True, 'background': background,
+            'manager': json.loads(_read(state / 'manager.json', os.getuid())),
+            'boot_policy_changed': False}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('operation', choices=['snapshot', 'schedule', 'resume',
+                        'user-stop', 'user-ready', 'user-desktop'])
+    parser.add_argument('--state')
+    parser.add_argument('--config-sha256')
+    parser.add_argument('--manager-pid', type=int)
+    parser.add_argument('--manager-birth')
+    parser.add_argument('--connection-pid', type=int)
+    parser.add_argument('--connection-birth', type=int)
+    args = parser.parse_args(argv)
+    try:
+        if args.operation.startswith('user-'):
+            value = user_operation(args.operation, args.state, args.config_sha256,
+                manager_pid=args.manager_pid, manager_birth=args.manager_birth,
+                connection_pid=args.connection_pid, connection_birth=args.connection_birth)
+        else:
+            value = globals()[args.operation]()
+        print(json.dumps(value))
+        return 0
+    except Exception as error:
+        print('FrameBridge package upgrade: ' + str(error), file=sys.stderr)
+        return 1
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
+

source/expanded/framebridge_expanded/runtime.py

@@ -466,11 +466,13 @@ def start(state):
     result.update(ok=False, error=result["error"] or "Startup failed. Check this profile's logs.")
     return result
 
-def stop(state):
+def stop(state, *, preserve_connection=False):
     state = state_path(state)
     load_config(state)
-    from .connection_background import disconnect
-    connection = disconnect(state)
+    from .connection_background import disconnect, status as connection_status
+    # Upgrade-only internal scope: retain an identity-verified background
+    # supervisor while local tools restart. Ordinary CLI stop is unchanged.
+    connection = connection_status(state) if preserve_connection else disconnect(state)
     if not connection.get("ok"):
         raise RuntimeError("The background ChatGPT connection could not be stopped safely; local services were left unchanged.")
     from . import tty

source/packaging/build_expanded.py

@@ -173,6 +173,28 @@ def maintainer_prerm(version):
     )
 
 
+
+def maintainer_preinst(version=None):
+    """Snapshot active profiles without stopping the process carrying APT."""
+    version = version or VERSION
+    if not re.fullmatch(r"[0-9][0-9A-Za-z.+:~\\-]*", version):
+        raise ValueError("Invalid package upgrade version")
+    module = (BASE / "expanded/framebridge_expanded/package_upgrade.py").read_text()
+    module = module.rsplit("if __name__ == '__main__':", 1)[0]
+    return ("#!/bin/sh\nset -eu\ncase \"$1\" in\n  upgrade)\n"
+            '/usr/bin/dpkg --validate-version "$2"\n'
+            'if /usr/bin/dpkg --compare-versions "$2" lt ' + shlex.quote(version) + "; then\n"
+            "/usr/bin/python3 -I -S - <<'FRAMEBRIDGE_UPGRADE_SNAPSHOT'\n"
+            + module + "\nraise SystemExit(main(['snapshot']))\n"
+            + "FRAMEBRIDGE_UPGRADE_SNAPSHOT\nfi\n    ;;\nesac\n")
+
+
+def maintainer_postinst():
+    return ("#!/bin/sh\nset -eu\ncase \"$1\" in\n  configure)\n"
+            "/usr/bin/python3 -I -S /usr/lib/framebridge-expanded/expanded/framebridge_expanded/package_upgrade.py schedule\n"
+            "    ;;\nesac\n")
+
+
 def package_relationships(profile):
     """Keep the opt-in virtual window manager out of automatic APT installs."""
     return {
@@ -304,6 +326,9 @@ def main(argv=None):
         " Installing the package alone does not activate a connection.\n")
     (control / "prerm").write_text(maintainer_prerm(PRIVILEGED_HELPER_FLOOR))
     (control / "prerm").chmod(0o755)
+    for hook, contents in (("preinst", maintainer_preinst()), ("postinst", maintainer_postinst())):
+        (control / hook).write_text(contents)
+        (control / hook).chmod(0o755)
     rows = []
     for path in sorted(STAGE.rglob("*")):
         if path.is_file() and not path.is_symlink() and control not in path.parents:

source/tests/test_package_upgrade.py

@@ -0,0 +1,339 @@
+import hashlib
+from contextlib import nullcontext
+import json
+import os
+from pathlib import Path
+from types import SimpleNamespace
+import tempfile
+import unittest
+from unittest import mock
+
+from framebridge_expanded import package_upgrade as upgrade
+
+
+class PackageUpgradeTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.original_read = upgrade._read
+        self.state = self.root / 'profile'
+        self.state.mkdir(mode=0o700)
+        self.config = {'role': 'node', 'node_id': 'test-node'}
+        self.write(self.state / 'config.json', self.config)
+        self.proc = self.root / '1234'
+        self.proc.mkdir()
+        self.identity = {'pid': 1234, 'birth': '99', 'uid': os.getuid()}
+        self.write(self.state / 'manager.json', self.identity)
+        self.args = ['/usr/bin/python3', '-m', 'framebridge_expanded.runtime',
+                     '_manager', '--state', str(self.state)]
+        (self.proc / 'cmdline').write_bytes(b'\0'.join(x.encode() for x in self.args) + b'\0')
+        self.environ()
+        (self.proc / 'stat').write_text('1234 (python3) S ' + ' '.join(['0'] * 18 + ['99']))
+        (self.proc / 'cgroup').write_text('0::/user.slice/user-1000.slice/user@1000.service/app.slice/test.scope\n')
+
+    def write(self, path, data):
+        path.write_text(json.dumps(data))
+        path.chmod(0o600)
+
+    def environ(self, prefix=None):
+        prefix = upgrade.PREFIX if prefix is None else prefix
+        (self.proc / 'environ').write_bytes(b'\0'.join(x.encode() for x in [
+            'FRAMEBRIDGE_EXPANDED_PREFIX=' + prefix,
+            'FRAMEBRIDGE_EXPANDED_STATE=' + str(self.state),
+            'PYTHONPATH=' + prefix + '/expanded:' + prefix + '/portable']) + b'\0')
+
+    def identify(self):
+        if os.getuid() == 0:
+            self.skipTest('Identity fixtures run as the ordinary build owner, not root')
+        return upgrade.identify(self.proc)
+
+    def test_identifies_only_installed_manager_and_exact_owned_record(self):
+        row = self.identify()
+        self.assertEqual(row['pid'], 1234)
+        self.assertEqual(row['state'], str(self.state))
+        self.assertEqual(row['birth'], '99')
+        self.assertIsNone(row['unit'])
+        self.assertNotIn('config', row)
+
+    def test_development_clone_is_not_an_installed_service(self):
+        self.environ('/home/example/source')
+        self.assertIsNone(self.identify())
+
+    def test_arbitrary_python_command_is_ignored(self):
+        (self.proc / 'cmdline').write_bytes(b'/usr/bin/python3\0-c\0print(1)\0')
+        self.assertIsNone(upgrade.identify(self.proc))
+
+    def test_pid_reuse_is_rejected(self):
+        self.write(self.state / 'manager.json', dict(self.identity, birth='98'))
+        if os.getuid() == 0:
+            self.skipTest('Ordinary-owner fixture required')
+        with self.assertRaisesRegex(RuntimeError, 'RECORD_MISMATCH'):
+            upgrade.identify(self.proc)
+
+    def test_recognizes_qa_unit_without_selecting_unrelated_services(self):
+        (self.proc / 'cgroup').write_text('0::/user.slice/user@1000.service/app.slice/framebridge-qa-mint223-private2-c3418304.service\n')
+        self.assertEqual(self.identify()['unit'], 'framebridge-qa-mint223-private2-c3418304.service')
+
+    def test_unrelated_unit_is_not_restarted(self):
+        (self.proc / 'cgroup').write_text('0::/user.slice/user@1000.service/app.slice/owner-important.service\n')
+        self.assertIsNone(self.identify()['unit'])
+
+    def test_world_readable_metadata_is_refused(self):
+        path = self.state / 'config.json'
+        path.chmod(0o644)
+        with self.assertRaisesRegex(RuntimeError, 'FILE_UNTRUSTED'):
+            upgrade._read(path, os.getuid())
+
+    def test_metadata_symlink_is_refused(self):
+        path = self.state / 'alias.json'
+        path.symlink_to(self.state / 'config.json')
+        with self.assertRaises(OSError):
+            upgrade._read(path, os.getuid())
+
+    def test_shared_profile_directory_is_refused(self):
+        self.state.chmod(0o755)
+        with self.assertRaisesRegex(RuntimeError, 'DIRECTORY_UNTRUSTED'):
+            upgrade._private(self.state, os.getuid())
+
+    def root_read(self, path, uid, maximum=upgrade.MAXIMUM):
+        # Root-owned production intent is represented by the test owner's file.
+        return self.original_read(path, os.getuid(), maximum)
+
+    def pending(self, profiles=None):
+        value = {'schema': 1, 'transaction': '1' * 16, 'phase': 'CAPTURED', 'profiles': profiles or []}
+        self.write(self.root / 'pending.json', value)
+        return value
+
+    def test_fresh_install_does_not_schedule_or_start_profiles(self):
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade.subprocess, 'run') as run:
+            self.assertEqual(upgrade.schedule()['phase'], 'NO_UPGRADE_SNAPSHOT')
+        run.assert_not_called()
+
+    def test_empty_upgrade_stays_stopped(self):
+        self.pending()
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade.subprocess, 'run') as run:
+            self.assertEqual(upgrade.schedule()['phase'], 'COMPLETE')
+        run.assert_not_called()
+
+    def test_restart_is_detached_and_delayed_past_package_transaction(self):
+        self.pending([{'state': str(self.state)}])
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade.subprocess, 'run', return_value=SimpleNamespace(returncode=0)) as run:
+            self.assertEqual(upgrade.schedule()['phase'], 'SCHEDULED')
+        args = run.call_args.args[0]
+        self.assertEqual(args[0], '/usr/bin/systemd-run')
+        self.assertIn('--on-active=2s', args)
+        self.assertEqual(args[-2:], [upgrade.MODULE, 'resume'])
+
+    def test_scheduling_failure_keeps_captured_intent(self):
+        self.pending([{'state': str(self.state)}])
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade.subprocess, 'run', return_value=SimpleNamespace(returncode=1)):
+            with self.assertRaisesRegex(RuntimeError, 'SCHEDULE_FAILED'):
+                upgrade.schedule()
+        self.assertEqual(json.loads((self.root / 'pending.json').read_text())['phase'], 'CAPTURED')
+
+    def test_repeated_configure_does_not_schedule_second_restart(self):
+        self.write(self.root / 'pending.json', {'phase': 'SCHEDULED'})
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade.subprocess, 'run') as run:
+            self.assertEqual(upgrade.schedule()['phase'], 'SCHEDULED')
+        run.assert_not_called()
+
+    def test_changed_manager_is_not_stopped_and_failure_is_recorded(self):
+        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner',
+               'state': str(self.state), 'pid': 1234, 'birth': '99',
+               'config_sha256': 'a' * 64, 'unit': None, 'phase': 'CAPTURED'}
+        self.write(self.root / 'pending.json', {'schema': 1, 'phase': 'SCHEDULED', 'profiles': [row]})
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade, '_wait_dpkg', return_value=nullcontext()), \
+             mock.patch.object(upgrade, '_resume_lock', return_value=nullcontext()), \
+             mock.patch.object(upgrade, 'identify', return_value=dict(row, birth='100')), \
+             mock.patch.object(upgrade, '_owner') as owner:
+            with self.assertRaisesRegex(RuntimeError, 'PROFILE_RESTART_FAILED'):
+                upgrade.resume()
+        owner.assert_not_called()
+        value = json.loads((self.root / 'pending.json').read_text())
+        self.assertEqual(value['phase'], 'FAILED')
+        self.assertIn('IDENTITY_CHANGED', value['profiles'][0]['error'])
+
+    def test_manual_profile_restart_never_enables_service(self):
+        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner',
+               'state': str(self.state), 'pid': 1234, 'birth': '99',
+               'config_sha256': 'a' * 64, 'unit': None, 'phase': 'CAPTURED'}
+        self.write(self.root / 'pending.json', {'schema': 1, 'phase': 'SCHEDULED', 'profiles': [row]})
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade, '_wait_dpkg', return_value=nullcontext()), \
+             mock.patch.object(upgrade, '_resume_lock', return_value=nullcontext()), \
+             mock.patch.object(upgrade, 'identify', return_value=row), \
+             mock.patch.object(upgrade, '_owner', return_value={'ok': True, 'ready': True}) as owner, \
+             mock.patch.object(upgrade, '_start_manual', return_value='framebridge-profile-' + 'a' * 20 + '.service'), \
+             mock.patch.object(upgrade.subprocess, 'run') as run:
+            self.assertEqual(upgrade.resume()['phase'], 'COMPLETE')
+        self.assertEqual([c.args[1] for c in owner.call_args_list], ['user-desktop', 'user-stop', 'user-ready'])
+        run.assert_not_called()
+
+
+    def test_snapshot_does_not_stop_the_process_running_apt(self):
+        row = {'uid': os.getuid(), 'state': str(self.state), 'pid': 1234}
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade, 'identify', return_value=row), \
+             mock.patch.object(upgrade.subprocess, 'run') as run:
+            value = upgrade.snapshot(self.root)
+        self.assertEqual(value['phase'], 'CAPTURED')
+        run.assert_not_called()
+
+    def test_optional_desktop_migration_failure_does_not_stop_runtime_recovery(self):
+        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner',
+               'state': str(self.state), 'pid': 1234, 'birth': '99',
+               'config_sha256': 'a' * 64, 'unit': None, 'phase': 'CAPTURED'}
+        self.write(self.root / 'pending.json', {'schema': 1, 'phase': 'SCHEDULED', 'profiles': [row]})
+        def owner(profile, operation):
+            if operation == 'user-desktop':
+                raise RuntimeError('Legacy console is closed')
+            return {'ok': True, 'ready': True}
+        with mock.patch.object(upgrade, 'ROOT', self.root), \
+             mock.patch.object(upgrade, '_root'), \
+             mock.patch.object(upgrade, '_read', side_effect=self.root_read), \
+             mock.patch.object(upgrade, '_wait_dpkg', return_value=nullcontext()), \
+             mock.patch.object(upgrade, '_resume_lock', return_value=nullcontext()), \
+             mock.patch.object(upgrade, 'identify', return_value=row), \
+             mock.patch.object(upgrade, '_owner', side_effect=owner), \
+             mock.patch.object(upgrade, '_start_manual', return_value='framebridge-profile-' + 'a' * 20 + '.service'):
+            value = upgrade.resume()
+        self.assertEqual(value['phase'], 'COMPLETE')
+        self.assertEqual(value['profiles'][0]['phase'], 'READY')
+        self.assertFalse(value['profiles'][0]['desktop']['ok'])
+
+    def check_stop_failure_retains_marker(self, before):
+        if os.getuid() == 0:
+            self.skipTest('Ordinary-owner fixture required')
+        from framebridge_expanded import runtime, autostart
+        directory = self.state / 'autostart'
+        directory.mkdir(mode=0o700)
+        marker = directory / 'manual-stop.json'
+        if before is not None:
+            marker.write_bytes(before)
+            marker.chmod(0o600)
+        def fail(state, **kwargs):
+            marker.write_text('{"temporary":true}')
+            marker.chmod(0o600)
+            raise RuntimeError('simulated bounded stop failure')
+        with mock.patch.object(runtime, 'stop', side_effect=fail), \
+             mock.patch.object(upgrade, '_birth', return_value='99'), \
+             mock.patch.object(upgrade, '_background', return_value=None), \
+             mock.patch.object(runtime, 'start') as start, \
+             mock.patch.object(autostart, '_configuration_action', side_effect=lambda state, action: action(state)):
+            with self.assertRaisesRegex(RuntimeError, 'bounded stop failure'):
+                upgrade.user_operation('user-stop', str(self.state),
+                    hashlib.sha256((self.state / 'config.json').read_bytes()).hexdigest(),
+                    manager_pid=1234, manager_birth='99')
+        start.assert_not_called()
+        if before is None:
+            self.assertFalse(marker.exists())
+        else:
+            self.assertEqual(marker.read_bytes(), before)
+
+    def test_stop_failure_retains_absent_manual_stop_policy(self):
+        self.check_stop_failure_retains_marker(None)
+
+    def test_stop_failure_retains_existing_manual_stop_bytes(self):
+        self.check_stop_failure_retains_marker(b'{"manual_stop": true}\n')
+
+
+
+    def test_user_unit_restart_rejects_account_reassignment(self):
+        row = {'uid': 1000, 'user': 'owner', 'home': '/home/owner'}
+        account = SimpleNamespace(pw_name='different', pw_dir='/home/different')
+        with mock.patch.object(upgrade.pwd, 'getpwuid', return_value=account):
+            with self.assertRaisesRegex(RuntimeError, 'ACCOUNT_CHANGED'):
+                upgrade._unit(row)
+
+
+
+    def test_manual_start_uses_an_independent_user_service_cgroup(self):
+        row = {'uid': os.getuid(), 'user': 'owner', 'home': '/home/owner', 'state': str(self.state)}
+        account = SimpleNamespace(pw_name='owner', pw_dir='/home/owner')
+        with mock.patch.object(upgrade.pwd, 'getpwuid', return_value=account), \
+             mock.patch.object(upgrade.subprocess, 'run', return_value=SimpleNamespace(returncode=0)) as run:
+            if os.getuid() == 0:
+                self.skipTest('Ordinary-owner fixture required')
+            unit = upgrade._start_manual(row)
+        args = run.call_args.args[0]
+        self.assertIn('/usr/bin/systemd-run', args)
+        self.assertIn('--property=ExitType=cgroup', args)
+        self.assertIn('--property=KillMode=control-group', args)
+        self.assertIn('--unit=' + unit, args)
+        self.assertEqual(args[-5:], ['/usr/bin/fb', '--state-dir', str(self.state), '--json', 'start'])
+        self.assertNotIn('enable', args)
+
+    def test_background_identity_failure_is_refused_without_signaling(self):
+        from framebridge_expanded import connection_background as background
+        with mock.patch.object(background, '_directory', return_value=self.state), \
+             mock.patch.object(background, '_read', return_value={'pid': os.getpid()}), \
+             mock.patch.object(background, '_valid_identity', return_value=False), \
+             mock.patch.object(background, 'disconnect') as disconnect:
+            with self.assertRaisesRegex(RuntimeError, 'BACKGROUND_IDENTITY_UNVERIFIED'):
+                upgrade._background(self.state)
+        disconnect.assert_not_called()
+
+    def test_verified_background_connection_is_retained_without_reading_credentials(self):
+        from framebridge_expanded import connection_background as background
+        record = {'pid': 1234, 'start_ticks': 88}
+        with mock.patch.object(background, '_directory', return_value=self.state), \
+             mock.patch.object(background, '_read', return_value=record), \
+             mock.patch.object(background, '_valid_identity', return_value=True), \
+             mock.patch.object(background, 'status', return_value={'verified': True}):
+            value = upgrade._background(self.state)
+        self.assertTrue(value['preserved_in_memory'])
+        self.assertFalse(value['process_reloaded'])
+        self.assertEqual(value['start_ticks'], 88)
+
+    def check_runtime_stop_connection(self, preserve):
+        from framebridge_expanded import runtime, connection_background, tty, autostart
+        with mock.patch.object(runtime, 'load_config', return_value=self.config), \
+             mock.patch.object(runtime, '_read', return_value={}), \
+             mock.patch.object(runtime, '_terminate'), \
+             mock.patch.object(runtime, '_atomic'), \
+             mock.patch.object(runtime, 'status', return_value={'running': False}), \
+             mock.patch.object(tty, 'stop', return_value={'ok': True, 'cleanup_verified': True}), \
+             mock.patch.object(autostart, 'mark_manual_stop'), \
+             mock.patch.object(connection_background, 'disconnect', return_value={'ok': True}) as disconnect, \
+             mock.patch.object(connection_background, 'status', return_value={'ok': True, 'alive': True}) as status:
+            if preserve:
+                runtime.stop(self.state, preserve_connection=True)
+                disconnect.assert_not_called()
+                status.assert_called_once_with(self.state)
+            else:
+                runtime.stop(self.state)
+                disconnect.assert_called_once_with(self.state)
+                status.assert_not_called()
+
+    def test_normal_runtime_stop_still_disconnects(self):
+        self.check_runtime_stop_connection(False)
+
+    def test_upgrade_runtime_stop_retains_existing_connection(self):
+        self.check_runtime_stop_connection(True)
+
+
+if __name__ == '__main__':
+    unittest.main()