#!/usr/bin/python3
"""Actual system-manager D09 qualification inside an explicitly named disposable container.

Uses inert Python executables, not deployment artifacts. Never run on a node.
No mock replaces systemctl, Unix peer credentials, or the production owner/worker.
"""
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import pwd
import secrets
import shutil
import socket
import subprocess
import sys
import time

assert os.geteuid() == 0
assert socket.gethostname() == 'd09-qualification'
assert Path('/run/.containerenv').exists()
assert os.environ.get('D09_DISPOSABLE_SYSTEMD') == '1'
PACKAGE = Path('/opt/d09-fixture')
# podman cp retains source-directory group-write bits. The provisioner rightly
# refuses those; install the disposable package as protected code first.
for directory, directories, files in os.walk(PACKAGE):
    for path in [Path(directory), *(Path(directory) / name for name in files)]:
        path.chmod(path.stat().st_mode & 0o755)
sys.path.insert(0, str(PACKAGE / 'scripts'))
import d09_wire as wire
import d09_store as store
spec = importlib.util.spec_from_file_location('provision', PACKAGE / 'scripts/provision-d09-lifecycle.py')
provision = importlib.util.module_from_spec(spec)
spec.loader.exec_module(provision)
owner = provision.owner
assert not owner.BINDING.exists() and not (owner.ROOT / 'ledger.json').exists(), 'retain previous authority evidence; use a fresh container'


def run(*args, check=True):
    return subprocess.run(args, check=check, capture_output=True, text=True, timeout=45)


def wait_for(predicate, label, seconds=45):
    end = time.monotonic() + seconds
    while time.monotonic() < end:
        value = predicate()
        if value:
            return value
        time.sleep(0.1)
    raise AssertionError('timed out: ' + label)


