strop-remote 0.19.0

"""Fixed owned save helper. Inputs are framed data, never executable text."""
import errno
import fcntl
import hashlib
import json
import os
import secrets
import signal
import stat
import struct
import sys
import unicodedata

VERSION = 1
MAX_BYTES = 256 * 1024 * 1024
MAX_HEADER = 16 * 1024
MAX_ATTRIBUTES = 64 * 1024
MAX_ENTRIES = 100_000
CHUNK = 64 * 1024
LOCK_PREFIX = b'.strop-lock-'
STAGE_PREFIX = b'.strop-save-'


class Refusal(Exception):
    def __init__(self, kind, detail):
        self.kind = kind
        self.detail = detail
        super().__init__(detail)


def terminated(signum, frame):
    raise Refusal('cancelled', 'termination observed')


for termination_signal in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
    signal.signal(termination_signal, terminated)


def reserved(component):
    normalized = unicodedata.normalize('NFKC', os.fsdecode(component)).casefold()
    return normalized.startswith(('.strop-lock-', '.strop-save-'))


def exact_name(directory, name):
    # Exact directory-entry spelling gives one stable basename key even on a
    # filesystem which also accepts case/normalization aliases.
    with os.scandir(directory) as entries:
        for count, entry in enumerate(entries):
            if count >= MAX_ENTRIES:
                raise Refusal('unsupported', 'directory spelling check exceeds 100000 entries')
            stored = os.fsencode(entry.name)
            if stored == name:
                if reserved(stored):
                    raise Refusal('invalid_path', 'remote-save control paths are reserved')
                return
    raise Refusal('invalid_path', 'use exact stored path spelling; reopen through the directory browser')


def identity(info):
    return info.st_dev, info.st_ino


def file_info(info):
    return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns,
            info.st_ctime_ns, info.st_mode, info.st_uid, info.st_gid, info.st_nlink)


def attributes(descriptor):
    if not all(hasattr(os, name) for name in ('listxattr', 'getxattr', 'setxattr', 'removexattr')):
        raise Refusal('unsupported', 'extended-attribute preservation is unavailable')
    result = {}
    total = 0
    try:
        names = os.listxattr(descriptor)
    except OSError as error:
        if error.errno in (errno.ENOTSUP, errno.EOPNOTSUPP):
            return result
        raise
    for name in names:
        name = os.fsencode(name)
        value = os.getxattr(descriptor, name)
        total += len(name) + len(value)
        if total > MAX_ATTRIBUTES:
            raise Refusal('metadata', 'attribute inventory exceeds 64 KiB')
        result[name] = value
    return result


def attribute_digest(values):
    digest = hashlib.sha256()
    for name, value in sorted(values.items()):
        digest.update(struct.pack('>I', len(name)))
        digest.update(name)
        digest.update(struct.pack('>I', len(value)))
        digest.update(value)
    return list(digest.digest())


def validate_file(info):
    if not stat.S_ISREG(info.st_mode) or info.st_nlink != 1:
        raise Refusal('invalid_path', 'only regular single-link files may be edited')
    if info.st_uid != os.geteuid():
        raise Refusal('permission', 'remote saving requires a file owned by the authenticated user')
    if info.st_size > MAX_BYTES:
        raise Refusal('too_large', 'remote file exceeds 256 MiB')


def snapshot(descriptor):
    before = os.fstat(descriptor)
    validate_file(before)
    values = attributes(descriptor)
    digest = hashlib.sha256()
    os.lseek(descriptor, 0, os.SEEK_SET)
    length = 0
    while True:
        chunk = os.read(descriptor, CHUNK)
        if not chunk:
            break
        length += len(chunk)
        if length > MAX_BYTES:
            raise Refusal('too_large', 'remote file grew beyond 256 MiB')
        digest.update(chunk)
    after = os.fstat(descriptor)
    if file_info(before) != file_info(after) or values != attributes(descriptor):
        raise Refusal('conflict', 'file changed while its baseline was read')
    return {
        'device': after.st_dev, 'inode': after.st_ino, 'size': after.st_size,
        'mtime_ns': after.st_mtime_ns, 'ctime_ns': after.st_ctime_ns,
        'mode': stat.S_IMODE(after.st_mode), 'uid': after.st_uid, 'gid': after.st_gid,
        'content': list(digest.digest()), 'attributes': attribute_digest(values),
    }, values


def preserves(current, expected):
    return all(current[key] == expected[key] for key in
               ('device', 'mode', 'uid', 'gid', 'mtime_ns', 'attributes'))


