#!/usr/bin/python3
"""Project firmware upload/online download with board-specific slot writers.

No whole-disk writes, partition changes or bootloader updates.
The user owns the uploaded image; its SHA256 is confirmed in the LuCI dialog.
"""
import contextlib
import fcntl
import hashlib
import json
import lzma
import os
from pathlib import Path
import re
import shutil
import struct
import subprocess
import sys
import tempfile
import time
import zlib

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'lib'))
import one_kvm_ota as ota

MIB = 1024 * 1024
LAYOUT = 'n1-ophub-boot-rootfs1-rootfs2-shared-ext4-v1'
PROFILES = {
    'n1': dict(compatible='phicomm,n1', dtb='meson-gxl-s905d-phicomm-n1.dtb',
               kernel='zImage', staged_dtb='n1.dtb', fat_types=(0xb, 0xc),
               geometry=((4, 256, 260), (68, 256, 800))),
    'onecloud': dict(compatible='xunlei,onecloud', dtb='meson8b-onecloud.dtb',
               kernel='uImage', staged_dtb='onecloud.dtb', fat_types=(0xe,),
               geometry=((4, 256, 260), (16, 256, 272))),
    'onecloud-pro': dict(compatible='amlogic,q200', dtb='meson-gxm-octopus-planet.dtb',
               kernel='zImage', staged_dtb='onecloud-pro.dtb', fat_types=(0xb, 0xc),
               geometry=((4, 256, 260), (68, 512, 800), (700, 512, 1432))),
}
PRESERVE = ('network', 'wireless', 'dhcp', 'firewall', 'system', 'dropbear',
            'one-kvm', 'one-kvm-os', 'uhttpd', 'luci', 'dockerd', 'dockerman')
IDENTITY = Path('/etc/one-kvm-os-release')
INBOX = Path('/mnt/shared/firmware/upload.img')
STATE = Path('/mnt/shared/firmware/status.json')
JOBS = Path('/mnt/shared/firmware/jobs')
LOCK = Path('/var/lock/one-kvm-upgrade.lock')
SOURCE = Path('/etc/one-kvm-release-source.json')


def require(condition, message):
    if not condition:
        raise ValueError(message)


