room_glitch/tools/deploy/receiver.py

"""Uploaded over authenticated SSH; no daemon and no root required."""
import argparse
import configparser
import hashlib
import importlib.util
import io
import json
import os
from pathlib import Path
import platform
import re
import shutil
import subprocess
import sys
import tarfile
import tempfile

FILES = (
    'ArcadeEscape.arm64', 'ArcadeEscape.pck',
    'tools/launch/launch.py', 'tools/launch/pi_service.py',
    'tools/launch/remote_cabinets.py', 'tools/pi_gpio/pi_gpio_bridge.py',
    'tools/room_controller/server.py', 'tools/room_controller/operator.html',
    'tools/room_controller/catalog.json',
    'tools/pi_gpio/device_feedback.py',
    'tools/pi_gpio/rfid_helper/Cargo.toml', 'tools/pi_gpio/rfid_helper/src/main.rs',
)
MAX_BYTES = 1024 * 1024 * 1024
SERVICE = 'arcade-cabinet.service'
MARKER = '# Managed by Arcade Escape pi_service.py'


def release_id(value):
    if not isinstance(value, str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,63}', value):
        raise ValueError('Invalid release ID')
    return value


def private_write(path, text):
    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
    fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, 'w', encoding='utf-8', newline='\n') as output:
        output.write(text)
    path.chmod(0o600)


def unpack(stream, destination):
    """Read only the exact regular files named by our bounded manifest."""
    with tarfile.open(fileobj=stream, mode='r|gz') as archive:
        first = next(iter(archive), None)
        if first is None or first.name != 'manifest.json' or not first.isfile() or first.size > 65536:
            raise ValueError('Release must begin with a small regular manifest.json')
        manifest = json.load(archive.extractfile(first))
        release_id(manifest['release'])
        expected = manifest['files']
        if set(expected) != set(FILES):
            raise ValueError('Release contains an unexpected file set')
        if any(not isinstance(v, str) or not re.fullmatch('[0-9a-f]{64}', v) for v in expected.values()):
            raise ValueError('Invalid release checksums')
        seen, total = set(), 0
        # TarFile iteration begins at its cached first member; skip the manifest once.
        for member in archive:
            if member is first:
                continue
            if member.name not in expected or member.name in seen or not member.isfile():
                raise ValueError('Unexpected, duplicate, or nonregular archive entry')
            total += member.size
            if total > MAX_BYTES:
                raise ValueError('Release exceeds size limit')
            target = destination / member.name
            target.parent.mkdir(parents=True, exist_ok=True)
            digest = hashlib.sha256()
            with archive.extractfile(member) as source, target.open('xb') as output:
                while chunk := source.read(1024 * 1024):
                    digest.update(chunk)
                    output.write(chunk)
            if digest.hexdigest() != expected[member.name]:
                raise ValueError('Checksum mismatch: ' + member.name)
            target.chmod(0o700 if member.name == 'ArcadeEscape.arm64' else 0o600)
            seen.add(member.name)
        if seen != set(expected):
            raise ValueError('Incomplete release')
    private_write(destination / 'manifest.json', json.dumps(manifest, indent=2))
    return manifest


def verify_release(folder):
    manifest = json.loads((folder / 'manifest.json').read_text(encoding='utf-8'))
    if set(manifest['files']) != set(FILES):
        raise ValueError('Unexpected release contents')
    for name, expected in manifest['files'].items():
        target = folder / name
        if target.is_symlink() or hashlib.sha256(target.read_bytes()).hexdigest() != expected:
            raise ValueError('Release verification failed: ' + name)
    return manifest


def preflight():
    runtime = Path(os.environ.get('XDG_RUNTIME_DIR', '/run/user/' + str(os.getuid())))
    return dict(architecture=platform.machine(), python=list(sys.version_info[:3]),
                gpio_packages=all(importlib.util.find_spec(m) is not None for m in ('gpiozero', 'lgpio')),
                display_ready=(runtime / 'arcade-cabinet-display.env').is_file(),
                free_bytes=shutil.disk_usage(Path.home()).free)


