"""Fixed Linux file helper protocol. Invoked in memory; never installed on the server.

One helper serves the Monaco editor save, privileged File Manager operations and the
final publication of SFTP transfers, so every replacement of an existing file follows
the same rules: staged copy, verified content, original owner/mode/ACL/xattr/SELinux
metadata, conflict re-check, atomic rename in the target directory, fsync and a final
verification. All path walks use directory descriptors and O_NOFOLLOW; symbolic links
are resolved explicitly first so a link (for example sites-enabled/x.conf) is kept and
its target file is edited. The editor working copy is retained after replacement: only
a subsequent confirmed cleanup removes it.
"""
import base64
import errno
import fcntl
import hashlib
import json
import os
import pwd
import re
import secrets
import stat
import sys

LIMIT = 25 * 1024 * 1024
CHUNK = 1024 * 1024
MAX_LINKS = 40
MAX_ENTRIES = 10000
MAX_REMOVALS = 200000
REPLACED = False
NO_XATTR = (errno.ENOTSUP, errno.EOPNOTSUPP)
NO_DIR_SYNC = (errno.EBADF, errno.EINVAL)


def fail(code):
    raise ValueError(code)


def check_path(path):
    if not isinstance(path, str) or not path.startswith('/') or len(path) > 4096 or '\x00' in path \
            or any(p in ('.', '..') for p in path.split('/')):
        fail('FILE_UNSAFE_PATH')
    return path


def b64(data):
    return base64.b64encode(data).decode('ascii')


def trusted(fd, privileged):
    # A privileged walk only refuses directories that any account may rewrite
    # (world-writable without the sticky bit). Descriptor-relative O_NOFOLLOW
    # opens already exclude symlink swaps of path components.
    if privileged:
        s = os.fstat(fd)
        if s.st_mode & 0o002 and not s.st_mode & stat.S_ISVTX:
            fail('FILE_UNSAFE_PRIVILEGED_DIRECTORY')


def directory(path, privileged=False, readable=False):
    """Open path without following any symlink. Intermediate components only need
    search permission (O_PATH); the final directory is opened for reading when
    possible so it can be listed and fsynced."""
    check_path(path)
    parts = [p for p in path.split('/') if p]
    fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY)
    try:
        trusted(fd, privileged)
        for index, part in enumerate(parts):
            last = index == len(parts) - 1
            flags = os.O_DIRECTORY | os.O_NOFOLLOW
            try:
                nxt = os.open(part, flags | (os.O_RDONLY if last else os.O_PATH), dir_fd=fd)
            except PermissionError:
                if not last or readable:
                    raise
                nxt = os.open(part, flags | os.O_PATH, dir_fd=fd)
            os.close(fd)
            fd = nxt
            trusted(fd, privileged)
        return fd
    except BaseException:
        os.close(fd)
        raise


def resolve(path, missing='none'):
    """Canonical path with every symbolic link resolved (at most MAX_LINKS hops).
    missing='leaf' accepts a missing final component, 'tail' any missing suffix."""
    check_path(path)
    pending = [p for p in path.split('/') if p]
    done = []
    links = 0
    absent = False
    while pending:
        part = pending.pop(0)
        if part == '.':
            continue
        if part == '..':
            if done and not absent:
                done.pop()
            elif absent:
                fail('FILE_UNSAFE_PATH')
            continue
        if absent:
            done.append(part)
            continue
        current = '/' + '/'.join(done + [part])
        try:
            st = os.lstat(current)
        except FileNotFoundError:
            if links and not pending and missing == 'none':
                fail('FILE_SYMLINK_TARGET_MISSING')
            if (missing == 'leaf' and not pending) or missing == 'tail':
                absent = True
                done.append(part)
                continue
            raise
        if stat.S_ISLNK(st.st_mode):
            links += 1
            if links > MAX_LINKS:
                fail('FILE_SYMLINK_LOOP')
            value = os.readlink(current)
            if value.startswith('/'):
                done = []
            pending = [p for p in value.split('/') if p] + pending
        else:
            done.append(part)
    return '/' + '/'.join(done), links > 0


def locate(path, privileged=False, missing='none'):
    """Resolve symlinks, then open the real parent directory safely."""
    resolved, linked = resolve(path, missing)
    parent, name = os.path.split(resolved)
    if not name:
        fail('FILE_UNSAFE_PATH')
    return directory(parent, privileged), name, resolved, linked