def cmd(*args):
    result = subprocess.run([str(x) for x in args], stdin=subprocess.DEVNULL,
                            stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    if result.returncode:
        raise RuntimeError('%s: %s' % (args[0], result.stderr.decode(errors='replace').strip()))
    return result.stdout.decode().strip()


def sha(path, progress=None):
    value = hashlib.sha256()
    total = Path(path).stat().st_size
    completed = 0
    with Path(path).open('rb') as stream:
        for chunk in iter(lambda: stream.read(MIB), b''):
            value.update(chunk)
            completed += len(chunk)
            if progress:
                progress(completed, total)
    return value.hexdigest()


def atomic(path, data, durable=True):
    path = Path(path)
    temporary = path.with_name(path.name + '.new')
    with temporary.open('w') as stream:
        stream.write(data)
        stream.flush()
        if durable:
            os.fsync(stream.fileno())
    os.replace(temporary, path)
    if durable:
        os.sync()


def report(phase, **fields):
    record = dict(phase=phase, updated_at=int(time.time()), **fields)
    # Intermediate progress must not flush all device/application writes.
    atomic(STATE, json.dumps(record, ensure_ascii=False) + '\n',
           durable=phase in ('ready', 'rebooting', 'booted', 'failed'))
    return record


def progress_reporter(phase, **fields):
    started = time.monotonic()
    last = [started]
    report(phase, completed_bytes=0, **fields)

    def progress(completed, total):
        now = time.monotonic()
        if now - last[0] >= 1 or completed == total:
            report(phase, completed_bytes=completed, total_bytes=total,
                   bytes_per_second=int(completed / max(now - started, 0.001)), **fields)
            last[0] = now
    return progress


def mount_table():
    return [line.split() for line in Path('/proc/mounts').read_text().splitlines()]


def part_device(parent, number):
    return Path('/dev') / (parent + ('p' if parent[-1].isdigit() else '') + str(number))


def table(path):
    with Path(path).open('rb') as stream:
        header = stream.read(512)
    require(len(header) == 512 and header[510:] == b'\x55\xaa', 'Invalid DOS partition table')
    parts = []
    for index in range(4):
        entry = header[446 + index * 16:462 + index * 16]
        start, sectors = struct.unpack('<II', entry[8:16])
        parts.append(dict(start=start * 512, size=sectors * 512, type=entry[4]))
    return parts


def check_identity(info, installed=None):
    board = info.get('board')
    require(board in PROFILES, 'Unsupported One-KVM board')
    engine = info.get('upgrade_engine', 'one-kvm-ophub-amlogic-v1' if board == 'n1' else None)
    adapter = ota.ADAPTERS.get(engine)
    require(adapter is not None, 'Unsupported upgrade adapter')
    for field, value in adapter.items():
        require(info.get('board' if field == 'device' else field,
                         'amlogic' if field == 'platform' else None) == value,
                'Incompatible image ' + field)
    if installed is not None:
        check_identity(installed)
        require(all(info.get(key) == installed.get(key) for key in
                    ('board', 'arch', 'layout_version', 'layout')),
                'Image does not match the running board/layout')
    return PROFILES[board]


def check_device_geometry(parts, board):
    profile = PROFILES[board]
    geometry = (parts[0]['start'], parts[0]['size'], parts[1]['start'])
    require(geometry in [tuple(n * MIB for n in row) for row in profile['geometry']]
            and parts[0]['type'] in profile['fat_types'], 'Unknown board BOOT/root geometry')
    slot_size = parts[1]['size']
    require(valid_root_size(slot_size) and all(parts[i]['type'] == 0x83
            and parts[i]['size'] == slot_size for i in (1, 2)), 'Expected matching supported root slots')
    check_slot_geometry(parts)
    require(parts[3]['size'] >= 128 * MIB, 'SHARED is too small')
    return slot_size


def check_slot_geometry(parts):
    # ophub USB provisioning leaves a 1 MiB alignment gap before ROOTFS2.
    end = parts[1]['start'] + parts[1]['size']
    require(parts[2]['start'] in (end, end + MIB), 'Unexpected slot geometry')
    require(parts[3]['type'] == 0x83 and parts[3]['start'] == parts[2]['start'] + parts[2]['size'], 'Unexpected SHARED geometry')


def env_root_arguments(env):
    append = re.findall(r'^APPEND=(.*)$', env, flags=re.M)
    require(len(append) == 1, 'Missing or duplicate APPEND in boot config')
    return re.findall(r'(?:^|\s)root=([^\s]+)', append[0])


def device():
    require(os.geteuid() == 0, 'Root is required')
    identity = json.loads(IDENTITY.read_text())
    profile = check_identity(identity)
    require(profile['compatible'].encode() in Path('/proc/device-tree/compatible').read_bytes().split(b'\0'),
            'Device tree does not match the installed board')
    mounts = mount_table()
    roots = [row for row in mounts if row[1] == '/']
    require(len(roots) == 1 and roots[0][2] == 'btrfs', 'Expected one Btrfs root mount')
    root = Path(roots[0][0]).resolve(strict=True)
    require(root.is_block_device(), 'Root source must be a block device')
    sysdev = (Path('/sys/class/block') / root.name).resolve(strict=True)
    slot = int((sysdev / 'partition').read_text())
    require(slot in (2, 3), 'Root must be slot 2 or 3')
    parent = sysdev.parent.name
    require(re.fullmatch(r'(sd[a-z]+|mmcblk\d+)', parent), 'Unsupported disk')
    disk = Path('/dev') / parent
    require(disk.is_block_device(), 'Missing parent disk')
    parts = table(disk)
    slot_size = check_device_geometry(parts, identity['board'])
    bootdev, shareddev = part_device(parent, 1), part_device(parent, 4)
    boot = [r for r in mounts if r[1] == '/boot']
    require(len(boot) == 1 and Path(boot[0][0]).resolve() == bootdev and boot[0][2] == 'vfat' and 'rw' in boot[0][3].split(','), 'BOOT must be writable on the running disk')
    shared = Path('/mnt/shared').resolve(strict=True)
    require(any(Path(r[0]).resolve() == shareddev and r[1] == str(shared) and r[2] == 'ext4' and 'rw' in r[3].split(',') for r in mounts), 'SHARED must be writable ext4 on the running disk')
    uuid = cmd('blkid', '-s', 'PARTUUID', '-o', 'value', root)
    rootarg = re.findall(r'(?:^|\s)root=([^\s]+)', Path('/proc/cmdline').read_text())
    require(len(rootarg) == 1 and rootarg[0] == 'PARTUUID=' + uuid, 'Running root PARTUUID is inconsistent')
    env = Path('/boot/uEnv.txt').read_text()
    require(env_root_arguments(env) == rootarg, 'Boot config differs from the running root')
    extlinux = None
    if identity['board'] == 'onecloud':
        script = Path('/boot/boot.cmd').read_bytes()
        require(b'0x20800000 ${LINUX}' in script and legacy_script(Path('/boot/boot.scr')) == script,
                'OneCloud boot.scr must support LINUX from uEnv.txt; use the rebuilt firmware image')
    elif identity['board'] == 'onecloud-pro':
        extlinux = Path('/boot/extlinux/extlinux.conf').read_text()
        require(extlinux_values(extlinux)['APPEND'] == parse_env(env)['APPEND'],
                'extlinux and uEnv boot arguments differ')
    target = part_device(parent, 5 - slot)
    # ophub may automount the spare slot. Permit only its normal /mnt/<name>
    # mount, and unmount it in the worker before any write. Other mounts abort.
    target_mounts = [r[1] for r in mounts if Path(r[0]).resolve() == target]
    require(all(p == '/mnt/' + target.name for p in target_mounts), 'Inactive slot has an unexpected mount')
    require(str(target) not in Path('/proc/swaps').read_text(), 'Inactive slot is used as swap')
    return dict(root=str(root), target=str(target), slot=slot, boot='/boot',
                shared=str(shared), target_mounts=target_mounts, env=env, slot_size=slot_size,
                identity=identity, extlinux=extlinux)


@contextlib.contextmanager
def locked():
    LOCK.parent.mkdir(parents=True, exist_ok=True)
    with LOCK.open('a') as stream:
        fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
        yield stream


def job_path(job_id):
    require(bool(re.fullmatch(r'[0-9a-f]{24}', job_id)), 'Invalid job ID')
    path = JOBS / job_id
    require(path.is_dir() and not path.is_symlink(), 'Unknown job')
    return path


def inherited_lock(fd):
    descriptor = os.fstat(fd)
    expected = LOCK.stat()
    require((descriptor.st_dev, descriptor.st_ino) == (expected.st_dev, expected.st_ino), 'Worker must inherit the upgrade lock')
    fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)


def valid_root_size(size):
    return type(size) is int and 1024 * MIB <= size <= 8192 * MIB and size % (256 * MIB) == 0


def valid_image_size(size):
    return valid_root_size(size - 260 * MIB)


def image_fits_slot(raw, dev):
    require(table(raw)[1]['size'] <= dev['slot_size'], 'Image system partition exceeds the installed slot; reflash to resize partitions')


def decompress_xz(source, destination, max_root_size=8192 * MIB, progress=None):
    require(source.is_file() and not source.is_symlink(), 'Upload is missing or is a symlink')
    require(0 < source.stat().st_size <= 1536 * MIB, 'Compressed image exceeds size limit')
    with source.open('rb') as stream:
        require(stream.read(6) == b'\xfd7zXZ\x00', '请选择 .img 或 .img.xz 固件')
    total = 0
    expected = None
    header = bytearray()
    compressed_hash, raw_hash = hashlib.sha256(), hashlib.sha256()
    decoder = lzma.LZMADecompressor(format=lzma.FORMAT_XZ, memlimit=128 * MIB)
    try:
        with source.open('rb') as src, destination.open('xb') as dst:
            def write_output(data):
                nonlocal total, expected
                total += len(data)
                require(total <= 260 * MIB + max_root_size, 'Decompressed image exceeds supported image size')
                if expected is None:
                    header.extend(data[:512 - len(header)])
                    if len(header) == 512:
                        require(header[510:] == b'\x55\xaa', 'Invalid DOS partition table')
                        start, sectors = struct.unpack_from('<II', header, 470)
                        expected = (start + sectors) * 512
                        require(start * 512 == 260 * MIB and valid_image_size(expected)
                                and expected <= 260 * MIB + max_root_size, 'Unsupported image capacity')
                if expected is not None:
                    require(total <= expected, 'Image length differs from partition geometry')
                # Raw images contain large unused zero-filled regions. Keep
                # those as holes on SHARED instead of writing them to eMMC.
                # Reads and hashes still see exactly the original image bytes.
                if data and data.count(0) == len(data):
                    dst.seek(len(data), os.SEEK_CUR)
                else:
                    dst.write(data)
                raw_hash.update(data)
                # 100% is reported only after the XZ checksum/EOF is verified.
                if progress and expected is not None and total < expected:
                    progress(total, expected)

            while True:
                data = src.read(MIB)
                if not data:
                    break
                compressed_hash.update(data)
                while True:
                    write_output(decoder.decompress(data, max_length=MIB))
                    data = b''
                    if decoder.eof:
                        require(not decoder.unused_data, 'Unexpected trailing data after XZ image')
                        break
                    if decoder.needs_input:
                        break
                if decoder.eof:
                    require(not src.read(1), 'Unexpected trailing data after XZ image')
                    break
            require(decoder.eof, 'Truncated XZ image')
            dst.truncate(total)
        require(total == expected and valid_image_size(total), 'Unsupported image capacity')
        if progress:
            progress(total, expected)
        return compressed_hash.hexdigest(), raw_hash.hexdigest()
    except Exception:
        destination.unlink(missing_ok=True)
        raise


