#!/usr/bin/env python3
"""Separate-mounted synthetic volumes; no backend execution or installed data."""
import os
from pathlib import Path
import sys
import time
import traceback
from types import SimpleNamespace

sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts'))
import c01_views as views
import c01_supervisor as supervisor
from c01_children import Children, drop_credentials, process_group_path

assert os.geteuid() == 0 and os.environ.get('ARCHY_C01_VIEW_LOCATORS_FIXTURE') == '1'
assert Path(process_group_path(os.getpid())).parent.name.startswith('opencode-c01-view-locators-')
for directory in ('/run', '/var/lib/archipelago', '/var/lib/containers'):
    rows = [line.split() for line in Path('/proc/self/mountinfo').read_text().splitlines()]
    assert any(row[4] == directory and row[row.index('-') + 1] == 'tmpfs' for row in rows)

# Create NEW mounts after entering our own namespace. Systemd's outer tmpfs
# alone did not reproduce the detached pre-unshare mount's relative fd rendering.
outer = views.PrivateMounts()
volume = Path('/var/lib/archipelago/volume'); volume.mkdir()
assert views.LIBC.mount(b'tmpfs', os.fsencode(volume), b'tmpfs',
                       views.MS_NOSUID | views.MS_NODEV | views.MS_NOEXEC, b'size=8m,mode=700') == 0
os.chown(volume, 1000, 1000)
lnd = volume / 'lnd'; lnd.mkdir(mode=0o700); os.chown(lnd, 1000, 1000)
fds = []
def pin(path):
    fd = views.pin_directory(path); fds.append(fd)
    return views.PinnedSource.capture(fd, path)
source = pin(lnd)
volume_source = pin(volume)
before = (views.identity(source.fd), os.fstatvfs(source.fd).f_flag)
parent_namespace = views.namespace_identity()
targets = Path('/run/locator-targets'); targets.mkdir(mode=0o755)


def refused(action):
    try: action()
    except (OSError, ValueError): return
    raise AssertionError('unexpected source admission')


def case(name, action):
    owner = Children()
    try:
        def child(control):
            try:
                action()
                control.send(b'PASS')
            except BaseException:
                traceback.print_exc(); raise
        record = owner.spawn('application', child, keep=fds)
        record.channel.settimeout(15)
        assert record.channel.recv(16) == b'PASS', name
        end = time.monotonic() + 5
        while owner.children:
            assert owner.poll() == []
            assert time.monotonic() < end, name
            time.sleep(0.005)
        assert owner.results['application'] == 0
    finally:
        owner.request_stop()
        end = time.monotonic() + 5
        while owner.children:
            owner.poll()
            assert time.monotonic() < end, 'fixture retained owned child'
            time.sleep(0.005)
        owner.close()
    print('PASS ' + name, flush=True)


def mapped_volume():
    # A fork alone leaves the preparing namespace alive in the test parent.
    # Framework's single-process probe destroys its old namespace at unshare.
    # Give this child a sole-owner staging namespace and pin its own mount FD.
    views.PrivateMounts()
    local_fd = views.pin_directory(lnd)
    detached = views.PinnedSource.capture(local_fd, lnd)
    view = views.PrivateMounts()
    rendered = os.readlink(f'/proc/self/fd/{local_fd}')
    assert rendered == '/lnd', ('separate-volume reproducer did not detach the FD path', rendered)
    destination = targets / 'readonly'; destination.mkdir()
    view.idmapped_directory(detached, destination, 1000, 60002, readonly=True)
    meta = destination.stat()
    assert (meta.st_dev, meta.st_ino, meta.st_uid, meta.st_gid, meta.st_mode & 0o7777) == (
        source.expected[0], source.expected[1], 60002, 60002, source.expected[4])
    assert os.statvfs(destination).f_flag & (views.PROTECTIONS) == source.protections | os.ST_RDONLY
    assert (views.identity(source.fd), os.fstatvfs(source.fd).f_flag) == before
    os.close(local_fd)
case('separate-mounted-volume-relative-fd-rendering-uses-preunshare-authenticated-locator', mapped_volume)


def hidden():
    view = views.PrivateMounts()
    view.cover(lnd, 0o700)
    destination = targets / 'hidden'; destination.mkdir()
    refused(lambda: view.idmapped_directory(source, destination, 1000, 60002))
case('covered-source-refused-despite-retained-old-fd', hidden)

for name in ('renamed', 'replaced', 'unlinked', 'symlink'):
    path = volume / name; path.mkdir(mode=0o700); os.chown(path, 1000, 1000)
    selected = pin(path)
    def mutate(name=name, path=path, selected=selected):
        view = views.PrivateMounts()
        if name == 'unlinked':
            path.rmdir()
        else:
            path.rename(volume / (name + '-original'))
        if name in ('replaced', 'unlinked'):
            path.mkdir(mode=0o700); os.chown(path, 1000, 1000)
        if name == 'symlink':
            path.symlink_to(volume / (name + '-original'))
        destination = targets / name; destination.mkdir()
        refused(lambda: view.idmapped_directory(selected, destination, 1000, 60002))
        assert destination.stat().st_uid == 0
    case(name + '-source-refused-without-fd-name-fallback', mutate)