def render_config(settings):
    cabinet, token, url = (settings[k] for k in ('id', 'token', 'url'))
    if not re.fullmatch(r'[A-Za-z0-9_]+', cabinet):
        raise ValueError('Invalid cabinet ID')
    if not re.fullmatch(r'[A-Za-z0-9_-]{16,128}', token):
        raise ValueError('Invalid cabinet token')
    from urllib.parse import urlsplit
    parsed = urlsplit(url)
    if parsed.scheme not in ('http', 'https') or not parsed.hostname or parsed.username or parsed.password or parsed.path not in ('', '/') or parsed.query or parsed.fragment or any(c.isspace() for c in url):
        raise ValueError('Invalid controller URL')
    if type(settings.get('gpio')) is not bool:
        raise ValueError('GPIO must be true or false')
    pins = settings.get('pins', [])
    if not isinstance(pins, list) or any(type(p) is not int or not 0 <= p <= 27 for p in pins) or len(set(pins)) != len(pins) or (settings['gpio'] and not pins):
        raise ValueError('Invalid BCM pin allowlist')
    controller = settings.get('gpio_controller', cabinet + '_gpio')
    if not re.fullmatch(r'[A-Za-z0-9_]+', controller):
        raise ValueError('Invalid GPIO controller ID')
    config = configparser.ConfigParser(interpolation=None)
    config['cabinet'] = dict(executable='../../ArcadeEscape.arm64', id=cabinet,
                             url=url.rstrip('/'), token=token, gpio=str(settings['gpio']).lower())
    config['gpio'] = dict(controller=controller, pins=','.join(map(str, pins)),
                           ledger='~/.local/state/arcade_gpio/requests.sqlite3')
    config['launcher'] = dict(logs='~/.local/state/arcade_cabinet/logs')
    output = io.StringIO()
    config.write(output)
    return output.getvalue()


def stage(root, stream):
    releases = root / 'releases'
    releases.mkdir(parents=True, exist_ok=True, mode=0o700)
    with tempfile.TemporaryDirectory(prefix='.incoming-', dir=releases) as temp:
        folder = Path(temp)
        manifest = unpack(stream, folder)
        target = releases / manifest['release']
        if target.exists():
            if verify_release(target) != manifest:
                raise ValueError('Release ID already exists with different contents')
            print('Release already staged and verified: ' + manifest['release'])
            return
        folder.rename(target)
    print('Staged and verified: ' + manifest['release'] + '; not activated')


def activate(root, release, settings):
    folder = root / 'releases' / release_id(release)
    verify_release(folder)
    text = render_config(settings)
    catalog = json.loads((folder / 'tools/room_controller/catalog.json').read_text(encoding='utf-8'))
    if settings['id'] not in catalog['cabinets']:
        raise ValueError('Cabinet ID is not included in this release catalog')
    if subprocess.run(['systemctl', '--user', 'is-active', '--quiet', SERVICE]).returncode == 0:
        raise ValueError('Cabinet is running. Stop it explicitly before activating a release.')
    home = Path.home() / '.config'
    managed = [home / 'systemd/user' / SERVICE, home / 'autostart/arcade-cabinet-display.desktop']
    previous = {}
    for path in managed:
        previous[path] = path.read_bytes() if path.exists() else None
        if previous[path] is not None and not previous[path].decode('utf-8').startswith(MARKER):
            raise ValueError('Refusing to replace an unrelated service/helper')
    config = folder / 'tools/launch/launcher.ini'
    old_config = config.read_bytes() if config.exists() else None
    private_write(config, text)
    try:
        subprocess.run([sys.executable, str(folder / 'tools/launch/pi_service.py'), 'install',
                        '--remote', '--config', str(config)], check=True)
        # Informational receipt; service refers directly to the immutable release path.
        private_write(root / 'active.json', json.dumps(dict(release=release, cabinet=settings['id'])))
    except Exception:
        for path, content in previous.items():
            if content is None:
                path.unlink(missing_ok=True)
            else:
                private_write(path, content.decode('utf-8'))
        if old_config is None:
            config.unlink(missing_ok=True)
        else:
            private_write(config, old_config.decode('utf-8'))
        subprocess.run(['systemctl', '--user', 'daemon-reload'], check=False)
        raise
    print('Activated ' + release + '. Service is stopped; start after desktop readiness is confirmed.')


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('action', choices=['preflight', 'stage', 'activate'])
    parser.add_argument('--release')
    args = parser.parse_args()
    try:
        if not sys.platform.startswith('linux') or os.getuid() == 0:
            raise ValueError('Use the normal Pi desktop account, not root')
        if platform.machine() not in ('aarch64', 'arm64'):
            raise ValueError('A 64-bit ARM Raspberry Pi OS is required')
        root = Path.home() / '.local/share/arcade-deploy'
        if args.action == 'preflight':
            print(json.dumps(preflight()))
        elif args.action == 'stage':
            stage(root, sys.stdin.buffer)
        else:
            activate(root, args.release, json.loads(sys.stdin.read(65536)))
        return 0
    except (OSError, ValueError, KeyError, TypeError, tarfile.TarError, subprocess.CalledProcessError) as exc:
        print('Deployment failed: ' + str(exc), file=sys.stderr)
        return 1


if __name__ == '__main__':
    sys.exit(main())