@contextlib.contextmanager
def image_mounts(raw, job):
    parts = table(raw)
    require(parts[0]['start'] == 4 * MIB and parts[0]['size'] == 256 * MIB
            and parts[0]['type'] in (0xc, 0xe), 'Image BOOT layout mismatch')
    require(parts[1]['start'] == 260 * MIB and parts[1]['type'] == 0x83 and valid_root_size(parts[1]['size']), 'Image ROOTFS layout mismatch')
    require(raw.stat().st_size == parts[1]['start'] + parts[1]['size'], 'Image length differs from partition geometry')
    require(all(p['start'] == p['size'] == p['type'] == 0 for p in parts[2:]), 'Unexpected extra image partitions')
    loops, mounted = [], []
    try:
        paths = []
        for name, part, fs in zip(('image-boot', 'image-root'), parts[:2], ('vfat', 'btrfs')):
            path = job / name
            path.mkdir(exist_ok=True)
            loop = cmd('losetup', '--find', '--show', '--read-only', '--offset', part['start'], '--sizelimit', part['size'], raw)
            loops.append(loop)
            if fs == 'btrfs':
                image_uuid = cmd('blkid', '-s', 'UUID', '-o', 'value', loop)
                for row in mount_table():
                    if row[2] == 'btrfs' and Path(row[0]).exists():
                        require(cmd('blkid', '-s', 'UUID', '-o', 'value', row[0]) != image_uuid,
                                'Image UUID matches a mounted filesystem; upload a newly built image')
            cmd('mount', '-t', fs, '-o', 'ro,nosuid,nodev,noexec' + (',nologreplay' if fs == 'btrfs' else ''), loop, path)
            mounted.append(path)
            paths.append(path)
        yield tuple(paths)
    finally:
        for path in reversed(mounted):
            cmd('umount', path)
        for loop in reversed(loops):
            cmd('losetup', '-d', loop)


def legacy_payload(path, kind):
    data = path.read_bytes()
    require(len(data) >= 64, 'Truncated U-Boot payload')
    magic, header_crc, _, size, _, _, payload_crc = struct.unpack('>7I', data[:28])
    header = data[:4] + b'\0' * 4 + data[8:64]
    payload = data[64:]
    require(magic == 0x27051956 and data[29:32] == bytes((2, kind, 0))
            and size == len(payload) and zlib.crc32(header) == header_crc
            and zlib.crc32(payload) == payload_crc, 'Invalid ARM U-Boot payload: ' + path.name)
    return payload


def legacy_script(path):
    payload = legacy_payload(path, 6)
    require(len(payload) >= 8, 'Truncated U-Boot script')
    size, terminator = struct.unpack('>II', payload[:8])
    require(terminator == 0 and size == len(payload) - 8, 'Invalid U-Boot script length')
    return payload[8:]


def validate_payload(boot, root, installed=None):
    for name in ('etc', 'etc/config', 'etc/dropbear', 'root', 'www', 'usr', 'usr/bin',
                 'usr/libexec', 'lib', 'lib/modules'):
        path = root / name
        require(path.is_dir() and not path.is_symlink(), 'Unsafe image directory: ' + name)
    for name in ('etc/one-kvm-os-release', 'etc/fstab', 'etc/shadow'):
        require((root / name).is_file() and not (root / name).is_symlink(), 'Unsafe image metadata: ' + name)
    info = json.loads((root / 'etc/one-kvm-os-release').read_text())
    profile = check_identity(info, installed)
    require(info.get('upload_upgrade') is True, 'Image predates upload upgrade support; rebuild it first')
    require(re.fullmatch(r'[A-Za-z0-9_.+-]+', info.get('kernel', '')) is not None, 'Invalid kernel release')
    require((root / 'lib/modules' / info['kernel']).is_dir(), 'Kernel modules are missing')
    expected = {'zImage', 'uInitrd', 'dtb/amlogic/' + profile['dtb']}
    if info['board'] == 'onecloud':
        expected.add('uImage')
    require(set(info.get('boot_payload', {})) == expected, 'Missing kernel/DTB pairing metadata')
    for name, digest in info['boot_payload'].items():
        path = boot / name
        require(path.is_file() and not path.is_symlink() and sha(path) == digest, 'Boot payload checksum mismatch: ' + name)
    if info['board'] == 'onecloud':
        require(legacy_payload(boot / 'uImage', 2) == (boot / 'zImage').read_bytes(),
                'OneCloud uImage does not contain the paired kernel')
        require((boot / 'uImage').stat().st_size <= 16 * MIB, 'OneCloud kernel exceeds boot load region')
    for name in ('usr/bin/one-kvm', 'usr/libexec/one-kvm-upgrade', 'www/index.html'):
        require((root / name).is_file(), 'Required runtime payload missing: ' + name)
    return info