def locate_entry(path, privileged=False):
    """Parent resolved, final component taken literally (rename/remove act on links)."""
    check_path(path)
    parent, name = os.path.split(path.rstrip('/') or '/')
    if not name:
        fail('FILE_UNSAFE_PATH')
    resolved, _ = resolve(parent)
    return directory(resolved, privileged), name, os.path.join(resolved, name)


def regular(parent, name, owner=None, private=False, single=True):
    fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
    s = os.fstat(fd)
    code = None
    if not stat.S_ISREG(s.st_mode):
        code = 'FILE_UNSAFE_TYPE_OR_LINK'
    elif single and s.st_nlink != 1:
        # Replacing the inode would silently split content shared with other links.
        code = 'FILE_HARDLINK_UNSUPPORTED' if owner is None else 'FILE_UNSAFE_TYPE_OR_LINK'
    elif (owner is not None and s.st_uid != owner) or (private and stat.S_IMODE(s.st_mode) != 0o600):
        code = 'FILE_UNSAFE_TYPE_OR_LINK'
    if code:
        os.close(fd)
        fail(code)
    return fd


def attributes(fd):
    # Linux ACLs, capabilities and SELinux labels are extended attributes too.
    # A filesystem without xattr support simply has none to preserve.
    try:
        keys = os.listxattr(fd)
    except OSError as e:
        if e.errno in NO_XATTR:
            return {}
        raise
    result = {}
    for key in sorted(keys):
        try:
            result[key] = b64(os.getxattr(fd, key))
        except OSError as e:
            if e.errno != errno.ENODATA:
                raise
    return result


def describe(fd):
    s = os.fstat(fd)
    return dict(dev=s.st_dev, ino=s.st_ino, uid=s.st_uid, gid=s.st_gid, mode=stat.S_IMODE(s.st_mode),
                size=s.st_size, mtimeNs=str(s.st_mtime_ns), ctimeNs=str(s.st_ctime_ns), xattrs=attributes(fd))


def snapshot(fd):
    before = os.fstat(fd)
    if before.st_size > LIMIT:
        fail('FILE_TOO_LARGE')
    os.lseek(fd, 0, os.SEEK_SET)
    pieces = []
    length = 0
    while True:
        piece = os.read(fd, min(65536, LIMIT + 1 - length))
        if not piece:
            break
        pieces.append(piece)
        length += len(piece)
        if length > LIMIT:
            fail('FILE_TOO_LARGE')
    data = b''.join(pieces)
    meta = describe(fd)
    after = os.fstat(fd)
    if (before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns):
        fail('FILE_CONFLICT')
    meta['hash'] = hashlib.sha256(data).hexdigest()
    return data, meta


def lite(s):
    return dict(dev=s.st_dev, ino=s.st_ino, size=s.st_size, mtimeNs=str(s.st_mtime_ns), ctimeNs=str(s.st_ctime_ns),
                uid=s.st_uid, gid=s.st_gid, mode=stat.S_IMODE(s.st_mode), nlink=s.st_nlink)


def kind(s):
    return 'directory' if stat.S_ISDIR(s.st_mode) else 'symlink' if stat.S_ISLNK(s.st_mode) else 'file' if stat.S_ISREG(s.st_mode) else 'other'


def stream_hash(fd, out=None, limit=None):
    os.lseek(fd, 0, os.SEEK_SET)
    digest = hashlib.sha256()
    total = 0
    while True:
        piece = os.read(fd, CHUNK)
        if not piece:
            break
        digest.update(piece)
        total += len(piece)
        if limit is not None and total > limit:
            fail('FILE_VERIFY_FAILED')
        if out is not None:
            view = memoryview(piece)
            while view:
                view = view[os.write(out, view):]
    return digest.hexdigest(), total


def fsync_dir(fd):
    try:
        os.fsync(fd)
    except OSError as e:
        if e.errno not in NO_DIR_SYNC:
            raise


def protected_metadata(meta):
    # Copying a content/inode-bound integrity signature would leave invalid
    # security metadata. Re-signing requires a separate flow.
    if set(meta['xattrs']) & {'security.ima', 'security.evm'}:
        fail('FILE_METADATA_UNSUPPORTED')


