#!/usr/bin/python3 -I -S
"""Finite ALSA ACL owner. No RPC, application paths, groups, or device IO.

Each start is an independently journaled device incarnation. An unplug, inode
replacement, policy/ACL drift or owner death retires the dependent runtime.
HDMI connector changes that retain the ALSA nodes do not change this lease.
Re-admission after node replacement requires runtime retirement and a fresh
root appointment; existing uncertain journals are never overwritten.
"""
import errno
import fcntl
import hashlib
import json
import os
from pathlib import Path
import re
import signal
import socket
import stat
import struct
import sys
import time

CONFIG = Path('/etc/archipelago-c01/audio-devices.json')
STATE = Path('/var/lib/archipelago-c01/audio-devices')
GROUP = Path('/sys/fs/cgroup/user.slice/user-1000.slice/user@1000.service')
ATTRIBUTE = 'system.posix_acl_access'
UNDEFINED = 0xffffffff
USER_OBJ, USER, GROUP_OBJ, ACL_GROUP, MASK, OTHER = 1, 2, 4, 8, 16, 32


def require(value, reason):
    if not value:
        raise ValueError(reason)


def exact(pairs):
    result = {}
    for key, value in pairs:
        require(key not in result, 'duplicate audio appointment key')
        result[key] = value
    return result


def protected(path, directory=False):
    path = Path(path)
    require(path.is_absolute() and '..' not in path.parts, 'canonical audio custody path')
    for item in (*reversed(path.parents), path):
        meta = item.lstat()
        require(meta.st_uid == 0 and not meta.st_mode & 0o022 and not stat.S_ISLNK(meta.st_mode),
                'mutable audio custody')
        require(stat.S_ISDIR(meta.st_mode) if item != path or directory else stat.S_ISREG(meta.st_mode),
                'audio custody type')
    return path


def encode(entries):
    return struct.pack('<I', 2) + b''.join(struct.pack('<HHI', *row) for row in entries)


def decode(raw, mode):
    if raw is None:
        return [(USER_OBJ, (mode >> 6) & 7, UNDEFINED),
                (GROUP_OBJ, (mode >> 3) & 7, UNDEFINED), (OTHER, mode & 7, UNDEFINED)]
    require(len(raw) >= 28 and len(raw) <= 4096 and (len(raw) - 4) % 8 == 0 and
            struct.unpack('<I', raw[:4])[0] == 2, 'ALSA ACL encoding')
    rows = [struct.unpack('<HHI', raw[i:i+8]) for i in range(4, len(raw), 8)]
    require(rows == sorted(rows, key=lambda row: (row[0], row[2])), 'noncanonical ALSA ACL')
    require(len({(tag, uid) for tag, _, uid in rows}) == len(rows), 'duplicate ALSA ACL entry')
    for tag, perms, uid in rows:
        require(tag in (USER_OBJ, USER, GROUP_OBJ, ACL_GROUP, MASK, OTHER) and 0 <= perms <= 7 and
                ((uid != UNDEFINED) if tag in (USER, ACL_GROUP) else uid == UNDEFINED), 'ALSA ACL entry')
    require(all(sum(t == tag for t, _, _ in rows) == 1 for tag in (USER_OBJ, GROUP_OBJ, OTHER)), 'ALSA ACL base entries')
    require(sum(t == MASK for t, _, _ in rows) == 1, 'extended ALSA ACL mask required')
    return rows


def permissions(rows, owner, group, uid, groups):
    by_tag = {tag: perms for tag, perms, _ in rows if tag not in (USER, ACL_GROUP)}
    mask = by_tag.get(MASK, 7)
    if uid == owner:
        return by_tag[USER_OBJ]
    for tag, perms, who in rows:
        if tag == USER and who == uid:
            return perms & mask
    matches = ([by_tag[GROUP_OBJ]] if group in groups else [])
    matches += [perms for tag, perms, who in rows if tag == ACL_GROUP and who in groups]
    if matches:
        result = 0
        for value in matches:
            result |= value
        return result & mask
    return by_tag[OTHER]


def grant(raw, mode, owner, group, prior_groups):
    """Preserve prior UID1000 access and EVERY other entry's effective rights.

    Preserve the preimage verbatim in the journal. When expanding a mask, first
    clamp its existing class entries to their old effective permissions; never
    activate previously masked permissions belonging to somebody else.
    """
    rows = decode(raw, mode)
    allowed = permissions(rows, owner, group, 1000, prior_groups)
    require(allowed in (4, 6) and owner != 1000, 'no prior finite ALSA read/write access')
    mask = next((p for t, p, _ in rows if t == MASK), 7)
    result = []
    for tag, perms, who in rows:
        if tag == MASK or (tag == USER and who == 1000):
            continue
        if tag in (USER, GROUP_OBJ, ACL_GROUP):
            perms &= mask
        result.append((tag, perms, who))
    result += [(USER, allowed, 1000), (MASK, (mask if raw is not None else (mode >> 3) & 7) | allowed, UNDEFINED)]
    return encode(sorted(result, key=lambda row: (row[0], row[2])))