def check_upload(release=None):
    dev = device()
    JOBS.mkdir(parents=True, exist_ok=True, mode=0o700)
    job_id = os.urandom(12).hex()
    job = JOBS / job_id
    job.mkdir(mode=0o700)
    try:
        require(INBOX.is_file() and not INBOX.is_symlink(), 'Upload is missing')
        with INBOX.open('rb') as stream:
            magic = stream.read(6)
        if magic == b'\xfd7zXZ\x00':
            require(shutil.disk_usage(JOBS).free >= dev['slot_size'] + 576 * MIB,
                    'Not enough free space on SHARED to unpack the image')
            upload_name = 'upload.img.xz'
            upload = job / upload_name
            os.replace(INBOX, upload)
            os.chmod(upload, 0o600)
            upload_hash, raw_hash = decompress_xz(upload, job / 'image.img', dev['slot_size'],
                                                  progress_reporter('unpacking', job_id=job_id))
        else:
            require(valid_image_size(INBOX.stat().st_size),
                    'Upload must be a raw .img or XZ .img.xz image')
            require(shutil.disk_usage(JOBS).free >= 256 * MIB,
                    'Need 256 MiB free on SHARED to validate the raw image')
            upload_name = 'image.img'
            upload = job / upload_name
            os.replace(INBOX, upload)
            os.chmod(upload, 0o600)
            upload_hash = raw_hash = sha(upload, progress_reporter('hashing', job_id=job_id))
        report('checking', job_id=job_id)
        image_fits_slot(job / 'image.img', dev)
        with image_mounts(job / 'image.img', job) as (boot, root):
            info = validate_payload(boot, root, dev['identity'])
            require(shutil.disk_usage('/boot').free >= sum((boot / name).stat().st_size for name in info['boot_payload']) + 8 * MIB, 'Not enough free space on BOOT')
        record = dict(job_id=job_id, upload_name=upload_name, sha256=upload_hash,
                      raw_sha256=raw_hash,
                      image=info, current_slot=dev['slot'], target=dev['target'])
        if release is not None:
            require(info['kernel'] == release['kernel'], 'Downloaded kernel differs from release metadata')
            require(info.get('version') == release['version'], 'Downloaded OS version differs from release metadata')
            require(record['sha256'] == release['artifact']['sha256'], 'Downloaded image differs from release metadata')
            record['release'] = release
        atomic(job / 'checked.json', json.dumps(record) + '\n')
        return report('ready', **record)
    except Exception:
        shutil.rmtree(job)
        raise


def regular_copy(source, target):
    if source.exists() or source.is_symlink():
        require(source.is_file() and not source.is_symlink(), 'Configuration must be a regular file: ' + str(source))
        require(not target.is_symlink(), 'Configuration destination is a symlink: ' + str(target))
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(source, target)


def restore_settings(source, target, keep):
    # Never replace /etc wholesale: keep new package defaults, module autoloads,
    # upgrade policy, rpc ACLs and factory scripts from the new image.
    for name in ('etc', 'etc/config', 'etc/dropbear', 'etc/docker', 'etc/one-kvm-image', 'root', 'root/.ssh'):
        path = target / name
        require(not path.is_symlink() and (not path.exists() or path.is_dir()), 'Unsafe configuration directory: ' + name)
    if keep:
        for name in PRESERVE:
            regular_copy(source / 'etc/config' / name, target / 'etc/config' / name)
        for name in ('etc/docker/daemon.json', 'etc/uhttpd.crt', 'etc/uhttpd.key',
                     'etc/one-kvm-release-source.json'):
            regular_copy(source / name, target / name)
    regular_copy(source / 'etc/shadow', target / 'etc/shadow')
    for key in (source / 'etc/dropbear').glob('dropbear_*_host_key'):
        regular_copy(key, target / 'etc/dropbear' / key.name)
    regular_copy(source / 'etc/dropbear/authorized_keys', target / 'etc/dropbear/authorized_keys')
    regular_copy(source / 'root/.ssh/authorized_keys', target / 'root/.ssh/authorized_keys')
    # Board MAC identity survives upgrades even when network settings are reset.
    regular_copy(source / 'etc/one-kvm-image/eth0.mac', target / 'etc/one-kvm-image/eth0.mac')


def parse_env(old):
    values = {}
    for line in old.splitlines():
        if '=' in line and not line.startswith('#'):
            key, value = line.split('=', 1)
            require(key not in values, 'Duplicate uEnv key')
            values[key] = value
    require(all(key in values for key in ('LINUX', 'INITRD', 'FDT', 'APPEND')), 'Incomplete uEnv')
    require(len(re.findall(r'(?:^|\s)root=[^\s]+', values['APPEND'])) == 1, 'Ambiguous boot root argument')
    return values


def next_env(old, target_uuid, prefix, board='n1'):
    require(re.fullmatch(r'[0-9a-f]{8}-0[23]', target_uuid) is not None, 'Invalid target PARTUUID')
    require(re.fullmatch(r'/one-kvm-upgrades/[0-9a-f]{24}', prefix) is not None, 'Invalid boot payload path')
    profile = PROFILES[board]
    values = parse_env(old)
    values['APPEND'] = re.sub(r'(?<!\S)root=\S+', 'root=PARTUUID=' + target_uuid, values['APPEND'])
    values.update(LINUX=prefix + '/' + profile['kernel'], INITRD=prefix + '/uInitrd',
                  FDT=prefix + '/' + profile['staged_dtb'])
    return ''.join(key + '=' + value + '\n' for key, value in values.items())


def extlinux_values(text):
    require(len(re.findall(r'^\s*LABEL\s+\S+', text, re.M | re.I)) == 1,
            'Expected one extlinux boot entry')
    values = {}
    for key in ('LINUX', 'INITRD', 'FDT', 'APPEND'):
        matches = re.findall(r'^\s*' + key + r'\s+([^\n]+)', text, re.M | re.I)
        require(len(matches) == 1, 'Missing or duplicate extlinux ' + key)
        values[key] = matches[0].strip()
    return values


def next_extlinux(old, updated_env):
    extlinux_values(old)
    values = parse_env(updated_env)
    for key in ('LINUX', 'INITRD', 'FDT', 'APPEND'):
        old = re.sub(r'^(\s*' + key + r'\s+)[^\n]+', lambda m: m[1] + values[key],
                     old, flags=re.M | re.I)
    return old


class BootSwitchError(RuntimeError):
    def __init__(self, message, boot_switched):
        super().__init__(message)
        self.boot_switched = boot_switched