def write(path, data, mode=0o644):
    path = Path(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(data.encode() if isinstance(data, str) else data)
    path.chmod(mode)


def config_digest():
    return wire.digest(owner.effective_configuration().encode())


def sha(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()


def ledger():
    return json.loads((owner.ROOT / 'ledger.json').read_text())


def receipt_phase(phase):
    value = ledger()
    receipt = value.get('receipt')
    if receipt and receipt['phase'] == 'CONFLICT':
        raise AssertionError('owner entered conflict: ' + json.dumps(receipt))
    return receipt if receipt and receipt['phase'] == phase else None


# Every changed pathname below belongs to this disposable container's overlay.
run('systemctl', 'stop', 'archipelago.service', check=False)
assert owner.service()['MainPID'] == '0'
for path in (owner.ROOT, owner.BINDING.parent, Path('/etc/systemd/system/archipelago.service.d'),
             Path('/opt/archipelago/web-ui')):
    if path.exists():
        shutil.rmtree(path)
write('/etc/sudoers', 'root ALL=(ALL:ALL) ALL\n', 0o440)
if Path('/etc/sudoers.d').exists():
    shutil.rmtree('/etc/sudoers.d')
Path('/etc/sudoers.d').mkdir(mode=0o755)
account = pwd.getpwnam('archipelago')
control = Path('/run/d09-fixture-client')
control.mkdir(mode=0o700, exist_ok=True)
os.chown(control, account.pw_uid, account.pw_gid)
home = Path('/home/archipelago')
home.mkdir(exist_ok=True)
os.chown(home, account.pw_uid, account.pw_gid)
cache = Path('/var/lib/archipelago/d09-requests')
cache.mkdir(parents=True, exist_ok=True, mode=0o700)
os.chown(cache, account.pw_uid, account.pw_gid)

requester = '''import json,os,pathlib,subprocess,time
p=pathlib.Path('/run/d09-fixture-client')
while True:
 for request in p.glob('request-*.json'):
  output=request.with_name(request.name.replace('request-','result-'))
  if output.exists():continue
  args=json.loads(request.read_text())
  r=subprocess.run(['/usr/libexec/archipelago-d09ctl',*args],capture_output=True,text=True,timeout=15)
  output.write_text(json.dumps({'exit':r.returncode,'stdout':r.stdout,'stderr':r.stderr}))
 time.sleep(.1)
'''
write(PACKAGE / 'requester.py', requester)
python = Path(sys.executable).resolve().read_bytes()
write(owner.DESTINATIONS['backend'], python, 0o755)
web = owner.DESTINATIONS['frontend']
web.parent.mkdir(parents=True, exist_ok=True)
os.chown(web.parent, 0, 0)
web.parent.chmod(0o755)
write(web / 'index.html', 'old fixture frontend\n')
template = Path('archipelago-runtime/image-recipe/configs/nginx-archipelago.conf')
write(web / template, (PACKAGE / 'image-recipe/configs/nginx-archipelago.conf').read_bytes())
write('/etc/systemd/system/archipelago.service', '''[Unit]
Description=Disposable D09 requester
[Service]
Type=exec
User=archipelago
ExecStart=/usr/local/bin/archipelago /opt/d09-fixture/requester.py
NoNewPrivileges=yes
ProtectSystem=strict
ProtectHome=read-only
ReadWritePaths=/run/d09-fixture-client /var/lib/archipelago/d09-requests
Restart=no
''')
preserved_paths = [home / 'Work/operator.txt',
                   home / '.local/state/archipelago/sessions/history.txt',
                   Path('/etc/systemd/system/archipelago.service.d/90-operator.conf')]
for path in preserved_paths[:2]:
    write(path, 'retained operator bytes\n', 0o600)
    for parent in [path, *path.parents]:
        if parent == home.parent:
            break
        os.chown(parent, account.pw_uid, account.pw_gid)
write(preserved_paths[2], '[Service]\nEnvironment=D09_OPERATOR_FIXTURE=preserved\n')
preserved_before = [(str(path), path.stat().st_ino, path.stat().st_mode,
                     path.stat().st_uid, path.stat().st_gid, sha(path)) for path in preserved_paths]
run('systemctl', 'daemon-reload')
provision.install_package()

# Measure actual denied writes and sudo before admitting any transaction.
denials = []
for args in (['touch', '/usr/local/bin/d09-unauthorized'],
             ['touch', str(web / 'unauthorized')],
             ['python3', '-c', "import os;os.rename('/opt/archipelago/web-ui','/opt/archipelago/unauthorized-move')"],
             ['sudo', '-n', 'true']):
    result = run('runuser', '-u', 'archipelago', '--', *args, check=False)
    assert result.returncode != 0, args
    denials.append({'operation': args[0], 'exit': result.returncode})
exclusion = wire.canonical({'disposable': True, 'observed_denials': denials})
store.atomic(owner.ROOT / 'incoming/exclusion.json', exclusion)
new_config = (PACKAGE / 'image-recipe/configs/60-terminal-workspace.conf').read_bytes()
for generation in ('new', 'restore'):
    destination = owner.ROOT / 'incoming' / (generation + '-frontend')
    shutil.copytree(web, destination)
    if generation == 'new':
        write(destination / 'index.html', 'new fixture frontend\n')
    write(owner.ROOT / 'incoming' / (generation + '-backend'),
          python + (b'\nD09-inert-fixture-generation-two\n' if generation == 'new' else b''), 0o755)
write(owner.ROOT / 'incoming/new-config', new_config)
# Old/restored config is absent, rather than an invented empty file.
restore_digest = config_digest()
write(owner.DESTINATIONS['config'], new_config)
run('systemctl', 'daemon-reload')
apply_digest = config_digest()
owner.DESTINATIONS['config'].unlink()
run('systemctl', 'daemon-reload')
assert config_digest() == restore_digest
binding = dict(schema_version=1, target_binding_digest=secrets.token_hex(32),
    composition_digest=secrets.token_hex(32), admission_epoch_id=secrets.token_hex(16),
    transaction_id=secrets.token_hex(16), grants={route: secrets.token_hex(32)
        for route in ('UPDATER', 'AUTOMATIC', 'MANUAL', 'GUARD', 'OBSERVER')},
    objects={name: {'old': owner.object_digest(destination),
        'new': owner.object_digest(owner.ROOT / 'incoming' / ('new-' + name)),
        'restore': owner.object_digest(owner.ROOT / 'incoming' / ('restore-' + name))}
        for name, destination in owner.DESTINATIONS.items()},
    artifacts={'backend': sha(owner.ROOT / 'incoming/new-backend'),
        'frontend': owner.object_digest(owner.ROOT / 'incoming/new-frontend')},
    restore_artifacts={'backend': sha(owner.DESTINATIONS['backend']),
        'frontend': owner.object_digest(web)},
    effective_config={'APPLY': apply_digest, 'RESTORE': restore_digest},
    exclusion_evidence_digest=wire.digest(exclusion),
    helper_digest=sha('/usr/libexec/archipelago-d09-prepare-workspace'),
    prerequisite={'grant_digest': secrets.token_hex(32),
        'backend_sha256': sha(owner.DESTINATIONS['backend']), 'effective_config_digest': restore_digest})
binding['release_identity'] = ['B', [['archipelago', 'B', binding['artifacts']['backend'], (owner.ROOT / 'incoming/new-backend').stat().st_size]]]
admission = owner.ROOT / 'incoming/admission.json'
store.atomic(admission, wire.canonical(binding))
provision.provision(admission)
run('systemctl', 'start', 'archipelago-d09-lifecycle.service')
wait_for(lambda: owner.PRIVATE.exists() and owner.SOCKET.exists(), 'owner sockets')
def prerequisite_ready():
    result=provision.activate_prerequisite()
    return result if result['prerequisite_started'] else None
started = wait_for(prerequisite_ready, 'one submitted prerequisite READY')
assert started['prerequisite_started']
first_invocation = started['invocation_id']


def history_fields():
    v = ledger()
    active = v.get('active_binding', binding)
    index = {k: active[k] for k in ('target_binding_digest', 'composition_digest', 'transaction_id')}
    return dict(index, predecessor_epoch_id=v['epoch'], head_sequence=v['sequence'],
        preserved_head_digest=wire.digest(wire.canonical(v['receipt'])) if v['receipt'] else wire.digest((owner.ROOT / 'ledger.json').read_bytes()),
        preserved_journal_digest=wire.digest(wire.canonical({'receipts': v['journal'], 'intents': v['intents']})),
        preserved_consumed_history_digest=wire.digest(wire.canonical(owner.consumption_history(v))),
        preserved_target_index_digest=wire.digest(wire.canonical(index)))


def external_record(fields):
    # Disposable coordinator fixture, separate from the owner's rollback tree.
    # This qualifies source binding, NOT operational external history transport.
    directory = PACKAGE / 'external-history'
    directory.mkdir(exist_ok=True, mode=0o700)
    identity = secrets.token_hex(16)
    history = dict(fields, sequence=1)
    history_path = directory / (identity + '-history.json')
    store.atomic(history_path, wire.canonical(history))
    record = dict(fields, schema_version=1, external_history_sequence=1,
        external_history_digest=sha(history_path), observed_boot_id=owner.boot_id())
    path = directory / (identity + '-record.json')
    store.atomic(path, wire.canonical(record))
    return path, history_path


def restart_reactivate(action):
    before = (owner.ROOT / 'ledger.json').read_bytes()
    run('systemctl', 'restart', 'archipelago-d09-lifecycle.service')
    wait_for(lambda: owner.service()['MainPID'] == '0' and owner.service()['ActiveState'] == 'inactive', 'restart drain')
    def unknown():
        result = run('/usr/libexec/archipelago-d09ctl', 'inspect', check=False)
        return result.returncode == 0 and json.loads(result.stdout)['state'] == 'UNKNOWN'
    wait_for(unknown, 'fresh owner holds unknown')
    assert (owner.ROOT / 'ledger.json').read_bytes() == before
    probe = owner.Owner.__new__(owner.Owner)
    disposition = wait_for(lambda: probe.drained_disposition(ledger()), 'worker drain')
    fields = dict(history_fields(), allowed_action=action, new_epoch_id=secrets.token_hex(16),
        new_grant_digest=secrets.token_hex(32), execution_grant_digest=secrets.token_hex(32), disposition_digest=disposition)
    record, history = external_record(fields)
    provision.readmit(record, history, activation=True)
    wait_for(lambda: owner.service()['ActiveState'] == 'active', 'external reactivation')
    if action == 'REACTIVATE_RETAINED':
        wait_for(lambda: receipt_phase('AWAITING_OBSERVATION'), 'fresh observation required')
    else:
        wait_for(lambda: ledger()['prerequisite']['state']=='STARTED', 'prerequisite observed READY')
    invocation = owner.service()['InvocationID']
    # Lost acknowledgement retry consumes neither another grant nor activation.
    provision.readmit(record, history, activation=True)
    assert owner.service()['InvocationID'] == invocation
    return invocation


first_invocation = restart_reactivate('REACTIVATE_PREREQUISITE')
assert ledger()['prerequisite_history'][0]['grant_digest'] == binding['prerequisite']['grant_digest']
assert ledger()['sequence'] == 0
print('PASS sequence-zero owner loss: external reconciliation retains consumed grant A, starts once, retry nonactivating', flush=True)


def request(arguments):
    identity = secrets.token_hex(8)
    write(control / ('request-' + identity + '.json'), json.dumps(arguments), 0o644)
    response = control / ('result-' + identity + '.json')
    wait_for(response.exists, 'request ' + arguments[0])
    value = json.loads(response.read_text())
    assert value['exit'] == 0, value
    return json.loads(value['stdout'])


initial_status = request(['inspect'])
assert initial_status['state'] == 'NO_D09_TRANSACTION', initial_status
selected = json.dumps(binding['release_identity'], separators=(',', ':'))
before_wrong = (owner.ROOT / 'ledger.json').read_bytes()
wrong = run('/usr/libexec/archipelago-d09ctl', 'prepare', '--selected-release', json.dumps(['C', binding['release_identity'][1]]), check=False)
assert wrong.returncode != 0 and (owner.ROOT / 'ledger.json').read_bytes() == before_wrong
request(['prepare', '--selected-release', selected, '--staging-generation', secrets.token_hex(16)])
applied = wait_for(lambda: receipt_phase('AWAITING_OBSERVATION'), 'actual apply')
assert owner.service()['InvocationID'] != first_invocation
assert sha(owner.DESTINATIONS['backend']) == binding['artifacts']['backend']
assert (web / 'index.html').read_text() == 'new fixture frontend\n'
assert owner.DESTINATIONS['config'].read_bytes() == new_config
assert request(['inspect'])['state'] == 'PENDING'
print('PASS real systemd apply: requester cgroup, external owner/worker, exact payload/config pair, pending until independent observation', flush=True)


def complete_fixture_observation():
    r = receipt_phase('AWAITING_OBSERVATION')
    assert r
    pid = owner.service()['MainPID']
    # OS PTY surrogate in the actual mount namespace. This is deliberately not
    # native terminal integration/UAT; the evidence records that qualification.
    run('nsenter', '-t', pid, '-m', '--', 'runuser', '-u', 'archipelago', '--',
        'bash', '-c', 'cd /home/archipelago/Work && script -q -c "pwd" /dev/null')
    for path in ('/home/archipelago/Work/d09-observer-probe', '/home/archipelago/.local/state/archipelago/sessions/d09-observer-probe'):
        run('nsenter', '-t', pid, '-m', '--', 'runuser', '-u', 'archipelago', '--', 'touch', path)
    for path in ('/usr/local/bin/d09-observer-denied', '/opt/archipelago/web-ui/d09-observer-denied'):
        assert run('nsenter', '-t', pid, '-m', '--', 'runuser', '-u', 'archipelago', '--', 'touch', path, check=False).returncode != 0
    after = [(str(path), path.stat().st_ino, path.stat().st_mode, path.stat().st_uid, path.stat().st_gid, sha(path)) for path in preserved_paths]
    assert after == preserved_before
    base = dict(transaction_id=r['transaction_id'], observed_invocation_id=r['observed_invocation_id'])
    namespace = dict(base, qualification='inert executable / OS PTY surrogate; NOT native terminal',
        default_terminal_created=True, default_terminal_workspace='/home/archipelago/Work',
        namespace_writable=['/home/archipelago/Work', '/home/archipelago/.local/state/archipelago/sessions'])
    preservation = dict(base, preserved={name: {'before': wire.digest(wire.canonical(before)), 'after': wire.digest(wire.canonical(now))}
        for name, before, now in zip(('work', 'history', 'custom_fragments'), preserved_before, after)})
    def publish(value):
        raw = wire.canonical(value)
        hashed = wire.digest(raw)
        store.atomic(owner.ROOT / 'observations' / (hashed + '.json'), raw)
        return hashed
    observation = {k: r[k] for k in ('transaction_id', 'target_binding_digest', 'composition_digest',
        'observed_invocation_id', 'effective_config_observation_digest', 'config_current_digest')}
    observation.update(namespace_observation_digest=publish(namespace), preservation_observation_digest=publish(preservation))
    public = json.loads(owner.BINDING.with_name('client.json').read_text())
    request_value = dict(schema_version=1, operation='ACK_OBSERVATION', route='OBSERVER', reason='OBSERVATION_READY',
        transaction_id=r['transaction_id'], request_id=secrets.token_hex(16), target_binding_digest=r['target_binding_digest'],
        composition_digest=r['composition_digest'], grant_digest=public['grants']['OBSERVER'], expected_sequence=ledger()['sequence'],
        requester_invocation_id=None, observer_receipt_digest=publish(observation), selected_release_digest=None, staging_generation=None)
    with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
        conn.connect(str(owner.SOCKET))
        conn.sendall(wire.canonical(request_value))
        conn.shutdown(socket.SHUT_WR)
        assert wire.parse(conn.recv(wire.MAX))['status'] == 'ACCEPTED'
    wait_for(lambda: receipt_phase('COMPLETE'), 'fixture observer completion')


complete_fixture_observation()
restart_reactivate('REACTIVATE_RETAINED')
complete_fixture_observation()
print('PASS applied COMPLETE owner-loss reactivation: exact B retained, new invocation, fresh OS-surrogate observation', flush=True)

# Admit B -> C without resetting any ledger/journal or predecessor snapshots.
old_request = next(e['request'] for e in ledger()['requests'].values() if e['request']['operation'] == 'APPLY')
predecessor_receipts = list(ledger()['journal'])
incoming = owner.ROOT / 'incoming'
retained_incoming = PACKAGE / 'incoming-retained-B'
retained_incoming.mkdir(mode=0o700)
for entry in incoming.iterdir():
    entry.rename(retained_incoming / entry.name)
store.atomic(incoming / 'exclusion.json', exclusion)
for generation in ('new', 'restore'):
    for name, destination in owner.DESTINATIONS.items():
        target = incoming / (generation + '-' + name)
        if destination.is_dir():
            shutil.copytree(destination, target)
        elif destination.exists():
            shutil.copy2(destination, target)
write(incoming / 'new-backend', python + b'\nD09-inert-fixture-generation-three\n', 0o755)
write(incoming / 'new-frontend/index.html', 'third fixture frontend\n')
successor_binding = dict(binding, transaction_id=secrets.token_hex(16), admission_epoch_id=secrets.token_hex(16),
    composition_digest=secrets.token_hex(32), grants={route: secrets.token_hex(32) for route in binding['grants']},
    objects={name: dict(old=owner.object_digest(dest), new=owner.object_digest(incoming / ('new-' + name)),
        restore=owner.object_digest(incoming / ('restore-' + name))) for name, dest in owner.DESTINATIONS.items()},
    artifacts=dict(backend=sha(incoming / 'new-backend'), frontend=owner.object_digest(incoming / 'new-frontend')),
    restore_artifacts=binding['artifacts'], effective_config=dict(APPLY=apply_digest, RESTORE=apply_digest),
    prerequisite=dict(grant_digest=secrets.token_hex(32), backend_sha256=sha(owner.DESTINATIONS['backend']), effective_config_digest=apply_digest))
successor_binding['release_identity'] = ['C', [['archipelago', 'C', successor_binding['artifacts']['backend'], (incoming / 'new-backend').stat().st_size]]]
successor_admission = incoming / 'admission.json'
store.atomic(successor_admission, wire.canonical(successor_binding))
protected = dict(successor_binding, requester_uid=account.pw_uid, requester_gid=account.pw_gid)
fields = dict(history_fields(), successor_binding_digest=wire.digest(wire.canonical(protected)), terminal_disposition='APPLY')
record, history = external_record(fields)
wait_for(lambda: owner.command('/usr/bin/systemctl', 'show', ledger()['execution']['unit'], '-p', 'ActiveState', '--value') == 'inactive', 'completed worker drain')
provision.successor(successor_admission, record, history)
assert ledger()['journal'] == predecessor_receipts
assert binding['transaction_id'] in ledger()['predecessors']
before_replay = (owner.ROOT / 'ledger.json').read_bytes()
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as conn:
    conn.connect(str(owner.SOCKET))
    conn.sendall(wire.canonical(old_request))
    conn.shutdown(socket.SHUT_WR)
    assert wire.parse(conn.recv(wire.MAX))['status'] == 'REFUSED'
assert before_replay == (owner.ROOT / 'ledger.json').read_bytes()
provision.successor(successor_admission, record, history)
assert before_replay == (owner.ROOT / 'ledger.json').read_bytes()
restart_reactivate('REACTIVATE_PREREQUISITE')
request(['prepare', '--selected-release', json.dumps(successor_binding['release_identity']), '--staging-generation', secrets.token_hex(16)])
wait_for(lambda: receipt_phase('AWAITING_OBSERVATION'), 'successor C apply')
assert sha(owner.DESTINATIONS['backend']) == successor_binding['artifacts']['backend']
complete_fixture_observation()
binding = successor_binding
print('PASS A -> B -> C successor admission: terminal predecessor, retained journal, exact B rollback provenance, old APPLY and admission replay nonactivating', flush=True)
request(['recover', '--reason', 'manual'])
restored = wait_for(lambda: receipt_phase('AWAITING_OBSERVATION')
    if ledger().get('receipt', {}).get('config_action') == 'RESTORE' else None, 'actual recovery')
assert sha(owner.DESTINATIONS['backend']) == binding['restore_artifacts']['backend']
assert (web / 'index.html').read_text() == 'new fixture frontend\n'
assert owner.DESTINATIONS['config'].read_bytes() == new_config
assert restored['frontend_restoration'] == restored['backend_restoration'] == 'SUCCEEDED'
print('PASS C -> B recovery: exact predecessor pair/config restored', flush=True)
restart_reactivate('REACTIVATE_RETAINED')
complete_fixture_observation()
restart_reactivate('REACTIVATE_RETAINED')
complete_fixture_observation()
print('PASS restored AWAITING and restored COMPLETE reactivation: one-use authority, new invocations, fresh observations', flush=True)
preserved_after = [(str(path), path.stat().st_ino, path.stat().st_mode,
                    path.stat().st_uid, path.stat().st_gid, sha(path)) for path in preserved_paths]
assert preserved_after == preserved_before
print('PASS actual Work, history and custom-unit bytes/inodes/modes/owners preserved', flush=True)

# A fresh owner must not turn an on-disk journal into renewed execution rights.
# Exercise the real BindsTo hold and authenticated Unix inspect after restart.
ledger_before_restart = (owner.ROOT / 'ledger.json').read_bytes()
run('systemctl', 'restart', 'archipelago-d09-lifecycle.service')
wait_for(lambda: owner.service()['ActiveState'] == 'inactive', 'owner restart holds requester')
def restarted_inspection():
    # The retained pathname can precede the new listener. Probe the actual
    # authenticated, read-only exchange, not socket-file existence.
    result = run('/usr/libexec/archipelago-d09ctl', 'inspect', check=False)
    return json.loads(result.stdout) if result.returncode == 0 else None
unknown = wait_for(restarted_inspection, 'restarted owner authenticated inspect')
assert unknown['state'] == 'UNKNOWN', unknown
assert (owner.ROOT / 'ledger.json').read_bytes() == ledger_before_restart
assert sha(owner.DESTINATIONS['backend']) == binding['restore_artifacts']['backend']
assert (web / 'index.html').read_text() == 'new fixture frontend\n'
print('PASS actual owner restart: no disk-seeded authority, requester held, journal and recovered payloads retained', flush=True)