"""Fixed, root-protected C01 access policy and guarded filesystem operations.

There is deliberately no RPC for installing policy, selecting code or paths, or
supplying observations. Only the offline `transition` entry consumes this file.
The manifest is an operator-reviewed inventory, not evidence that keys were never
copied. An unresolved/external signing copy prevents transition.
"""
import hashlib
import json
import os
from pathlib import Path
import re
import stat

CONFIG = Path('/etc/archipelago-c01/access.json')
STAGED = Path('/etc/archipelago-c01/access-policy')
STATE = Path('/var/lib/archipelago-c01/access')
RUNTIME = Path('/etc/archipelago-c01/runtime.json')
LAUNCH = Path('/etc/archipelago-c01/launch.json')
LIMIT = 1024 * 1024
TARGETS = {
    'passwd': Path('/etc/passwd'), 'shadow': Path('/etc/shadow'),
    'group': Path('/etc/group'), 'gshadow': Path('/etc/gshadow'),
    'sudoers': Path('/etc/sudoers'),
    'polkit': Path('/etc/polkit-1/rules.d/00-archipelago-c01.rules'),
    'sshd': Path('/etc/ssh/sshd_config'),
    'cron-deny': Path('/etc/cron.deny'), 'at-deny': Path('/etc/at.deny'),
}


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


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


def canonical(path):
    path = Path(path)
    require(path.is_absolute() and '..' not in path.parts and str(path) != '/', 'canonical absolute path required')
    return path


def protected(path, private=False, directory=False):
    path = canonical(path)
    for parent in reversed(path.parents):
        meta = parent.lstat()
        require(stat.S_ISDIR(meta.st_mode) and meta.st_uid == 0 and not meta.st_mode & 0o022,
                'mutable access-policy ancestry')
    meta = path.lstat()
    require(meta.st_uid == 0 and not meta.st_mode & (0o077 if private else 0o022) and
            (stat.S_ISDIR(meta.st_mode) if directory else stat.S_ISREG(meta.st_mode)) and
            (directory or meta.st_nlink == 1), 'unprotected access-policy object: ' + str(path))
    return meta


def read(path, private=False):
    before = protected(path, private)
    require(0 <= before.st_size <= LIMIT, 'access input size')
    fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC | os.O_NONBLOCK)
    try:
        def stable(meta):
            # Reading may legitimately advance atime (including relatime on
            # private tmpfs). It is not part of content/ownership identity.
            fields = ('st_dev', 'st_ino', 'st_mode', 'st_uid', 'st_gid', 'st_nlink',
                      'st_size', 'st_mtime_ns', 'st_ctime_ns')
            return all(getattr(meta, key) == getattr(before, key) for key in fields)
        require(stable(os.fstat(fd)), 'access input replaced')
        raw = os.read(fd, LIMIT + 1)
        require(len(raw) == before.st_size and stable(os.fstat(fd)), 'access input changed')
        return raw
    finally:
        os.close(fd)


def sha(raw):
    return hashlib.sha256(raw).hexdigest()


def hex_digest(value):
    return type(value) is str and len(value) == 64 and all(c in '0123456789abcdef' for c in value)


def json_read(path):
    raw = read(path, True)
    return json.loads(raw, object_pairs_hook=exact), sha(raw)


def identity(meta):
    return [meta.st_dev, meta.st_ino, meta.st_uid, meta.st_gid, stat.S_IMODE(meta.st_mode)]


def open_path(path, directory=False):
    """No-follow traversal, including runtime-owned legacy ancestry."""
    path = canonical(path)
    fd = os.open('/', os.O_PATH | os.O_DIRECTORY | os.O_CLOEXEC)
    try:
        for index, component in enumerate(path.parts[1:], 1):
            flags = os.O_PATH | os.O_NOFOLLOW | os.O_CLOEXEC
            if index < len(path.parts) - 1 or directory:
                flags |= os.O_DIRECTORY
            child = os.open(component, flags, dir_fd=fd)
            os.close(fd)
            fd = child
            require(not stat.S_ISLNK(os.fstat(fd).st_mode), 'alias in access inventory')
        return fd
    except BaseException:
        os.close(fd)
        raise


def atomic(path, raw, mode=0o600):
    """Never truncate a credential/policy file in place; no secret diagnostics."""
    protected(path.parent, directory=True)
    temporary = path.with_name('.' + path.name + '.c01-new')
    fd = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, mode)
    try:
        os.fchmod(fd, mode)
        require(os.write(fd, raw) == len(raw), 'short policy write')
        os.fsync(fd)
    finally:
        os.close(fd)
    os.replace(temporary, path)
    parent = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_CLOEXEC)
    try:
        os.fsync(parent)
    finally:
        os.close(parent)