def switch_boot(dev, job, updated):
    """Keep both Pro boot entry points in sync; restore originals on I/O error."""
    boot = Path(dev['boot'])
    changes = [(boot / 'uEnv.txt', dev['env'], updated)]
    if dev.get('extlinux') is not None:
        changes.append((boot / 'extlinux/extlinux.conf', dev['extlinux'],
                        next_extlinux(dev['extlinux'], updated)))
    for path, before, after in changes:
        require(path.is_file() and not path.is_symlink() and path.read_text() == before,
                'Boot config changed during upgrade')
        name = 'uEnv' if path.name == 'uEnv.txt' else 'extlinux'
        atomic(job / (name + '.before.txt'), before)
        atomic(job / (name + '.after.txt'), after)
    changed = []
    try:
        for path, before, after in changes:
            # Include this path before atomic(): rename can succeed before fsync fails.
            changed.append((path, before))
            atomic(path, after)
            require(path.read_text() == after, 'Boot config readback mismatch')
    except Exception as error:
        restored = True
        for path, before in reversed(changed):
            try:
                atomic(path, before)
                require(path.read_text() == before, 'Boot config restore mismatch')
            except Exception:
                restored = False
        raise BootSwitchError(str(error), not restored) from error


def bind_boot_mount(root, boot_uuid):
    """OpenWrt block-mount reads UCI fstab, not the traditional /etc/fstab."""
    config = Path(root) / 'etc/config'
    require((config / 'fstab').is_file() and not (config / 'fstab').is_symlink(),
            'Missing or unsafe UCI mount configuration')
    with tempfile.TemporaryDirectory(prefix='one-kvm-uci-') as savedir:
        uci = ('uci', '-c', str(config), '-t', savedir)
        text = cmd(*uci, 'show', 'fstab')
        sections = re.findall(r"^(fstab\.[^.]+)\.target='/boot'$", text, re.M)
        # Reverse deletion keeps anonymous section indices valid.
        for section in reversed(sections):
            cmd(*uci, 'delete', section)
        section = 'fstab.one_kvm_boot'
        cmd(*uci, 'set', section + '=mount')
        for key, value in dict(target='/boot', uuid=boot_uuid, fstype='vfat', enabled='1').items():
            cmd(*uci, 'set', section + '.' + key + '=' + value)
        cmd(*uci, 'commit', 'fstab')
        require(cmd(*uci, 'get', section + '.uuid') == boot_uuid,
                'Installed UCI BOOT UUID differs')


def write_slot(dev, job, boot, root, keep, config_source=Path('/')):
    """Core writer; only the worker calls this with validated physical devices."""
    info = validate_payload(boot, root, dev['identity'])
    profile = PROFILES[info['board']]
    target = Path(dev['target'])
    boot_target = Path(dev['boot']) / 'one-kvm-upgrades' / job.name
    newroot = job / 'new-root'
    newroot.mkdir(exist_ok=True)
    # A fresh filesystem gets a unique Btrfs UUID, avoiding duplicate UUIDs
    # between the mounted image, running root, and inactive root.
    for path in dev['target_mounts']:
        cmd('umount', path)
    require(not any(Path(r[0]).resolve() == target for r in mount_table()), 'Target remains mounted')
    require(target.is_block_device(), 'Target is not a block device')
    require(int(cmd('blockdev', '--getsize64', target)) == dev['slot_size'], 'Target slot size changed')
    cmd('mkfs.btrfs', '-f', '-m', 'single', '-L', 'ROOTFS' + str(5 - dev['slot'] - 1), target)
    cmd('mount', '-t', 'btrfs', '-o', 'compress=zstd:6', target, newroot)
    try:
        cmd('btrfs', 'subvolume', 'create', newroot / 'etc')
        os.chmod(newroot / 'etc', (root / 'etc').stat().st_mode & 0o7777)
        for entry in root.iterdir():
            if entry.name not in ('etc', '.snapshots', 'lost+found'):
                cmd('cp', '-a', entry, newroot)
        for entry in (root / 'etc').iterdir():
            cmd('cp', '-a', entry, newroot / 'etc')
        restore_settings(config_source, newroot, keep)
        # Image-specific fstab must refer to the installed disk, not the image UUID.
        fstab = newroot / 'etc/fstab'
        text = fstab.read_text()
        text, count = re.subn(r'^UUID=\S+\s+/\s+.*$', 'UUID=' + cmd('blkid', '-s', 'UUID', '-o', 'value', target) + ' / btrfs defaults,noatime,compress=zstd:6 0 0', text, flags=re.M)
        require(count == 1, 'Missing or duplicate root fstab entry')
        boot_sources = [r[0] for r in mount_table() if r[1] == dev['boot'] and r[2] == 'vfat']
        require(len(boot_sources) == 1, 'BOOT mount changed during upgrade')
        boot_uuid = cmd('blkid', '-s', 'UUID', '-o', 'value', boot_sources[0])
        text, count = re.subn(r'^\S+\s+/boot\s+.*$', 'UUID=' + boot_uuid + ' /boot vfat defaults 0 2', text, flags=re.M)
        require(count == 1, 'Missing or duplicate BOOT fstab entry')
        fstab.write_text(text)
        bind_boot_mount(newroot, boot_uuid)
        (newroot / 'root/.todo_rootfs_resize').unlink(missing_ok=True)
        (newroot / '.snapshots').mkdir(exist_ok=True)
        cmd('btrfs', 'subvolume', 'snapshot', '-r', newroot / 'etc', newroot / '.snapshots/etc-000')
        cmd('sync', '-f', newroot)
        info = validate_payload(boot, newroot, dev['identity'])
    finally:
        cmd('umount', newroot)
    # Flush and remount to verify bytes read back from the installed slot.
    cmd('mount', '-t', 'btrfs', '-o', 'ro,nologreplay', target, newroot)
    try:
        validate_payload(boot, newroot, dev['identity'])
        for name in ('usr/bin/one-kvm', 'usr/libexec/one-kvm-upgrade'):
            require(sha(root / name) == sha(newroot / name), 'Installed rootfs readback differs: ' + name)
        require(sha(config_source / 'etc/shadow') == sha(newroot / 'etc/shadow'), 'Password preservation failed')
    finally:
        cmd('umount', newroot)
    boot_target.mkdir(parents=True, exist_ok=False)
    try:
        for source, name in ((profile['kernel'], profile['kernel']), ('uInitrd', 'uInitrd'),
                             ('dtb/amlogic/' + profile['dtb'], profile['staged_dtb'])):
            shutil.copyfile(boot / source, boot_target / name)
            require(sha(boot_target / name) == info['boot_payload'][source], 'Staged boot checksum mismatch')
        os.sync()
        uuid = cmd('blkid', '-s', 'PARTUUID', '-o', 'value', target)
        updated = next_env(dev['env'], uuid, '/one-kvm-upgrades/' + job.name, info['board'])
        # This is the only boot switch. Existing kernel/DTB files and all
        # bootloader sectors stay intact. FAT rename is not power-loss rollback.
        switch_boot(dev, job, updated)
    except Exception:
        # Leave staged payloads when uEnv may already refer to them.
        raise