def apply_metadata(fd, meta):
    """Owner first (chown clears setuid/capabilities), then mode, then the exact
    xattr set of the original. Only differing values are written, so filesystems
    that refuse no-op changes (and unprivileged saves of own files) still work."""
    try:
        s = os.fstat(fd)
        if (s.st_uid, s.st_gid) != (meta['uid'], meta['gid']):
            os.fchown(fd, meta['uid'], meta['gid'])
        if stat.S_IMODE(os.fstat(fd).st_mode) != meta['mode']:
            os.fchmod(fd, meta['mode'])
        current = attributes(fd)
        for key in current:
            if key not in meta['xattrs']:
                os.removexattr(fd, key)
        for key, value in meta['xattrs'].items():
            if current.get(key) != value:
                os.setxattr(fd, key, base64.b64decode(value))
    except PermissionError:
        fail('FILE_METADATA_PERMISSION')
    except OSError as e:
        if e.errno in NO_XATTR:
            fail('FILE_METADATA_UNSUPPORTED')
        raise


def preserved(actual, original):
    return all(actual[k] == original[k] for k in ('uid', 'gid', 'mode', 'xattrs'))


def visible(meta):
    # trusted.* attributes are visible to root only (e.g. XFS mirrors POSIX ACLs in
    # trusted.SGI_ACL_FILE). A baseline captured by the SSH account cannot contain them,
    # so conflict detection compares the attributes both sides can see; any xattr change
    # still updates ctime, which stays part of the comparison.
    return dict(meta, xattrs={k: v for k, v in meta['xattrs'].items() if not k.startswith('trusted.')})


def same_state(current, baseline):
    return visible(current) == visible(baseline)


def account_home(uid):
    try:
        return pwd.getpwuid(uid).pw_dir
    except KeyError:
        return None


def subdir(d, name, uid):
    created = False
    try:
        os.mkdir(name, 0o700, dir_fd=d)
        created = True
    except FileExistsError:
        pass
    fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=d)
    try:
        s = os.fstat(fd)
        if s.st_uid != uid:
            if not (created and os.geteuid() == 0):
                fail('FILE_UNSAFE_WORKSPACE')
            os.fchown(fd, uid, -1)
        os.fchmod(fd, 0o700)
        return fd
    except BaseException:
        os.close(fd)
        raise


def workspace(uid=None):
    """Private 0700 working directory of the account: ~/ch-j-sm, or /var/tmp/chj-sm-<uid>
    when the home directory is missing or not private to the account."""
    uid = os.getuid() if uid is None else uid
    home = account_home(uid)
    h = None
    if home:
        try:
            h = directory(check_path(home), readable=True)
            hs = os.fstat(h)
            if hs.st_uid != uid or hs.st_mode & 0o022 or 'system.posix_acl_access' in attributes(h):
                os.close(h)
                h = None
        except (OSError, ValueError):
            if h is not None:
                os.close(h)
            h = None
    if h is None:
        home = '/var/tmp'
        h = directory(home, readable=True)
        name = 'chj-sm-%d' % uid
    else:
        name = 'ch-j-sm'
    try:
        return subdir(h, name, uid), os.path.join(home, name), uid
    finally:
        os.close(h)


def operation_id(value):
    if not re.fullmatch('[a-f0-9]{32}', str(value)):
        fail('FILE_INVALID_RECOVERY_ID')
    return value


def names(value):
    value = operation_id(value)
    return '.' + value + '.tmp', '.' + value + '.json'


def put_record(d, name, value):
    data = json.dumps(value).encode('ascii')
    if len(data) > 256 * 1024:
        fail('FILE_METADATA_UNSUPPORTED')
    fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=d)
    with os.fdopen(fd, 'wb') as out:
        os.fchmod(out.fileno(), 0o600)
        out.write(data)
        out.flush()
        os.fsync(out.fileno())
    os.fsync(d)


def get_record(d, name, uid):
    fd = regular(d, name, uid, True)
    with os.fdopen(fd, 'rb') as inp:
        raw = inp.read(256 * 1024 + 1)
    if len(raw) > 256 * 1024:
        fail('FILE_INVALID_RECOVERY_RECORD')
    record = json.loads(raw)
    if not re.fullmatch('[a-f0-9]{64}', record['hash']):
        fail('FILE_INVALID_RECOVERY_RECORD')
    return record