class Transaction:
    def __init__(self, path):
        self.descriptors = []
        self.bindings = []
        self.parent = None
        self.target = None
        self.lock = None
        self.stage_directory = None
        self.stage_name = None
        self.stage_identity = None
        self.stage = None
        self.commit_started = False
        self.path = path

    def own(self, descriptor):
        self.descriptors.append(descriptor)
        return descriptor

    def open(self):
        if not self.path.startswith(b'/') or b'\0' in self.path:
            raise Refusal('invalid_path', 'an absolute native path is required')
        components = [part for part in self.path.split(b'/') if part]
        if not components or any(part in (b'.', b'..') or reserved(part) for part in components):
            raise Refusal('invalid_path', 'canonical non-control path components are required')
        flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
        parent = self.own(os.open(b'/', flags))
        for component in components[:-1]:
            exact_name(parent, component)
            child = self.own(os.open(component, flags, dir_fd=parent))
            self.bindings.append((parent, component, child))
            parent = child
        self.parent = parent
        self.name = components[-1]
        exact_name(parent, self.name)
        self.lock_name = LOCK_PREFIX + hashlib.sha256(self.name).hexdigest().encode('ascii')
        self.lock = self.own(os.open(self.lock_name, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600, dir_fd=parent))
        info = os.fstat(self.lock)
        if (not stat.S_ISREG(info.st_mode) or info.st_nlink != 1 or
                info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o600):
            raise Refusal('permission', 'save lock is not a private owned single-link file')
        try:
            fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise Refusal('busy', 'another remote-save participant holds this file')
        self.target = self.own(os.open(self.name, os.O_RDWR | os.O_NOFOLLOW, dir_fd=parent))
        validate_file(os.fstat(self.target))
        self.revalidate()

    def revalidate(self, target=None):
        for parent, name, child in self.bindings:
            current = os.stat(name, dir_fd=parent, follow_symlinks=False)
            if not stat.S_ISDIR(current.st_mode) or identity(current) != identity(os.fstat(child)):
                raise Refusal('conflict', 'a parent directory changed identity')
        lock = os.stat(self.lock_name, dir_fd=self.parent, follow_symlinks=False)
        if identity(lock) != identity(os.fstat(self.lock)) or lock.st_nlink != 1:
            raise Refusal('conflict', 'save lock changed identity')
        current = os.stat(self.name, dir_fd=self.parent, follow_symlinks=False)
        descriptor = self.target if target is None else target
        validate_file(current)
        if file_info(current) != file_info(os.fstat(descriptor)):
            raise Refusal('conflict', 'destination changed identity or metadata')

    def create_stage(self):
        self.stage_name = STAGE_PREFIX + secrets.token_hex(16).encode('ascii')
        os.mkdir(self.stage_name, 0o700, dir_fd=self.parent)
        self.stage_identity = identity(os.stat(self.stage_name, dir_fd=self.parent, follow_symlinks=False))
        self.stage_directory = self.own(os.open(self.stage_name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=self.parent))
        info = os.fstat(self.stage_directory)
        if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700:
            raise Refusal('permission', 'transaction directory is not private')
        self.stage = self.own(os.open(b'contents', os.O_CREAT | os.O_EXCL | os.O_RDWR | os.O_NOFOLLOW, 0o600, dir_fd=self.stage_directory))

    def restore_metadata(self, expected, values):
        descriptor = self.stage
        os.fchown(descriptor, expected['uid'], expected['gid'])
        os.fchmod(descriptor, expected['mode'])
        inherited = attributes(descriptor)
        for name in inherited:
            if name not in values:
                os.removexattr(descriptor, name)
        for name, value in values.items():
            if inherited.get(name) != value:
                os.setxattr(descriptor, name, value)
        os.utime(descriptor, ns=(os.fstat(self.target).st_atime_ns, expected['mtime_ns']))
        actual = os.fstat(descriptor)
        if (stat.S_IMODE(actual.st_mode) != expected['mode'] or actual.st_uid != expected['uid'] or
                actual.st_gid != expected['gid'] or actual.st_mtime_ns != expected['mtime_ns'] or
                attributes(descriptor) != values):
            raise Refusal('metadata', 'restored mode, ownership, timestamp or attributes differ')

    def cleanup(self):
        # Cleanup is bounded by the supervisor's TERM grace; SIGKILL may leave
        # this private directory, never a partially overwritten destination.
        signal.pthread_sigmask(signal.SIG_BLOCK, {signal.SIGTERM, signal.SIGINT, signal.SIGHUP})
        try:
            if self.stage_identity is not None:
                if self.stage_directory is not None:
                    try:
                        os.unlink(b'contents', dir_fd=self.stage_directory)
                    except FileNotFoundError:
                        pass  # atomic rename already removed this owned entry
                current = os.stat(self.stage_name, dir_fd=self.parent, follow_symlinks=False)
                if identity(current) != self.stage_identity:
                    raise Refusal('conflict', 'transaction directory changed identity during cleanup')
                os.rmdir(self.stage_name, dir_fd=self.parent)
                os.fsync(self.parent)
        finally:
            for descriptor in reversed(self.descriptors):
                os.close(descriptor)