def identity(meta):
    return [meta.st_dev, meta.st_ino, meta.st_rdev, meta.st_uid, meta.st_gid]


def acl(path):
    try:
        return os.getxattr(path, ATTRIBUTE)
    except OSError as error:
        if error.errno == errno.ENODATA:
            return None
        raise


def restore(path, raw):
    if raw is None:
        try:
            os.removexattr(path, ATTRIBUTE)
        except OSError as error:
            if error.errno != errno.ENODATA:
                raise
    else:
        os.setxattr(path, ATTRIBUTE, raw)


def read_policy():
    raw = protected(CONFIG).read_bytes()
    require(len(raw) <= 65536, 'audio policy bound')
    value = json.loads(raw, object_pairs_hook=exact)
    require(set(value) == {'version', 'uid', 'prior_groups', 'devices'} and type(value['version']) is int and value['version'] == 1 and
            value['uid'] == 1000 and value['prior_groups'] == [20, 27, 29, 1000], 'audio principal appointment')
    require(type(value['devices']) is list and 1 <= len(value['devices']) <= 64, 'finite audio device set')
    names = set()
    for item in value['devices']:
        require(set(item) == {'name', 'sysfs', 'rdev', 'uid', 'gid', 'mode', 'acl_sha256'}, 'audio device fields')
        name = item['name']
        require(type(name) is str and re.fullmatch(r'(?:controlC[0-9]+|hwC[0-9]+D[0-9]+|pcmC[0-9]+D[0-9]+[pc]|seq|timer)', name), 'ALSA node name')
        require(name not in names, 'duplicate ALSA device')
        names.add(name)
        require(type(item['sysfs']) is str and item['sysfs'].startswith('/sys/devices/') and
                '..' not in Path(item['sysfs']).parts and Path(item['sysfs']).name == name, 'kernel ALSA locator')
        require(type(item['rdev']) is int and os.major(item['rdev']) == 116 and item['uid'] == 0 and
                item['gid'] == 29 and type(item['mode']) is int and item['mode'] & ~0o777 == 0,
                'original ALSA ownership/device appointment')
        require(item['acl_sha256'] is None or (type(item['acl_sha256']) is str and
                re.fullmatch('[0-9a-f]{64}', item['acl_sha256'])), 'ACL preimage digest')
    return value, hashlib.sha256(raw).hexdigest()


def runtime_retired():
    try:
        values = dict(line.split() for line in (GROUP / 'cgroup.events').read_text().splitlines())
    except FileNotFoundError:
        require(not GROUP.exists(), 'runtime cgroup events unavailable')
        return
    require(values.get('populated') == '0', 'retire complete runtime hierarchy before audio reconciliation')


def device_path(item):
    return Path('/dev/snd') / item['name']


def detached(record):
    """Retirement proof for an unplugged inode, never permission on its replacement.

    The runtime has already retired. Preserve the journal and refuse to call an
    inode detached if it remains under another name in the fixed sound tree.
    This also prevents an administrator rename from masquerading as unplug.
    """
    path = device_path(record['device'])
    protected(path.parent, directory=True)
    try:
        current = path.lstat()
        if [current.st_dev, current.st_ino] == record['identity'][:2]:
            return False
    except FileNotFoundError:
        current = None
    entries = list(path.parent.iterdir())
    require(len(entries) <= 128, 'sound-tree retirement inventory bound')
    for entry in entries:
        meta = entry.lstat()
        require([meta.st_dev, meta.st_ino] != record['identity'][:2],
                'old ALSA incarnation remains under another name')
    record['phase'] = 'detached'
    record['replacement_observed'] = identity(current) if current is not None else None
    return True


class Device:
    def __init__(self, item):
        self.item = item
        self.path = device_path(item)
        protected(self.path.parent, directory=True)
        self.fd = os.open(self.path, os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC)
        self.handle = Path('/proc/self/fd') / str(self.fd)
        try:
            meta = os.fstat(self.fd)
            require(stat.S_ISCHR(meta.st_mode) and meta.st_rdev == item['rdev'] and
                    (meta.st_uid, meta.st_gid) == (item['uid'], item['gid']), 'ALSA kernel object differs')
            self.original_identity = identity(meta)
            self.check_kernel()
        except BaseException:
            self.close()
            raise

    def check_kernel(self):
        meta = self.path.lstat()
        require(stat.S_ISCHR(meta.st_mode) and identity(meta) == self.original_identity and
                identity(os.fstat(self.fd)) == self.original_identity and meta.st_nlink == 1, 'ALSA incarnation replaced/unplugged')
        number = f'{os.major(meta.st_rdev)}:{os.minor(meta.st_rdev)}'
        kernel = Path('/sys/dev/char') / number
        require(str(kernel.resolve(strict=True)) == self.item['sysfs'] and
                (kernel / 'subsystem').resolve(strict=True) == Path('/sys/class/sound'), 'not the appointed kernel sound device')
        protected(Path(self.item['sysfs']), directory=True)

    def close(self):
        if self.fd is not None:
            os.close(self.fd)
            self.fd = None