other = volume / 'other'; other.mkdir(mode=0o700); os.chown(other, 1000, 1000)
refused(lambda: views.PinnedSource.capture(source.fd, other))
link = volume / 'intermediate'; link.symlink_to(volume)
refused(lambda: views.PinnedSource.capture(source.fd, link / 'lnd'))
print('PASS capture-rejects-substituted-inode-and-intermediate-symlink', flush=True)

def late_capture():
    view = views.PrivateMounts()
    late = views.PinnedSource.capture(source.fd, lnd)
    destination = targets / 'late'; destination.mkdir()
    refused(lambda: view.idmapped_directory(late, destination, 1000, 60002))
case('locator-captured-after-namespace-change-refused', late_capture)

# Equal inode/UID/mode is insufficient if the reopened mount weakened flags.
alias = Path('/var/lib/archipelago/restricted-alias'); alias.mkdir()
assert views.LIBC.mount(os.fsencode(lnd), os.fsencode(alias), None, 4096, None) == 0
assert views.LIBC.mount(None, os.fsencode(alias), None, 4096 | 32 | 1 | 2 | 4 | 8, None) == 0
restricted = pin(alias)
restricted_flags = os.statvfs(alias).f_flag
def weakened():
    view = views.PrivateMounts()
    assert views.LIBC.mount(None, os.fsencode(alias), None, 4096 | 32 | 2 | 4 | 8, None) == 0
    assert views.identity(restricted.fd) == restricted.expected
    assert not os.statvfs(alias).f_flag & os.ST_RDONLY
    destination = targets / 'weakened'; destination.mkdir()
    refused(lambda: view.idmapped_directory(restricted, destination, 1000, 60002))
case('same-inode-reopened-mount-with-weaker-protections-refused', weakened)
assert os.statvfs(alias).f_flag == restricted_flags

def changed_mode():
    view = views.PrivateMounts()
    destination = targets / 'changed-mode'; destination.mkdir()
    try:
        lnd.chmod(0o750)  # This is synthetic fixture data only.
        refused(lambda: view.idmapped_directory(source, destination, 1000, 60002))
    finally:
        lnd.chmod(0o700)
case('pinned-source-mode-change-refused', changed_mode)

# Exercise actual Prepared.role_views callers with a synthetic, NEVER executed
# proxy file on a second separate mount. No old backend/proxy artifact is used.
prepared = supervisor.Prepared.__new__(supervisor.Prepared)
prepared.fds = []
prepared.value = {'runtime_uid': 1000, 'runtime_gid': 1000, 'runtime_data_dir': str(volume)}
prepared.backings, prepared.backing_sources = {}, {}
private = Path('/var/lib/containers/private'); private.mkdir(mode=0o700)
for name in ('manager', 'snapshot', 'authority'):
    path = private / name; path.mkdir(mode=0o700); os.chown(path, 1000, 1000)
    cap = pin(path)
    prepared.backings[name], prepared.backing_sources[name] = cap.fd, cap
prepared.runtime, prepared.runtime_source = volume_source.fd, volume_source
payload = Path('/var/lib/containers/payload'); payload.mkdir()
assert views.LIBC.mount(b'tmpfs', os.fsencode(payload), b'tmpfs', 2 | 4, b'size=1m,mode=755') == 0
proxy = payload / 'proxy'; proxy.touch(mode=0o555)
prepared.proxy = os.open(proxy, os.O_PATH | os.O_CLOEXEC); fds.append(prepared.proxy)
prepared.proxy_source = views.PinnedSource.capture(prepared.proxy, proxy)
proxy_identity = views.identity(prepared.proxy)
def composed_role():
    views.PrivateMounts()
    # Capture mounts owned by this sole-process staging namespace; role_views
    # will destroy it at its own unshare, like the failing real-volume probe.
    for name, cap in list(prepared.backing_sources.items()):
        local = cap.reopen()
        prepared.backings[name] = local
        prepared.backing_sources[name] = views.PinnedSource.capture(local, cap.locator)
    prepared.runtime = prepared.runtime_source.reopen()
    prepared.runtime_source = views.PinnedSource.capture(prepared.runtime, volume)
    prepared.proxy = prepared.proxy_source.reopen()
    prepared.proxy_source = views.PinnedSource.capture(prepared.proxy, proxy)
    prepared.role_views(SimpleNamespace(pw_uid=60001, pw_gid=60001), 'application')
    assert os.readlink(f'/proc/self/fd/{prepared.proxy}') == '/proxy'
    drop_credentials(60001, 60001, os.getppid())
    assert (supervisor.BASE / 'manager').stat().st_uid == 60001
    assert (volume / 'lnd').stat().st_uid == 60001
    alias = supervisor.BASE / 'bin/podman'
    assert (alias.stat().st_dev, alias.stat().st_ino) == proxy_identity[:2]
    assert os.statvfs(alias).f_flag & os.ST_RDONLY
    refused(lambda: (private / 'authority').stat())
case('prepared-role-views-and-proxy-alias-use-captured-separate-mount-locators', composed_role)

assert views.namespace_identity() == parent_namespace
assert (views.identity(source.fd), os.fstatvfs(source.fd).f_flag) == before
assert views.identity(prepared.proxy) == proxy_identity
assert not Path(f'/proc/self/task/{os.getpid()}/children').read_text().strip()
for fd in fds: os.close(fd)
print('PASS parent-mount-flags-metadata-and-lifecycle-preserved', flush=True)
# No recursive cleanup: private mounts die with this disposable process/unit.