def sudo_policy(operator, authentication="password"):
    require(authentication in ("password", "preserve-verified-noninteractive"),
            "finite administrator sudo authentication choice required")
    require(authentication == "password" or operator == "archipelago-admin",
            "noninteractive preservation is restricted to the separate administrator")
    tag = "NOPASSWD: " if authentication == "preserve-verified-noninteractive" else ""
    # No include, group, netgroup, LDAP or runtime-specific positive grant.
    return ('Defaults env_reset\nDefaults use_pty\nroot ALL=(ALL:ALL) ALL\n' +
            operator + ' ALL=(ALL:ALL) ' + tag + 'ALL\n').encode()


def polkit_policy(names):
    return ('polkit.addRule(function(action, subject) {\n' +
            '  if (subject.user !== "root" && subject.user !== "archipelago-admin" &&\n' +
            '      (action.id.indexOf("org.freedesktop.login1.") === 0 ||\n' +
            '       action.id.indexOf("org.freedesktop.systemd1.") === 0 ||\n' +
            '       action.id.indexOf("org.freedesktop.timedate1.") === 0 ||\n' +
            '       action.id === "org.freedesktop.policykit.exec")) return polkit.Result.NO;\n' +
            '  if (' + ' || '.join('subject.user === ' + json.dumps(name) for name in names) +
            ') return polkit.Result.NO;\n});\n').encode()


def ssh_policy(baseline, names):
    """Finite transformation of an already verified administrator configuration.

    Deliberately support global-only, self-contained configurations. Includes or
    Match blocks need separately reviewed flattening BEFORE this transaction;
    sampling addresses cannot establish a universal denial across those blocks.
    """
    require(b'\\' not in baseline and b'\x00' not in baseline, 'unsupported SSH continuation/input')
    for line in baseline.decode().splitlines():
        text = line.strip()
        if not text or text.startswith('#'):
            continue
        match = re.match(r'^([a-zA-Z][a-zA-Z0-9]*)(?:\s|=|$)', text)
        require(match is not None and match[1].lower() not in ('include', 'match', 'denyusers'),
                'SSH baseline must be global, self-contained and without DenyUsers')
    return ('DenyUsers ' + ' '.join(names) + '\n').encode() + baseline


def ssh_baseline(postimage, names, before):
    prefix = ('DenyUsers ' + ' '.join(names) + '\n').encode()
    require(postimage.startswith(prefix), 'global managed SSH denial missing')
    baseline = postimage[len(prefix):]
    require(sha(baseline) == before and ssh_policy(baseline, names) == postimage,
            'SSH transformation must preserve exact verified administrator baseline')
    return baseline