def inspect_record(d, uid, value, include_text=False):
    tmp, journal = names(value)
    r = get_record(d, journal, uid)
    fd = regular(d, tmp, uid, True)
    try:
        data, staged = snapshot(fd)
    finally:
        os.close(fd)
    complete = staged['hash'] == r['hash']
    state = 'upload-incomplete'
    if complete:
        state = 'recovery-available'
        try:
            td, name, resolved, _ = locate(r['path'])
            try:
                t = regular(td, name)
                try:
                    _, current = snapshot(t)
                finally:
                    os.close(t)
                if current['hash'] == r['hash'] and preserved(visible(current), visible(r['baseline'])):
                    state = 'confirmed'
                elif not same_state(current, r['baseline']) or resolved != r.get('resolvedPath', resolved):
                    state = 'conflict'
            finally:
                os.close(td)
        except (OSError, ValueError):
            pass  # Permission denied/disconnected from target: copy remains useful.
    result = dict(recoveryId=value, path=r['path'], status=state, hash=staged['hash'], complete=complete)
    if include_text:
        if not complete:
            fail('FILE_UPLOAD_INCOMPLETE')
        result['data'] = b64(data)
    return result


def direct_save(d, meta):
    euid = os.geteuid()
    if euid == 0:
        return True
    try:
        groups = set(os.getgroups()) | {os.getegid()}
        return meta['uid'] == euid and meta['gid'] in groups and os.access('.', os.W_OK | os.X_OK, dir_fd=d)
    except OSError:
        return False


def backup_copy(ident, data, meta):
    """Optional rollback copy of the original in the effective account's private
    workspace (root's own under sudo), never next to the system file."""
    d, home, _ = workspace(os.geteuid())
    try:
        b = subdir(d, 'backups', os.geteuid())
        try:
            name = ident + '.orig'
            fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=b)
            with os.fdopen(fd, 'wb') as out:
                out.write(data)
                out.flush()
                os.fsync(out.fileno())
            fsync_dir(b)
            return os.path.join(home, 'backups', name)
        finally:
            os.close(b)
    finally:
        os.close(d)


def finalize(a, d, uid):
    global REPLACED
    tmp, _ = names(a['id'])
    source = regular(d, tmp, uid, True)
    try:
        data, staged = snapshot(source)
        if staged['hash'] != a['hash']:
            fail('FILE_UPLOAD_INCOMPLETE')
        os.fsync(source)
        td, name, resolved, linked = locate(a['path'], os.geteuid() == 0)
        if a.get('resolvedPath') and a['resolvedPath'] != resolved:
            os.close(td)
            fail('FILE_CONFLICT')  # The symbolic link now points elsewhere.
        destination = '.chj-save-' + secrets.token_hex(16) + '.tmp'
        created = False
        old = None
        try:
            old = regular(td, name)
            fcntl.flock(old, fcntl.LOCK_EX | fcntl.LOCK_NB)
            # The full (possibly root) view of the original is the metadata source.
            original, baseline = snapshot(old)
            if not same_state(baseline, a['baseline']):
                fail('FILE_CONFLICT')
            protected_metadata(baseline)
            dest = os.open(destination, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=td)
            created = True
            try:
                # Write fully before applying security metadata (capabilities
                # and setuid bits can be cleared by writes or chown).
                with os.fdopen(os.dup(dest), 'wb') as out:
                    out.write(data)
                    out.flush()
                apply_metadata(dest, baseline)
                _, ready = snapshot(dest)
                if ready['hash'] != a['hash'] or not preserved(ready, baseline):
                    fail('FILE_METADATA_UNSUPPORTED')
                os.fsync(dest)
                # Detect writes, chmod/chown, link changes and path replacement
                # while preparing. flock serializes cooperating editor saves.
                _, current = snapshot(old)
                linked_now = os.stat(name, dir_fd=td, follow_symlinks=False)
                if current != baseline or linked_now.st_ino != baseline['ino'] or linked_now.st_dev != baseline['dev'] or linked_now.st_nlink != 1:
                    fail('FILE_CONFLICT')
                backup = backup_copy(a['id'], original, baseline) if a.get('backup') else None
                os.replace(destination, name, src_dir_fd=td, dst_dir_fd=td)
                REPLACED = True
                created = False
                fsync_dir(td)
                check_dir, check_name, _, _ = locate(a['path'])
                try:
                    verified = regular(check_dir, check_name)
                    try:
                        _, final = snapshot(verified)
                    finally:
                        os.close(verified)
                finally:
                    os.close(check_dir)
                if final['ino'] != ready['ino'] or final['hash'] != a['hash'] or not preserved(final, baseline):
                    fail('FILE_RESULT_UNCONFIRMED')
                result = dict(status='saved', baseline=final, hash=final['hash'], resolvedPath=resolved, link=linked)
                if backup:
                    result['backupPath'] = backup
                return result
            finally:
                os.close(dest)
        finally:
            if created:
                try:
                    os.unlink(destination, dir_fd=td)
                except OSError:
                    pass
            if old is not None:
                os.close(old)
            os.close(td)
    finally:
        os.close(source)