def request():
    header = sys.stdin.buffer.readline(MAX_HEADER + 1)
    if len(header) > MAX_HEADER or not header.endswith(b'\n'):
        raise Refusal('protocol', 'missing or oversized request header')
    value = json.loads(header)
    if value.get('version') != VERSION or value.get('operation') not in ('edit', 'save', 'verify'):
        raise Refusal('protocol', 'unsupported request version or operation')
    length = value.get('length')
    digest = value.get('digest')
    if type(length) is not int or not 0 <= length <= MAX_BYTES:
        raise Refusal('too_large', 'request length exceeds the supported bound')
    if not isinstance(digest, list) or len(digest) != 32 or any(type(byte) is not int or not 0 <= byte <= 255 for byte in digest):
        raise Refusal('protocol', 'invalid content digest')
    return value


def perform(transaction, requested):
    transaction.open()
    before, values = snapshot(transaction.target)
    transaction.revalidate()
    operation = requested['operation']
    length, wanted = requested['length'], requested['digest']
    if operation == 'edit':
        if before['size'] != length or before['content'] != wanted:
            raise Refusal('conflict', 'displayed snapshot differs from the remote file; refresh before editing')
        transaction.create_stage()
        transaction.restore_metadata(before, values)
        os.fsync(transaction.stage)
        os.fsync(transaction.stage_directory)
        os.fsync(transaction.parent)
        return {'status': 'ready', 'stamp': before}
    expected = requested.get('before')
    if not isinstance(expected, dict):
        raise Refusal('protocol', 'missing expected file version')
    if operation == 'verify':
        if before == expected:
            return {'status': 'unchanged', 'stamp': before}
        if before['size'] != length or before['content'] != wanted or not preserves(before, expected):
            raise Refusal('conflict', 'remote state matches neither the original nor the intended snapshot')
        os.fsync(transaction.target)
        os.fsync(transaction.parent)
        checked, _ = snapshot(transaction.target)
        transaction.revalidate()
        if checked != before:
            raise Refusal('conflict', 'remote file changed during verification')
        return {'status': 'written', 'stamp': checked}
    if before != expected:
        raise Refusal('conflict', 'remote content or metadata changed since editing was admitted')
    transaction.create_stage()
    digest = hashlib.sha256()
    remaining = length
    while remaining:
        chunk = sys.stdin.buffer.read(min(CHUNK, remaining))
        if not chunk:
            raise Refusal('protocol', 'upload ended before its declared length')
        digest.update(chunk)
        view = memoryview(chunk)
        while view:
            written = os.write(transaction.stage, view)
            if written <= 0:
                raise Refusal('io', 'stage write made no progress')
            view = view[written:]
        remaining -= len(chunk)
    if list(digest.digest()) != wanted:
        raise Refusal('protocol', 'upload content digest differs')
    transaction.restore_metadata(before, values)
    os.fsync(transaction.stage)
    os.fsync(transaction.stage_directory)
    current, _ = snapshot(transaction.target)
    transaction.revalidate()
    if current != before:
        raise Refusal('conflict', 'remote file changed during stage preparation')
    # Set before the syscall: even a signal between rename and its return must
    # not be misclassified as a proven precommit failure.
    transaction.commit_started = True
    os.replace(b'contents', transaction.name, src_dir_fd=transaction.stage_directory, dst_dir_fd=transaction.parent)
    os.fsync(transaction.stage_directory)
    os.fsync(transaction.parent)
    committed, _ = snapshot(transaction.stage)
    transaction.revalidate(transaction.stage)
    if committed['size'] != length or committed['content'] != wanted or not preserves(committed, before):
        raise Refusal('metadata', 'committed file differs from intended bytes or preserved metadata')
    return {'status': 'written', 'stamp': committed}


def main():
    transaction = None
    result = None
    error = None
    try:
        required = ('O_NOFOLLOW', 'O_DIRECTORY', 'scandir', 'fchown', 'fsync', 'replace')
        if not all(hasattr(os, name) for name in required) or not hasattr(signal, 'pthread_sigmask'):
            raise Refusal('unsupported', 'descriptor and cancellation capabilities are unavailable')
        requested = request()
        if len(sys.argv) != 2:
            raise Refusal('protocol', 'one native destination argument is required')
        transaction = Transaction(os.fsencode(sys.argv[1]))
        result = perform(transaction, requested)
    except Refusal as failure:
        error = failure
    except PermissionError as failure:
        error = Refusal('permission', str(failure))
    except OSError as failure:
        kind = 'invalid_path' if failure.errno in (errno.ELOOP, errno.ENOTDIR) else 'io'
        error = Refusal(kind, str(failure))
    except (ValueError, TypeError, KeyError, AttributeError) as failure:
        error = Refusal('protocol', str(failure))
    finally:
        if transaction is not None:
            try:
                transaction.cleanup()
            except (OSError, Refusal) as failure:
                detail = str(failure)
                error = Refusal('io', ((error.detail + '; ') if error else '') + 'cleanup: ' + detail)
    if error is not None:
        result = {'status': 'refused', 'kind': error.kind, 'detail': error.detail[:4096],
                  'unconfirmed': bool(transaction and transaction.commit_started)}
    sys.stdout.write(json.dumps({'version': VERSION, 'result': result}, separators=(',', ':')) + '\n')
    sys.stdout.flush()
    return 1 if error is not None else 0


sys.exit(main())