class Policy:
    def __init__(self):
        self.value, self.digest = json_read(CONFIG)
        v = self.value
        fields = {'version', 'runtime_sha256', 'launch_sha256', 'operator', 'accounts',
                            'files', 'trees', 'cohorts', 'copies', 'preserved', 'restart_units',
                            'broker_cgroup', 'supervisor_cgroup', 'inventory_scope', 'authorization'}
        if v.get('version') in (2,3): fields.add('d09_workspace')
        if v.get('version')==3: fields.add('migration_layout')
        if 'operator_sudo_authentication' in v:
            require(v.get('version') == 3 and v['operator_sudo_authentication'] in
                    ('password', 'preserve-verified-noninteractive'), 'administrator sudo appointment')
            fields.add('operator_sudo_authentication')
        require(set(v) == fields,
                'access manifest schema')
        require(type(v['version']) is int and v['version'] in (1,2,3) and
                v['authorization'] in ('one-qualified-guarded-framework-migration',
                                       'one-qualified-guarded-yaya-migration') and
                v['inventory_scope'] == 'complete-known-writers-and-copies-no-unresolved-signers',
                'explicit migration authority/inventory required')
        self.runtime, digest = json_read(RUNTIME)
        require(digest == v['runtime_sha256'], 'runtime configuration binding changed')
        self.launch, digest = json_read(LAUNCH)
        import c01_access_generation
        self.generations = c01_access_generation.Generations(v, self.digest, self.launch, digest)
        if v['version'] in (2,3):
            require(all(not target.is_relative_to(canonical(path)) for path in v['trees']
                        for target in (LAUNCH,Path('/usr/local/bin/archipelago'))),
                    'native/launch must use explicit generation pins, not overlapping whole-tree pins')
        r, launch = self.runtime, self.launch
        require(v['operator'] == {'name': 'archipelago-admin', 'uid': 1001, 'gid': 1001},
                'separate approved operator required')
        require(set(v['accounts']) == {'runtime', 'manager', 'authority'}, 'managed account roles')
        for role, account in v['accounts'].items():
            require(set(account) == {'name', 'uid', 'gid'} and account['name'].replace('-', '').isalnum(),
                    'account schema')
            require((account['uid'], account['gid']) == (r[role + '_uid'], r[role + '_gid']),
                    'principal differs from runtime binding')
        require(v['accounts']['runtime'] == {'name': r['runtime_user'], 'uid': 1000, 'gid': 1000},
                'preserve original Framework runtime identity')
        require(v['accounts']['manager']['name'] == launch['application_user'] and
                v['accounts']['authority']['name'] == launch['authority_user'] and
                (r['operator_uid'], r['operator_gid']) == (1001, 1001), 'launch/operator binding')
        require(all(launch[key] == r[key] for key in ('runtime_uid', 'runtime_gid', 'runtime_data_dir')),
                'launch/runtime preservation binding differs')
        for kind in ('uid', 'gid'):
            ids = [a[kind] for a in [*v['accounts'].values(), v['operator']]]
            require(all(type(n) is int and 0 < n < 2**32 - 1 for n in ids) and len(set(ids)) == 4,
                    'principal overlap')
            allocations = r['sub' + kind + 's']
            require(type(allocations) is list and allocations, 'original subordinate map inventory missing')
            for pair in allocations:
                require(type(pair) is list and len(pair) == 2 and
                        all(type(n) is int for n in pair) and pair[0] > 0 and pair[1] > 0 and
                        pair[0] + pair[1] < 2**32 and
                        not any(pair[0] <= principal < pair[0] + pair[1] for principal in ids),
                        'rootless mapping can impersonate an exclusive/operator principal')
        require(set(v['files']) == set(TARGETS), 'complete finite policy transaction required')
        for item in v['files'].values():
            require(set(item) == {'before', 'after'} and
                    (item['before'] is None or hex_digest(item['before'])) and hex_digest(item['after']),
                    'policy preimage/postimage digest required')
        require(v['cohorts'] and v['copies'] and v['preserved'] and v['restart_units'] and v['trees'],
                'empty migration inventory')
        require(len(set(v['restart_units'])) == len(v['restart_units']), 'duplicate mask destinations')
        for unit in v['restart_units']:
            require(type(unit) is str and '/' not in unit and unit.endswith(('.service', '.socket', '.timer', '.path'))
                    and unit.replace('-', '').replace('@', '').replace('.', '').replace('_', '').isalnum(),
                    'invalid fixed restart unit')
            require(unit not in ('ssh.service', 'sshd.service', 'archipelago-c01-access.service'),
                    'operator/owner restart path cannot be retired')
        for role in ('manager', 'authority'):
            require(f'user@{v["accounts"][role]["uid"]}.service' in v['restart_units'],
                    'managed user manager must be masked')
        for field in ('broker_cgroup', 'supervisor_cgroup'):
            require(str(canonical(v[field])).startswith('/sys/fs/cgroup/') and v[field].endswith('.service'),
                     'fixed service cgroup required')
        if v['version']==3:
            import c01_migration_layout
            c01_migration_layout.validate(v['migration_layout'],self.launch,self.runtime)
        self.validate_staged()

    @property
    def names(self):
        return [self.value['accounts'][role]['name'] for role in ('runtime', 'manager', 'authority')]

    def validate_staged(self):
        v = self.value
        staged = {name: read(STAGED / name, True) for name in TARGETS}
        require(all(sha(raw) == v['files'][name]['after'] for name, raw in staged.items()),
                'staged policy digest changed')
        require(staged['sudoers'] == sudo_policy(v['operator']['name'], v.get('operator_sudo_authentication', 'password')) and
                staged['polkit'] == polkit_policy(self.names), 'finite privilege policy differs')
        ssh_baseline(staged['sshd'], self.names, v['files']['sshd']['before'])
        for name in ('cron-deny', 'at-deny'):
            require(set(self.names) <= set(staged[name].decode().splitlines()), 'job submission not denied')
        # Identity files are reviewed whole-file preimages, but only the three
        # managed accounts may change. Existing operator hashes are never logged.
        for name in ('passwd', 'shadow', 'group', 'gshadow'):
            current = read(TARGETS[name]).decode().splitlines()
            future = staged[name].decode().splitlines()
            def rows(lines):
                pairs = [line.split(':') for line in lines]
                require(len({row[0] for row in pairs}) == len(pairs), 'duplicate account row')
                return {row[0]: row for row in pairs}
            old, new = rows(current), rows(future)
            require(set(old) == set(new), 'access transition cannot create/delete accounts')
            if name in ('passwd', 'shadow'):
                require(all(row == new[key] for key, row in old.items() if key not in self.names),
                        'unrelated account/operator change')
            if name == 'passwd':
                for account in v['accounts'].values():
                    row = new[account['name']]
                    require(len(row) == 7 and row[:6] == old[account['name']][:6] and
                            (int(row[2]), int(row[3])) == (account['uid'], account['gid']) and
                            row[6] in ('/usr/sbin/nologin', '/sbin/nologin', '/bin/false'),
                            'account identity/home preservation or login lock failed')
                    require(sum(int(r[2]) == account['uid'] for r in new.values()) == 1, 'UID alias')
            elif name == 'shadow':
                for key in self.names:
                    require(len(new[key]) == 9 and new[key][1].startswith('!') and
                            new[key][1] in (old[key][1], '!' + old[key][1]) and
                            new[key][2:] == old[key][2:], 'managed password must be locked')
            else:
                for key, row in old.items():
                    require(len(row) == 4 and len(new[key]) == 4 and row[:2] == new[key][:2],
                            'group metadata changed')
                    if name == 'group':
                        require(row[2] == new[key][2], 'group ID changed')
                        fields = (3,)
                    else:
                        fields = (2, 3)
                    for field in fields:
                        require(new[key][field].split(',') ==
                                [x for x in row[field].split(',') if x not in self.names] or
                                (not new[key][field] and all(x in self.names for x in row[field].split(','))),
                                'only managed supplementary memberships may be removed')
        return staged

    def verify_files(self):
        for name, target in TARGETS.items():
            require(sha(read(target)) == self.value['files'][name]['after'], 'installed access policy changed: ' + name)
        self.generations.check()
        try:
            launch_digest=json_read(LAUNCH)[1]
        except FileNotFoundError:
            # The exact launch pathname can be absent between the two admitted
            # renames. Pending blocks all supervisor leases; runtime/data policy
            # is invariant and must not be revoked by that legitimate interval.
            require(self.generations.value['pending'] is not None,'active launch missing')
            launch_digest=None
        require(json_read(RUNTIME)[1] == self.value['runtime_sha256'] and
                (launch_digest in self.generations.allowed() or (launch_digest is None and self.generations.value['pending'] is not None)) and
                json_read(CONFIG)[1] == self.digest, 'protected configuration changed')

    def preflight_files(self):
        staged = self.validate_staged()
        for name, target in TARGETS.items():
            protected(target.parent, directory=True)
            expected = self.value['files'][name]['before']
            if expected is None:
                require(not target.exists() and not target.is_symlink(), 'new policy path already exists')
            else:
                require(sha(read(target)) == expected, 'policy preimage differs: ' + name)
        return staged

    def install_files(self):
        staged = self.preflight_files()
        # No rollback to old credentials/privileges after a partial operation.
        # A failure retains the pending journal and blocks every new admission.
        for name, target in TARGETS.items():
            expected = self.value['files'][name]['before']
            require(not os.path.lexists(target) if expected is None else sha(read(target)) == expected,
                    'policy preimage changed before replacement: ' + name)
            mode = 0o600 if name in ('shadow', 'gshadow') else 0o644
            if name == 'sudoers':
                mode = 0o440
            atomic(target, staged[name], mode)

    def preserved(self):
        required = {self.runtime[k] for k in ('home', 'graphroot', 'runtime_data_dir')}
        require(required <= {item['path'] for item in self.value['preserved']}, 'preservation roots missing')
        for item in self.value['preserved']:
            require(set(item) == {'path', 'identity'}, 'preserved locator schema')
            fd = open_path(item['path'], True)
            try:
                require(identity(os.fstat(fd)) == item['identity'], 'preserved root metadata changed')
            finally:
                os.close(fd)

    def verify_trees(self):
        """Exact root-reviewed effective-policy inventory; no symlink blind spots.

        Includes distro policy/plugin/unit/PAM/NSS/cron paths. This is an input
        integrity control, not a claim that arbitrary policy semantics were proven.
        Required semantic probes and kernel namespace checks are separate.
        """
        mandatory = {'/etc/sudo.conf', '/etc/nsswitch.conf', '/etc/pam.d', '/etc/ssh',
                     '/etc/polkit-1', '/usr/share/polkit-1/rules.d', '/etc/systemd/system',
                     '/etc/cron.d', '/var/spool/cron', '/etc/subuid', '/etc/subgid'}
        require(mandatory <= set(self.value['trees']), 'effective policy inventory incomplete')
        exclusions=self.generations.workspace_exclusions()
        for text, expected in self.value['trees'].items():
            path = canonical(text)
            require(path.is_relative_to('/etc') or path.is_relative_to('/usr') or
                    path == Path('/var/spool/cron'), 'unsupported policy inventory root')
            require(tree_digest(path,exclusions) == expected, 'policy inventory drift: ' + text)