def write_journal(path, value):
    raw = json.dumps(value, sort_keys=True, separators=(',', ':')).encode()
    temporary = path.with_suffix('.new')
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW, 0o600)
    try:
        with os.fdopen(fd, 'wb') as stream:
            stream.write(raw)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
        parent = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
        try:
            os.fsync(parent)
        finally:
            os.close(parent)
    except BaseException:
        # Uncertain durable preimage is retained, never removed as cleanup.
        raise


def reconcile():
    runtime_retired()
    protected(STATE, directory=True)
    for path in sorted(STATE.glob('*.json')):
        record = json.loads(protected(path).read_bytes(), object_pairs_hook=exact)
        if record['phase'] in ('revoked', 'expired-boot', 'detached'):
            continue
        if record['boot_id'] != Path('/proc/sys/kernel/random/boot_id').read_text().strip():
            # Prior-boot device inodes/FDs cannot survive. Never restore an old
            # preimage onto a new boot's coincidentally reused devtmpfs inode.
            record['phase'] = 'expired-boot'
            write_journal(path, record)
            continue
        if detached(record):
            write_journal(path, record)
            continue
        device = None
        try:
            device = Device(record['device'])
            require(device.original_identity == record['identity'], 'unresolved old ALSA incarnation; preserve journal')
            before = bytes.fromhex(record['before']) if record['before'] is not None else None
            after = bytes.fromhex(record['after'])
            current = acl(device.handle)
            require(current in (before, after), 'ALSA ACL changed independently; no blind rollback')
            if current == after:
                restore(device.handle, before)
                os.chmod(device.handle, record['mode'])
            require(acl(device.handle) == before and stat.S_IMODE(os.fstat(device.fd).st_mode) == record['mode'], 'ALSA restoration differs')
            record['phase'] = 'revoked'
            write_journal(path, record)
        finally:
            if device is not None:
                device.close()


def notify(text):
    address = os.environ.get('NOTIFY_SOCKET')
    require(address is not None, 'audio owner requires service lifetime')
    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as channel:
        channel.connect('\0' + address[1:] if address.startswith('@') else address)
        channel.send(text.encode())


def run():
    runtime_retired()
    reconcile()
    value, digest = read_policy()
    devices = []
    stop = False
    def stopping(*_):
        nonlocal stop
        stop = True
    signal.signal(signal.SIGTERM, stopping)
    signal.signal(signal.SIGINT, stopping)
    try:
        # Capture and validate ALL preimages before the first permission change.
        for item in value['devices']:
            device = Device(item)
            devices.append((device, None))
            meta = os.fstat(device.fd)
            before = acl(device.handle)
            require(stat.S_IMODE(meta.st_mode) == item['mode'] and
                    (hashlib.sha256(before).hexdigest() if before is not None else None) == item['acl_sha256'], 'ALSA preimage differs')
            after = grant(before, item['mode'], item['uid'], item['gid'], value['prior_groups'])
            devices[-1] = (device, dict(version=1, policy_sha256=digest, device=item,
                boot_id=Path('/proc/sys/kernel/random/boot_id').read_text().strip(),
                identity=device.original_identity, mode=item['mode'], before=before.hex() if before is not None else None,
                after=after.hex(), phase='intent'))
        for device, record in devices:
            device.check_kernel()
            name = hashlib.sha256(os.urandom(32)).hexdigest() + '.json'
            write_journal(STATE / name, record)
            os.setxattr(device.handle, ATTRIBUTE, bytes.fromhex(record['after']))
            require(acl(device.handle).hex() == record['after'], 'ALSA grant readback differs')
            record['phase'] = 'granted'
            write_journal(STATE / name, record)
        notify('READY=1')
        while not stop:
            require(hashlib.sha256(protected(CONFIG).read_bytes()).hexdigest() == digest, 'audio policy changed')
            for device, record in devices:
                device.check_kernel()
                require(acl(device.handle) == bytes.fromhex(record['after']), 'ALSA grant drift')
            notify('WATCHDOG=1')
            time.sleep(0.25)
    finally:
        for device, _ in devices:
            device.close()
    # Dependent user@1000 must already be stopped by systemd ordering. Failure
    # leaves journaled grants for reconciliation, never claims FD revocation.
    reconcile()


def main():
    require(os.getresuid() == (0, 0, 0) and os.getresgid() == (0, 0, 0), 'root audio owner required')
    require(len(sys.argv) == 2 and sys.argv[1] in ('run', 'reconcile'), 'fixed audio operation required')
    protected(STATE, directory=True)
    lock = os.open(STATE / 'owner.lock', os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600)
    try:
        require(os.fstat(lock).st_uid == 0 and os.fstat(lock).st_nlink == 1 and
                stat.S_IMODE(os.fstat(lock).st_mode) == 0o600, 'audio lock custody')
        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        run() if sys.argv[1] == 'run' else reconcile()
    finally:
        os.close(lock)


if __name__ == '__main__':
    main()