def make_parents(path):
    """Create the missing directories of an already resolved path (mode 0755)."""
    fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY)
    try:
        trusted(fd, os.geteuid() == 0)
        for part in [p for p in check_path(path).split('/') if p]:
            try:
                os.mkdir(part, 0o755, dir_fd=fd)
            except FileExistsError:
                pass
            nxt = os.open(part, os.O_PATH | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
            os.close(fd)
            fd = nxt
            trusted(fd, os.geteuid() == 0)
    finally:
        os.close(fd)


def target_state(td, name):
    try:
        s = os.stat(name, dir_fd=td, follow_symlinks=False)
    except FileNotFoundError:
        return None
    if not stat.S_ISREG(s.st_mode):
        fail('FILE_UNSAFE_TYPE_OR_LINK')
    return s


def already_published(td, name, expected, size):
    s = target_state(td, name)
    if s is None or s.st_size != size:
        return False
    fd = regular(td, name, single=False)
    try:
        return stream_hash(fd)[0] == expected
    finally:
        os.close(fd)


def place(td, temp, name, exists):
    """Atomic replacement of an existing file, no-clobber publication of a new one."""
    if exists:
        os.replace(temp, name, src_dir_fd=td, dst_dir_fd=td)
        return
    try:
        os.link(temp, name, src_dir_fd=td, dst_dir_fd=td, follow_symlinks=False)
    except FileExistsError:
        fail('FILE_CONFLICT')
    except OSError as e:
        if e.errno not in (errno.EPERM, errno.ENOTSUP, errno.EOPNOTSUPP, errno.EMLINK, errno.EXDEV):
            raise
        if target_state(td, name) is not None:  # Filesystem without hard links.
            fail('FILE_CONFLICT')
        os.rename(temp, name, src_dir_fd=td, dst_dir_fd=td)
        return
    os.unlink(temp, dir_fd=td)


def publish(a):
    """Move a verified, completely transferred file into its destination."""
    global REPLACED
    expected, size, uid, strategy = a['hash'], int(a['size']), int(a['uid']), a['strategy']
    if not re.fullmatch('[a-f0-9]{64}', str(expected)) or strategy not in ('rename', 'copy'):
        fail('FILE_PROTOCOL_ERROR')
    new_mode = valid_mode(a.get('mode'), 0o644, 0o777)
    source = check_path(a['source'])
    if a.get('parents'):
        parent, _ = resolve(os.path.dirname(check_path(a['path'])), 'tail')
        make_parents(parent)
    td, name, resolved, linked = locate(a['path'], os.geteuid() == 0, 'leaf')
    sd = src = work = old = None
    temp = None
    try:
        base = a.get('baseline')
        current = target_state(td, name)
        if current is not None and current.st_nlink != 1:
            fail('FILE_HARDLINK_UNSUPPORTED')
        changed = (current is None) != (base is None) or (current is not None and lite(current) != base)
        if changed and a.get('retry') and already_published(td, name, expected, size):
            return dict(status='published', path=a['path'], resolvedPath=resolved, link=linked, size=size, hash=expected,
                        alreadyPublished=True, sourceRemoved=False)
        if current is not None and base is None and not a.get('overwrite'):
            fail('FILE_EXISTS')
        if changed and not (current is not None and base is None):
            fail('FILE_CONFLICT')
        meta = None
        if current is not None:
            old = regular(td, name)
            fcntl.flock(old, fcntl.LOCK_EX | fcntl.LOCK_NB)
            meta = describe(old)
            protected_metadata(meta)
        if strategy == 'rename':
            # Direct transfers are written next to the real target by the SSH account.
            if os.path.dirname(source) != os.path.dirname(resolved):
                fail('FILE_PROTOCOL_ERROR')
            temp = os.path.basename(source)
            src = work = regular(td, temp, uid, True)
            digest, length = stream_hash(src)
        else:
            # Staged in the account workspace: copy into a new file beside the target so
            # it receives the directory's default labels/ACLs (new) or the original's.
            sd = directory(os.path.dirname(source))
            src = regular(sd, os.path.basename(source), uid, True)
            temp = '.chj-publish-' + secrets.token_hex(16) + '.tmp'
            work = os.open(temp, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=td)
            digest, length = stream_hash(src, work)
        if digest != expected or length != size:
            fail('FILE_VERIFY_FAILED')
        if meta is not None:
            apply_metadata(work, meta)
            if not preserved(describe(work), meta):
                fail('FILE_METADATA_UNSUPPORTED')
        elif stat.S_IMODE(os.fstat(work).st_mode) != new_mode:
            os.fchmod(work, new_mode)
        os.fsync(work)
        ready = os.fstat(work)
        if current is not None:
            now = target_state(td, name)
            if now is None or lite(now) != lite(current):
                fail('FILE_CONFLICT')
        place(td, temp, name, current is not None)
        REPLACED = True
        fsync_dir(td)
        final = target_state(td, name)
        if final is None or final.st_ino != ready.st_ino or final.st_size != size:
            fail('FILE_RESULT_UNCONFIRMED')
        removed = strategy == 'rename'
        if sd is not None:
            try:
                os.unlink(os.path.basename(source), dir_fd=sd)
                fsync_dir(sd)
                removed = True
            except OSError:
                pass
        return dict(status='published', path=a['path'], resolvedPath=resolved, link=linked, size=size, hash=expected,
                    created=current is None, metadataPreserved=meta is not None, sourceRemoved=removed)
    finally:
        if strategy == 'copy' and work is not None:
            os.close(work)
            if not REPLACED:
                try:
                    os.unlink(temp, dir_fd=td)
                except OSError:
                    pass
        for fd in (src, old, sd, td):
            if fd is not None:
                os.close(fd)


def stage_out(a):
    """Privileged download: copy a protected file into the account's private
    transfer workspace so the SSH account can fetch it with resumable SFTP."""
    uid = int(a['uid'])
    part = operation_id(a['id']) + '.part'
    sd, sname, resolved, linked = locate(a['path'], os.geteuid() == 0)
    try:
        src = regular(sd, sname, single=False)
    finally:
        os.close(sd)
    try:
        wd, home, _ = workspace(uid)
        try:
            td = subdir(wd, 'transfers', uid)
        finally:
            os.close(wd)
        try:
            try:
                os.close(regular(td, part, uid, True))
                os.unlink(part, dir_fd=td)
            except FileNotFoundError:
                pass
            before = os.fstat(src)
            dest = os.open(part, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=td)
            try:
                digest, length = stream_hash(src, dest)
                after = os.fstat(src)
                if (before.st_ino, before.st_size, before.st_mtime_ns) != (after.st_ino, after.st_size, after.st_mtime_ns) or length != after.st_size:
                    fail('FILE_CONFLICT')
                if os.geteuid() != uid:
                    try:
                        gid = pwd.getpwuid(uid).pw_gid
                    except KeyError:
                        gid = -1
                    os.fchown(dest, uid, gid)
                os.fsync(dest)
            except BaseException:
                os.close(dest)
                try:
                    os.unlink(part, dir_fd=td)
                except OSError:
                    pass
                raise
            os.close(dest)
            fsync_dir(td)
            return dict(path=os.path.join(home, 'transfers', part), size=length, hash=digest, mtimeNs=str(after.st_mtime_ns),
                        ino=after.st_ino, dev=after.st_dev, resolvedPath=resolved, link=linked, mode=stat.S_IMODE(after.st_mode))
        finally:
            os.close(td)
    finally:
        os.close(src)


def remove_tree(parent, name, state):
    fd = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
    try:
        for entry in os.listdir(fd):
            state['entries'] += 1
            if state['entries'] > MAX_REMOVALS:
                fail('FILE_TOO_MANY_ENTRIES')
            s = os.stat(entry, dir_fd=fd, follow_symlinks=False)
            if stat.S_ISDIR(s.st_mode):
                remove_tree(fd, entry, state)
            else:
                os.unlink(entry, dir_fd=fd)
    finally:
        os.close(fd)
    os.rmdir(name, dir_fd=parent)


def list_directory(a):
    resolved, linked = resolve(a['path'])
    d = directory(resolved, readable=True)
    try:
        entries = []
        names_ = sorted(os.listdir(d))
        for name in names_[:MAX_ENTRIES]:
            try:
                s = os.stat(name, dir_fd=d, follow_symlinks=False)
            except FileNotFoundError:
                continue
            entry = dict(name=name, type=kind(s), size=s.st_size, mtime=int(s.st_mtime), mode=stat.S_IMODE(s.st_mode) | stat.S_IFMT(s.st_mode))
            if stat.S_ISLNK(s.st_mode):
                try:
                    entry['targetType'] = kind(os.stat(name, dir_fd=d))
                except OSError:
                    entry['targetType'] = 'missing'
            entries.append(entry)
        return dict(path=a['path'], resolvedPath=resolved, link=linked, entries=entries, truncated=len(names_) > MAX_ENTRIES)
    finally:
        os.close(d)


def valid_mode(value, default, mask=0o7777):
    mode = default if value is None else int(value)
    if mode < 0 or mode & ~mask:
        fail('FILE_PROTOCOL_ERROR')
    return mode


def file_operation(op, a):
    privileged = os.geteuid() == 0
    if op == 'read':
        d, name, resolved, linked = locate(a['path'])
        try:
            fd = regular(d, name)
            try:
                data, meta = snapshot(fd)
                return dict(data=b64(data), baseline=meta, resolvedPath=resolved, link=linked, directSave=direct_save(d, meta))
            finally:
                os.close(fd)
        finally:
            os.close(d)
    if op == 'stat':
        d, name, resolved, linked = locate(a['path'], missing='leaf')
        try:
            try:
                s = os.stat(name, dir_fd=d, follow_symlinks=False)
            except FileNotFoundError:
                return dict(exists=False, resolvedPath=resolved, link=linked,
                            writable=os.access('.', os.W_OK | os.X_OK, dir_fd=d))
            return dict(exists=True, resolvedPath=resolved, link=linked, type=kind(s), baseline=lite(s),
                        writable=os.access('.', os.W_OK | os.X_OK, dir_fd=d) and os.access(name, os.W_OK, dir_fd=d))
        finally:
            os.close(d)
    if op == 'listdir':
        return list_directory(a)
    if op == 'hash':
        d, name, resolved, _ = locate(a['path'])
        try:
            fd = regular(d, name, single=False)
        finally:
            os.close(d)
        try:
            digest, length = stream_hash(fd)
            return dict(hash=digest, size=length, resolvedPath=resolved)
        finally:
            os.close(fd)
    if op in ('mkdir', 'create'):
        if a.get('parents'):
            parent, _ = resolve(os.path.dirname(check_path(a['path'])), 'tail')
            make_parents(parent)
        d, name, path = locate_entry(a['path'], privileged)
        try:
            if op == 'mkdir':
                os.mkdir(name, valid_mode(a.get('mode'), 0o755, 0o1777), dir_fd=d)
            else:
                fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, valid_mode(a.get('mode'), 0o644, 0o777), dir_fd=d)
                try:
                    os.fsync(fd)
                finally:
                    os.close(fd)
            fsync_dir(d)
            return dict(path=a['path'], resolvedPath=path)
        finally:
            os.close(d)
    if op == 'rename':
        sd, sname, _ = locate_entry(a['from'], privileged)
        try:
            dd, dname, _ = locate_entry(a['to'], privileged)
            try:
                os.stat(sname, dir_fd=sd, follow_symlinks=False)
                try:
                    os.stat(dname, dir_fd=dd, follow_symlinks=False)
                    fail('FILE_EXISTS')
                except FileNotFoundError:
                    pass
                os.rename(sname, dname, src_dir_fd=sd, dst_dir_fd=dd)
                fsync_dir(dd)
                fsync_dir(sd)
                return {'from': a['from'], 'to': a['to']}
            finally:
                os.close(dd)
        finally:
            os.close(sd)
    if op == 'remove':
        d, name, _ = locate_entry(a['path'], privileged)
        try:
            s = os.stat(name, dir_fd=d, follow_symlinks=False)
            state = dict(entries=1)
            if stat.S_ISDIR(s.st_mode):
                if a.get('recursive'):
                    remove_tree(d, name, state)
                else:
                    os.rmdir(name, dir_fd=d)
            else:
                os.unlink(name, dir_fd=d)
            fsync_dir(d)
            return dict(path=a['path'], type=kind(s), entries=state['entries'])
        finally:
            os.close(d)
    if op == 'publish':
        return publish(a)
    if op == 'stage-out':
        return stage_out(a)
    return None