def policy_link_target(parent, text):
    """Validate directory components before applying any parent traversal.

    Lexical abspath would erase `directory-alias/..` even though the kernel
    follows the alias first. Directory aliases are deliberately unsupported.
    Ordinary protected ../file links retain their real loader semantics.
    """
    cursor = Path('/') if text.startswith('/') else parent
    parts = text.split('/')
    for index, part in enumerate(parts):
        if part in ('', '.'):
            continue
        if part == '..':
            # The previous normal component was checked before reaching here.
            cursor = cursor.parent
        else:
            cursor = cursor / part
        if index < len(parts) - 1 and cursor != Path('/'):
            protected(cursor, directory=True)
    return cursor


def tree_digest(path, exclusions=()):
    """Canonical metadata+content digest; root-only writes; bounded, no links.

    /etc/systemd and distro trees often contain trusted links. They are recorded
    verbatim, and every link target chain must independently be root protected.
    Directory aliases are refused rather than silently omitting descendants.
    """
    rows = []
    def walk(item):
        if item in exclusions:
            return
        require(len(rows) < 16384, 'policy tree bound')
        meta = item.lstat()
        require(meta.st_uid == 0 and (stat.S_ISLNK(meta.st_mode) or not meta.st_mode & 0o022),
                'mutable effective policy')
        entry = [str(item.relative_to(path)), stat.S_IMODE(meta.st_mode), meta.st_uid, meta.st_gid]
        if stat.S_ISLNK(meta.st_mode):
            # Validate every lexical hop, not only the final resolved target.
            cursor = item
            closure = []
            visited = set()
            while cursor.is_symlink():
                require(str(cursor) not in visited and len(visited) < 32, 'policy alias cycle')
                visited.add(str(cursor))
                for parent in cursor.parents:
                    parent_meta = parent.lstat()
                    require(stat.S_ISDIR(parent_meta.st_mode) and parent_meta.st_uid == 0 and
                            not parent_meta.st_mode & 0o022, 'mutable policy alias ancestry')
                link_meta = cursor.lstat()
                require(link_meta.st_uid == 0, 'mutable policy alias')
                text = os.readlink(cursor)
                # Link inode allocation is not policy (new /dev/null masks must
                # be precomputable before installation). Pin every lexical hop,
                # ownership/mode/link count and text; pin final regular identity.
                closure.append([str(cursor), link_meta.st_uid, link_meta.st_gid,
                                stat.S_IMODE(link_meta.st_mode), link_meta.st_nlink, text])
                cursor = policy_link_target(cursor.parent, text)
            target = cursor
            require(not target.is_dir(), 'policy directory alias refused')
            # /dev/null masks are the sole special object admitted.
            if target != Path('/dev/null'):
                target_meta = protected(target)
                closure.append([str(target), identity(target_meta), sha(read(target))])
            else:
                protected(target.parent, directory=True)
                target_meta = target.lstat()
                require(stat.S_ISCHR(target_meta.st_mode) and target_meta.st_uid == 0 and
                        target_meta.st_rdev == os.makedev(1, 3), 'invalid null mask target')
                # devtmpfs inode/device allocation changes across boot. The
                # tightly identified null device's type/rdev/permissions do not.
                closure.append([str(target), target_meta.st_uid, target_meta.st_gid,
                                stat.S_IMODE(target_meta.st_mode), target_meta.st_rdev])
            entry += ['link-closure', closure]
        elif stat.S_ISDIR(meta.st_mode):
            entry += ['directory']
        else:
            entry += ['file', sha(read(item))]
        rows.append(entry)
        if stat.S_ISDIR(meta.st_mode):
            for child in sorted(item.iterdir()):
                walk(child)
    walk(path)
    return sha(json.dumps(rows, separators=(',', ':')).encode())