#!/usr/bin/env python3
"""Root filesystem qualification in disposable mounts; NOT signature/OTA acceptance.

The signature verifier and service query are explicit test doubles. All snapshot
creation, no-follow checks, file copying, replay checks and failure retention use
the production implementation against real files in isolated tmpfs mounts.
"""
import hashlib
import importlib.util
import io
import json
import os
from pathlib import Path
import shutil
import stat
import sys
import tarfile
import unittest
from unittest.mock import patch


def require_isolation():
    if os.geteuid() != 0 or os.environ.get('ARCHY_PUBLIC_PREPARE_FIXTURE') != '1':
        raise RuntimeError('root disposable namespace fixture required')
    mounts = {}
    for line in Path('/proc/self/mountinfo').read_text().splitlines():
        before, after = line.split(' - ', 1)
        mounts[before.split()[4]] = after.split()[0]
    for path in ['/etc', '/var/lib/archipelago', '/var/lib/archipelago-public-updates']:
        if mounts.get(path) != 'tmpfs':
            raise RuntimeError(f'refusing filesystem test without isolated tmpfs: {path}')


require_isolation()
spec = importlib.util.spec_from_file_location(
    'public_update_prepare', Path(__file__).resolve().parents[1] / 'public_update_prepare.py')
prepare = importlib.util.module_from_spec(spec)
spec.loader.exec_module(prepare)


class RootPreparationTests(unittest.TestCase):
    def setUp(self):
        for parent in [prepare.SNAPSHOTS, prepare.STAGING.parent]:
            for entry in parent.iterdir():
                if entry.is_dir() and not entry.is_symlink():
                    shutil.rmtree(entry)
                else:
                    entry.unlink()
        prepare.STAGING.mkdir()
        (prepare.STAGING / 'archipelago').write_bytes(b'fixture backend: never executed')
        self.archive = 'archipelago-frontend-2.0.4-alpha.tar.gz'
        with tarfile.open(prepare.STAGING / self.archive, 'w:gz') as archive:
            for name in ['index.html', 'archipelago-runtime/RELEASE-PAYLOAD.json']:
                member = tarfile.TarInfo(name)
                member.size = 2
                member.mode = 0o644
                archive.addfile(member, io.BytesIO(b'{}'))
        components = []
        for name in ['archipelago', self.archive]:
            raw = (prepare.STAGING / name).read_bytes()
            components.append({'name': name, 'new_version': '2.0.4-alpha',
                               'size_bytes': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()})
        self.document = {'version': '2.0.4-alpha', 'application_update': prepare.POLICY,
                         'signed_by': prepare.RELEASE_ROOT, 'signature': '0' * 128,
                         'components': components}
        self.raw = json.dumps(self.document).encode()

    def run_prepare(self):
        output = io.StringIO()
        service = ('LoadState=loaded\nUser=archipelago\n'
                   'ExecStart={ path=/usr/local/bin/archipelago ; '
                   'argv[]=/usr/local/bin/archipelago ; }\n')
        with patch.object(sys, 'argv', ['fixture']), \
                patch.object(sys, 'stdin', io.TextIOWrapper(io.BytesIO(self.raw))), \
                patch.object(sys, 'stdout', output), \
                patch.object(prepare, 'authenticate', return_value='2.0.3-alpha'), \
                patch.object(prepare.subprocess, 'check_output', return_value=service):
            prepare.main()
        return json.loads(output.getvalue())

    def test_snapshot_replay_and_source_mutation_do_not_change_admitted_bytes(self):
        result = self.run_prepare()
        ready = Path(result['path'])
        self.assertEqual((ready / 'signed-manifest.json').read_bytes(), self.raw)
        self.assertEqual(result['manifest_sha256'], hashlib.sha256(self.raw).hexdigest())
        for name in ['archipelago', self.archive, 'signed-manifest.json']:
            info = (ready / name).stat()
            self.assertEqual(info.st_uid, 0)
            self.assertEqual(stat.S_IMODE(info.st_mode), 0o444)
        (prepare.STAGING / 'archipelago').write_bytes(b'changed mutable staging')
        self.assertEqual(self.run_prepare(), result)
        self.assertEqual((ready / 'archipelago').read_bytes(), b'fixture backend: never executed')

    def test_existing_snapshot_corruption_is_not_repaired_from_mutable_staging(self):
        result = self.run_prepare()
        binary = Path(result['path']) / 'archipelago'
        binary.write_bytes(b'X' * binary.stat().st_size)  # root fault injection
        with self.assertRaisesRegex(ValueError, 'digest changed'):
            self.run_prepare()
        self.assertEqual(binary.read_bytes(), b'X' * binary.stat().st_size)

    def test_failed_preparation_retains_private_partial_and_can_retry_fresh(self):
        with patch.object(prepare, 'write_manifest', side_effect=OSError('injected fsync/write failure')):
            with self.assertRaisesRegex(OSError, 'injected'):
                self.run_prepare()
        partials = list(prepare.SNAPSHOTS.glob('.partial-*'))
        self.assertEqual(len(partials), 1)
        self.assertEqual(stat.S_IMODE(partials[0].stat().st_mode), 0o700)
        self.assertEqual(len(list(prepare.SNAPSHOTS.glob('[0-9a-f]' * 64))), 0)
        self.run_prepare()
        self.assertTrue(partials[0].exists())

    def test_lock_symlink_refuses_without_touching_destination(self):
        sentinel = prepare.STAGING / 'sentinel'
        sentinel.write_bytes(b'preserve')
        (prepare.SNAPSHOTS / '.prepare.lock').symlink_to(sentinel)
        with self.assertRaises(OSError):
            self.run_prepare()
        self.assertEqual(sentinel.read_bytes(), b'preserve')

    def test_dangling_owner_directory_refuses_without_preparing(self):
        marker = Path('/etc/archipelago-d09')
        marker.symlink_to('/nonexistent-fixture-owner')
        try:
            with self.assertRaisesRegex(ValueError, 'protected updater ownership'):
                self.run_prepare()
            self.assertEqual(list(prepare.SNAPSHOTS.iterdir()), [])
        finally:
            marker.unlink()


if __name__ == '__main__':
    unittest.main(verbosity=2)