def transfer_operation(op, a, d, home, uid):
    if op == 'transfer-workspace':
        os.close(subdir(d, 'transfers', uid))
        return dict(path=os.path.join(home, 'transfers'), uid=uid)
    td = subdir(d, 'transfers', uid)
    try:
        if op == 'transfer-list':
            items = []
            for name in sorted(os.listdir(td))[:MAX_ENTRIES]:
                if re.fullmatch(r'[a-f0-9]{32}\.part', name):
                    s = os.stat(name, dir_fd=td, follow_symlinks=False)
                    items.append(dict(id=name[:32], size=s.st_size, mtime=int(s.st_mtime)))
            return dict(path=os.path.join(home, 'transfers'), items=items)
        if op == 'transfer-remove':
            part = operation_id(a['id']) + '.part'
            try:
                os.close(regular(td, part, uid, True))
            except FileNotFoundError:
                return dict(removed=False)
            os.unlink(part, dir_fd=td)
            fsync_dir(td)
            return dict(removed=True)
    finally:
        os.close(td)
    fail('FILE_INVALID_OPERATION')


def run(a):
    op = a['operation']
    result = file_operation(op, a)
    if result is not None:
        return result
    d, home, uid = workspace(a.get('uid') if op != 'prepare' else None)
    try:
        if op.startswith('transfer-'):
            return transfer_operation(op, a, d, home, uid)
        if op == 'identity':
            return dict(uid=uid)
        if op == 'prepare':
            tmp, journal = names(a['id'])
            record = dict(path=a['path'], baseline=a['baseline'], hash=a['hash'])
            if a.get('resolvedPath'):
                record['resolvedPath'] = a['resolvedPath']
            put_record(d, journal, record)
            return dict(temporary=home + '/' + tmp, uid=uid, recoveryId=a['id'])
        if op == 'list':
            results = []
            for name in sorted(os.listdir(d))[:MAX_ENTRIES]:
                if re.fullmatch(r'\.[a-f0-9]{32}\.json', name):
                    try:
                        results.append(inspect_record(d, uid, name[1:-5]))
                    except (OSError, ValueError, KeyError):
                        results.append(dict(recoveryId=name[1:-5], status='inspection-unavailable'))
            return dict(items=results)
        tmp, journal = names(a['id'])
        if op in ('inspect', 'cleanup'):
            result = inspect_record(d, uid, a['id'], op == 'inspect')
            if op == 'cleanup':
                if result['status'] != 'confirmed':
                    fail('FILE_RESULT_UNCONFIRMED')
                os.unlink(tmp, dir_fd=d)
                os.unlink(journal, dir_fd=d)
                os.fsync(d)
            return result
        if op != 'finalize':
            fail('FILE_INVALID_OPERATION')
        return finalize(a, d, uid)
    finally:
        os.close(d)