def stop_application_services(stopped_services, init_directory=Path('/etc/init.d')):
    for service in ('one-kvm', 'dockerd'):
        script = init_directory / service
        if not script.is_file():
            continue
        status = subprocess.run([str(script), 'status'], stdin=subprocess.DEVNULL,
                                stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        if status.returncode == 0:
            stopped_services.append(service)
        cmd(script, 'stop')


def worker(job_id, expected, keep, lock_fd):
    job = job_path(job_id)
    stopped_services = []
    switched = False
    try:
        dev = device()
        checked = json.loads((job / 'checked.json').read_text())
        require(checked.get('upload_name') in ('image.img', 'upload.img.xz'), 'Invalid checked upload name')
        require(expected == checked['sha256'] == sha(job / checked['upload_name']), 'Upload changed after confirmation')
        require(sha(job / 'image.img') == checked['raw_sha256'], 'Raw image changed after validation')
        require(checked['target'] == dev['target'] and checked['current_slot'] == dev['slot'], 'Running slot changed; validate again')
        report('validating', job_id=job_id)
        image_fits_slot(job / 'image.img', dev)
        with image_mounts(job / 'image.img', job) as (boot, root):
            validate_payload(boot, root, dev['identity'])
            require(shutil.disk_usage('/boot').free >= sum((boot / name).stat().st_size for name in checked['image']['boot_payload']) + 8 * MIB, 'BOOT free space changed')
            report('backing_up', job_id=job_id)
            # Stop application/database and container writes before retaining data.
            stop_application_services(stopped_services)
            deadline = time.monotonic() + 35
            while True:
                running = []
                for process in Path('/proc').glob('[0-9]*/exe'):
                    try:
                        name = str(process.readlink()).removesuffix(' (deleted)')
                    except OSError:
                        continue
                    if name in ('/usr/bin/one-kvm', '/usr/bin/dockerd'):
                        running.append(name)
                if not running:
                    break
                require(time.monotonic() < deadline, 'Application services are still running; backup aborted')
                time.sleep(0.2)
            backup = job / 'backup'
            backup.mkdir(mode=0o700)
            restore_settings(Path('/'), backup, True)
            data_dir = Path(cmd('uci', '-q', 'get', 'one-kvm.main.data_dir')).resolve(strict=True)
            require(data_dir.is_dir() and Path(dev['shared']) in data_dir.parents, 'One-KVM data directory is outside SHARED')
            db_names = ('one-kvm.db', 'one-kvm.db-wal', 'one-kvm.db-shm')
            db_size = sum((data_dir / name).stat().st_size for name in db_names if (data_dir / name).exists())
            require(shutil.disk_usage(backup).free >= db_size + 16 * MIB, 'Not enough space for the database backup')
            for name in db_names:
                regular_copy(data_dir / name, backup / 'one-kvm' / name)
            atomic(backup / 'uEnv.txt', dev['env'])
            backup_hashes = {str(path.relative_to(backup)): sha(path) for path in backup.rglob('*') if path.is_file()}
            atomic(backup / 'sha256.json', json.dumps(backup_hashes) + '\n')
            require(all(sha(backup / name) == digest for name, digest in backup_hashes.items()), 'Backup readback mismatch')
            os.sync()
            report('writing_rootfs', job_id=job_id, target=dev['target'])
            write_slot(dev, job, boot, root, keep)
            switched = True
        report('rebooting', job_id=job_id, target=dev['target'], sha256=expected,
               preserve_settings=keep, previous_slot=dev['slot'])
        time.sleep(4)
        cmd('/sbin/reboot')
    except Exception as error:
        switched = switched or getattr(error, 'boot_switched', False)
        report('failed', job_id=job_id, error=str(error), boot_switched=switched)
        if stopped_services and not switched:
            for service in stopped_services:
                subprocess.run(['/etc/init.d/' + service, 'start'], stdin=subprocess.DEVNULL,
                               stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    finally:
        os.close(lock_fd)


def prune_boot_payloads(boot, current_env, previous_env):
    """Retain the current and immediately previous boot payload for recovery."""
    keep = set(re.findall(r'/one-kvm-upgrades/([0-9a-f]{24})/', current_env + previous_env))
    directory = Path(boot) / 'one-kvm-upgrades'
    if directory.is_dir() and not directory.is_symlink():
        for payload in directory.iterdir():
            if (re.fullmatch(r'[0-9a-f]{24}', payload.name) and payload.name not in keep
                    and payload.is_dir() and not payload.is_symlink()):
                shutil.rmtree(payload)
        os.sync()


def source_config():
    config = ota.loads(SOURCE.read_bytes())
    require(set(config) == {'schema_version', 'server_url', 'channel'} and config['schema_version'] == 2, 'Invalid release source configuration')
    ota.identifier(config['channel'])
    return config


def source_info():
    info = json.loads(IDENTITY.read_text())
    config = source_config()
    return dict(device=info['board'], version=info.get('version', 'development'), kernel=info['kernel'],
                layout=info['layout'], config=config,
                online_configured=bool(config['server_url']))


def metadata(url, limit=ota.MAX_METADATA):
    with tempfile.TemporaryDirectory(prefix='one-kvm-metadata-') as directory:
        path = Path(directory) / 'metadata'
        ota.fetch(url, path, limit, timeout=20)
        return path.read_bytes()


def release_server():
    config = source_config()
    return config, ota.base_url(config['server_url'], True)


def github_repo(server):
    match = re.fullmatch(r'https://github.com/([A-Za-z0-9][A-Za-z0-9_.-]*)/([A-Za-z0-9][A-Za-z0-9_.-]*)', server)
    return '/'.join(match.groups()) if match else None


def github_releases(config, server, info):
    repo = github_repo(server)
    data = ota.loads(metadata('https://api.github.com/repos/' + repo + '/releases?per_page=20'))
    require(isinstance(data, list), 'Invalid GitHub Releases response')
    result = []
    manifest_name = 'one-kvm-os_' + info['board'] + '.json'
    image_names = {'one-kvm-os_' + info['board'] + suffix for suffix in ('.img', '.img.xz')}
    for release in data:
        tag = release.get('tag_name', '')
        if release.get('draft') or config['channel'] not in tag:
            continue
        try:
            ota.identifier(tag)
        except ValueError:
            continue
        assets = {asset.get('name') for asset in release.get('assets', [])}
        if manifest_name not in assets or not image_names.intersection(assets):
            continue
        result.append(dict(device=info['board'], version=tag, notes=(release.get('body') or '')[:16000],
            manifest=server + '/releases/download/' + tag + '/' + manifest_name))
    return dict(schema_version=2, releases=result)


def release_list():
    config, server = release_server()
    info = json.loads(IDENTITY.read_text())
    if github_repo(server):
        return github_releases(config, server, info)
    query = dict(device=info['board'], arch=info['arch'], platform=info.get('platform', 'amlogic'),
                 layout_version=str(info['layout_version']), layout=info['layout'],
                 adapter=info.get('upgrade_engine', 'one-kvm-ophub-amlogic-v1'), channel=config['channel'])
    for value in query.values():
        ota.identifier(value)
    data = ota.loads(metadata(server + '/api/v1/releases?' + '&'.join(k + '=' + v for k, v in query.items())))
    require(data.get('schema_version') == 2 and isinstance(data.get('releases'), list) and len(data['releases']) <= 100,
            'Invalid release catalog')
    for release in data['releases']:
        require(isinstance(release, dict) and all(str(release.get(k)) == v for k, v in query.items()), 'Catalog returned incompatible release')
        for key in ('version', 'kernel'):
            ota.identifier(release[key])
        ota.release_path(release['manifest'])
    return data


def verified_release(path):
    config, server = release_server()
    repo = github_repo(server)
    if repo:
        prefix = server + '/releases/download/'
        require(path.startswith(prefix), 'Selected release is outside configured GitHub repository')
        relative = path[len(prefix):].split('/')
        require(len(relative) == 2 and relative[1] == 'one-kvm-os_' + json.loads(IDENTITY.read_text())['board'] + '.json', 'Invalid GitHub manifest selection')
        ota.identifier(relative[0])
        raw = metadata(path)
    else:
        ota.release_path(path)
        require(path.endswith('/manifest.json'), 'Expected a release manifest')
        raw = metadata(server + path)
    release = ota.validate(ota.loads(raw), json.loads(IDENTITY.read_text()))
    require(release['channel'] == config['channel'], 'Release channel differs from configured source')
    prefix = '/releases/' + '/'.join((release['device'], release['channel'], release['version']))
    expected_artifact = prefix + '/one-kvm-os_' + release['device'] + '.' + release['artifact']['format']
    require(release['artifact']['path'] == expected_artifact, 'Artifact path differs from identity')
    require(release['version'] == relative[0] if repo else path == prefix + '/manifest.json', 'Release version/path differs from identity')
    return server, release


def clean_unconfirmed():
    if JOBS.is_dir():
        for previous in JOBS.iterdir():
            if (re.fullmatch(r'[0-9a-f]{24}', previous.name) and previous.is_dir()
                    and not previous.is_symlink() and (previous / 'checked.json').is_file()
                    and not (previous / 'worker.log').exists()):
                shutil.rmtree(previous)
    INBOX.unlink(missing_ok=True)
    INBOX.with_suffix('.part').unlink(missing_ok=True)


def upload_space(size, image_format, dev):
    require(image_format in ('img', 'img.xz'), '请选择 .img 或 .img.xz 固件')
    # The pinned uhttpd stores each request's Content-Length in a signed int.
    require(0 < size < 2**31, '单次上传需小于 2 GiB，大镜像请上传 .img.xz')
    if image_format == 'img.xz':
        require(0 < size <= 1536 * MIB, '压缩固件大小超出限制')
        reserve = dev['slot_size'] + 576 * MIB
    else:
        require(valid_image_size(size) and size <= 260 * MIB + dev['slot_size'],
                'IMG 大小与设备系统分区不匹配')
        reserve = 256 * MIB
    require(shutil.disk_usage(INBOX.parent).free >= size + reserve,
            '共享分区空间不足，请释放空间后重试')


def begin_check(lock):
    report('checking')
    with (STATE.parent / 'check.log').open('ab') as log:
        subprocess.Popen([sys.executable, str(Path(__file__).resolve()), 'checker', str(lock.fileno())],
                         pass_fds=(lock.fileno(),), stdin=subprocess.DEVNULL, stdout=log, stderr=log,
                         start_new_session=True)
    return {'phase': 'checking'}


def receive_upload(stream, env):
    require(env.get('REQUEST_METHOD') == 'POST', '仅支持 POST 上传')
    authorization = re.fullmatch(r'Bearer ([0-9a-f]{32})', env.get('HTTP_AUTHORIZATION', ''))
    require(authorization is not None, '请先登录')
    session = authorization[1]
    # Apply the same session and fixed-path ACL checks as LuCI cgi-upload.
    for scope, obj in (('cgi-io', 'upload'), ('file', str(INBOX))):
        access = json.loads(cmd('ubus', 'call', 'session', 'access', json.dumps(
            dict(ubus_rpc_session=session, scope=scope, object=obj, function='write'))))
        require(access.get('access') is True, '没有固件上传权限或登录已过期')
    content_type = env.get('CONTENT_TYPE')
    require(content_type in ('application/octet-stream', 'application/x-xz'), '无效的上传类型')
    image_format = 'img.xz' if content_type == 'application/x-xz' else 'img'
    length = env.get('CONTENT_LENGTH', '')
    require(re.fullmatch(r'[0-9]{1,11}', length) is not None, '无效的固件大小')
    size = int(length)
    with locked() as lock:
        dev = device()
        INBOX.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
        clean_unconfirmed()
        report('idle')
        upload_space(size, image_format, dev)
        temporary = INBOX.with_suffix('.part')
        temporary.unlink(missing_ok=True)
        try:
            progress = progress_reporter('uploading')
            completed = 0
            with temporary.open('xb') as target:
                os.chmod(temporary, 0o600)
                while completed < size:
                    chunk = stream.read(min(MIB, size - completed))
                    require(chunk, '上传中断，请重试')
                    target.write(chunk)
                    completed += len(chunk)
                    progress(completed, size)
                target.flush()
                os.fsync(target.fileno())
            os.replace(temporary, INBOX)
            return begin_check(lock)
        except Exception as error:
            INBOX.unlink(missing_ok=True)
            report('failed', error=str(error), boot_switched=False)
            raise
        finally:
            temporary.unlink(missing_ok=True)


def download_release(path):
    temporary = INBOX.with_suffix('.part')
    try:
        device()
        server, release = verified_release(path)
        size = release['artifact']['size_bytes']
        unpack_space = device()['slot_size'] + 576 * MIB if release['artifact']['format'] == 'img.xz' else 256 * MIB
        require(shutil.disk_usage(INBOX.parent).free >= size + unpack_space,
                'Not enough SHARED space for download and validation')
        temporary.unlink(missing_ok=True)
        last_report = [0.0]
        def progress(count):
            now = time.monotonic()
            if now - last_report[0] >= 1 or count == size:
                report('downloading', version=release['version'], downloaded_bytes=count, total_bytes=size)
                last_report[0] = now
        artifact_name = Path(release['artifact']['path']).name
        artifact_url = path.rsplit('/', 1)[0] + '/' + artifact_name if github_repo(server) else server + release['artifact']['path']
        actual = ota.fetch(artifact_url, temporary, size, progress)
        require(actual == size and sha(temporary) == release['artifact']['sha256'], 'Downloaded firmware size/SHA256 mismatch')
        os.replace(temporary, INBOX)
        report('checking', version=release['version'])
        check_upload(release)
    except Exception as error:
        INBOX.unlink(missing_ok=True)
        report('failed', error=str(error), boot_switched=False)
    finally:
        temporary.unlink(missing_ok=True)


def main(args):
    action = args[0] if args else 'status'
    if action == 'receive':
        require(len(args) == 1, 'Unexpected arguments')
        return receive_upload(sys.stdin.buffer, os.environ)
    if action == 'status':
        return json.loads(STATE.read_text()) if STATE.is_file() else {'phase': 'idle'}
    if action == 'info':
        return source_info()
    if action == 'releases':
        device()
        return release_list()
    if action == 'boot-check':
        if STATE.is_file():
            state = json.loads(STATE.read_text())
            if state.get('phase') == 'rebooting':
                dev = device()
                if dev['root'] == state.get('target'):
                    state.update(phase='booted', running_slot=dev['slot'])
                    atomic(STATE, json.dumps(state) + '\n')
                    # Release staging space after successful root switch;
                    # keep settings backups and the two uEnv copies for recovery.
                    job = job_path(state['job_id'])
                    for name in ('image.img', 'upload.img.xz'):
                        (job / name).unlink(missing_ok=True)
                    prune_boot_payloads(dev['boot'], dev['env'], (job / 'uEnv.before.txt').read_text())
        return {'ok': True}
    if action == 'worker':
        require(len(args) == 5 and args[3] in ('0', '1'), 'Invalid worker arguments')
        inherited_lock(int(args[4]))
        worker(args[1], args[2], args[3] == '1', int(args[4]))
        return {'ok': True}
    if action == 'checker':
        require(len(args) == 2, 'Invalid checker arguments')
        inherited_lock(int(args[1]))
        try:
            check_upload()
        except Exception as error:
            report('failed', error=str(error), boot_switched=False)
        finally:
            os.close(int(args[1]))
        return {'ok': True}
    if action == 'downloader':
        require(len(args) == 3, 'Invalid download worker arguments')
        inherited_lock(int(args[2]))
        try:
            download_release(args[1])
        finally:
            os.close(int(args[2]))
        return {'ok': True}
    with locked() as lock:
        dev = device()
        if action == 'source':
            require(len(args) == 3, 'Invalid source settings')
            url = ota.base_url(args[1], True)
            ota.identifier(args[2])
            atomic(SOURCE, json.dumps(dict(schema_version=2, server_url=url, channel=args[2])) + '\n')
            return source_info()
        if action == 'download':
            require(len(args) == 2, 'Invalid download selection')
            config, server = release_server()
            if github_repo(server):
                require(args[1].startswith(server + '/releases/download/'), 'Release is outside configured GitHub repository')
            else:
                ota.release_path(args[1])
            STATE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
            clean_unconfirmed()
            report('downloading', downloaded_bytes=0)
            with (STATE.parent / 'download.log').open('ab') as log:
                subprocess.Popen([sys.executable, str(Path(__file__).resolve()), 'downloader', args[1], str(lock.fileno())],
                                 pass_fds=(lock.fileno(),), stdin=subprocess.DEVNULL, stdout=log, stderr=log, start_new_session=True)
            return {'phase': 'downloading'}
        if action == 'cancel':
            require(len(args) == 1, 'Unexpected arguments')
            clean_unconfirmed()
            return report('idle')
        if action == 'prepare':
            require(len(args) == 3 and re.fullmatch(r'[0-9]{1,11}', args[1]), 'Invalid upload size')
            size = int(args[1])
            STATE.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
            # A new upload replaces earlier unconfirmed uploads. Retain every
            # job that reached execution, including failed-job recovery files.
            clean_unconfirmed()
            report('idle')
            upload_space(size, args[2], dev)
            return {'path': str(INBOX)}
        if action == 'check':
            require(len(args) == 1, 'Unexpected arguments')
            return begin_check(lock)
        if action == 'start':
            require(len(args) == 4 and args[3] in ('0', '1') and re.fullmatch(r'[0-9a-f]{64}', args[2]), 'Invalid confirmation')
            job = job_path(args[1])
            checked = json.loads((job / 'checked.json').read_text())
            require(checked['sha256'] == args[2], 'Confirmed hash differs')
            report('starting', job_id=args[1])
            log = (job / 'worker.log').open('ab')
            subprocess.Popen([sys.executable, str(Path(__file__).resolve()), 'worker', *args[1:], str(lock.fileno())],
                             pass_fds=(lock.fileno(),), stdin=subprocess.DEVNULL, stdout=log, stderr=log, start_new_session=True)
            # The inherited fd retains the flock after this request exits.
            return {'phase': 'starting', 'job_id': args[1]}
        raise ValueError('Unknown action')


if __name__ == '__main__':
    os.umask(0o077)
    try:
        print(json.dumps(main(sys.argv[1:]), ensure_ascii=False))
    except Exception as error:
        print(json.dumps({'error': str(error)}, ensure_ascii=False))
        sys.exit(1)