def read_request():
    if len(sys.argv) > 1:
        return json.loads(sys.argv[1])
    # Privileged runs receive the request as a stdin frame after the READY marker.
    line = sys.stdin.buffer.readline(16 * 1024 * 1024)
    if not line.startswith(b'CHJREQ:'):
        fail('FILE_PROTOCOL_ERROR')
    return json.loads(base64.b64decode(line[7:].strip()))


ERRNO_CODES = {
    errno.ELOOP: 'FILE_UNSAFE_PATH', errno.ENOTDIR: 'FILE_UNSAFE_PATH', errno.ENAMETOOLONG: 'FILE_UNSAFE_PATH',
    errno.ENOSPC: 'FILE_NO_SPACE', errno.EDQUOT: 'FILE_NO_SPACE', errno.EROFS: 'FILE_READ_ONLY_FILESYSTEM',
    errno.EXDEV: 'FILE_CROSS_DEVICE', errno.ENOTEMPTY: 'FILE_DIRECTORY_NOT_EMPTY', errno.EBUSY: 'FILE_BUSY',
    errno.ETXTBSY: 'FILE_BUSY', errno.EISDIR: 'FILE_IS_DIRECTORY', errno.EEXIST: 'FILE_EXISTS'
}


def respond(code):
    print(json.dumps(dict(ok=False, code='FILE_RESULT_UNKNOWN' if REPLACED else code)))


try:
    if sys.platform != 'linux':
        fail('FILE_LINUX_REQUIRED')
    print(json.dumps(dict(ok=True, value=run(read_request()))))
except BlockingIOError:
    respond('FILE_CONFLICT')
except PermissionError:
    respond('FILE_PERMISSION_DENIED')
except FileNotFoundError:
    respond('FILE_NOT_FOUND')
except OSError as e:
    respond(ERRNO_CODES.get(e.errno, 'FILE_REMOTE_IO_FAILED'))
except (ValueError, KeyError, TypeError):
    e = sys.exc_info()[1]
    respond(str(e) if re.fullmatch('FILE_[A-Z_]+', str(e)) else 'FILE_PROTOCOL_ERROR')