"""Pure access-owner regressions: no accounts, service bus or cgroup operations.

Kernel/lifecycle tests live in access_owner_regressions and require the guarded
disposable fixture. Run this file with python3 -B -m unittest after source freeze.
"""
import copy
from pathlib import Path
import sys
import types
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import c01_access as access
import c01_access_policy as policy


class SudoPreservationTests(unittest.TestCase):
    def test_default_still_requires_password(self):
        self.assertNotIn(b'NOPASSWD', policy.sudo_policy('archipelago-admin'))

    def test_explicit_choice_only_grants_separate_administrator(self):
        rendered = policy.sudo_policy('archipelago-admin', 'preserve-verified-noninteractive')
        self.assertEqual(rendered,
            b'Defaults env_reset\nDefaults use_pty\nroot ALL=(ALL:ALL) ALL\n'
            b'archipelago-admin ALL=(ALL:ALL) NOPASSWD: ALL\n')
        for name in ('archipelago', 'c01-manager', 'c01-authority', '%sudo', 'root'):
            with self.subTest(name=name), self.assertRaises(ValueError):
                policy.sudo_policy(name, 'preserve-verified-noninteractive')

    def test_unknown_authentication_choice_refused(self):
        for choice in ('', 'nopasswd', True, None):
            with self.subTest(choice=choice), self.assertRaises(ValueError):
                policy.sudo_policy('archipelago-admin', choice)


class SSHPolicyTests(unittest.TestCase):
    names = ['archipelago', 'c01-manager', 'c01-authority']

    def test_real_openssh_repeated_rows_preserve_all_denials_and_admin_policy(self):
        baseline = b'PubkeyAuthentication yes\n'
        obj = types.SimpleNamespace(names=self.names, value={
            'operator': {'name': 'archipelago-admin'}, 'files': {'sshd': {'before': policy.sha(baseline)}}})
        original = 'pubkeyauthentication yes\nallowusers archipelago-admin\nallowusers other-admin\n'
        denial = ''.join('denyusers ' + name + '\n' for name in self.names)
        installed = False
        def command(argv):
            if '-T' not in argv:
                return 0, ''
            return 0, original if not installed and str(policy.TARGETS['sshd']) in argv else denial + original
        with patch.object(policy, 'read', return_value=policy.ssh_policy(baseline, self.names)), \
                patch.object(access, 'fixed_command', side_effect=command):
            access.ssh_checks(obj, staged=True)
            installed = True
            access.ssh_checks(obj)
        self.assertEqual(access.ssh_effective(denial)['denyusers'], self.names)

    def test_repeated_setting_first_row_drift_cannot_hide_behind_same_last_row(self):
        baseline = b'PubkeyAuthentication yes\n'
        obj = types.SimpleNamespace(names=self.names, value={
            'operator': {'name': 'archipelago-admin'}, 'files': {'sshd': {'before': policy.sha(baseline)}}})
        def command(argv):
            if '-T' not in argv:
                return 0, ''
            if str(policy.TARGETS['sshd']) in argv:
                return 0, 'allowusers archipelago-admin\nallowusers other-admin\n'
            return 0, (''.join('denyusers ' + name + '\n' for name in self.names) +
                       'allowusers attacker\nallowusers other-admin\n')
        with patch.object(policy, 'read', return_value=policy.ssh_policy(baseline, self.names)), \
                patch.object(access, 'fixed_command', side_effect=command), self.assertRaises(ValueError):
            access.ssh_checks(obj, staged=True)

    def test_preserves_verified_global_baseline_exactly(self):
        baseline = (b'# reviewed administrator key route\nPort 2222\nPubkeyAuthentication yes\n'
                    b'PasswordAuthentication no\nAllowUsers archipelago-admin\n'
                    b'AuthorizedKeysFile /etc/ssh/admin_keys/%u\n')
        post = policy.ssh_policy(baseline, self.names)
        self.assertEqual(policy.ssh_baseline(post, self.names, policy.sha(baseline)), baseline)
        with self.assertRaises(ValueError):
            policy.ssh_baseline(post.replace(b'Port 2222', b'Port 22'), self.names, policy.sha(baseline))

    def test_no_loopback_match_or_include_can_substitute_for_global_denial(self):
        for baseline in (b'Match Address 127.0.0.1\nDenyUsers archipelago\n',
                         b'mAtCh=all\n', b'Include /etc/ssh/sshd_config.d/*.conf\n',
                         b'iNcLuDe=/run/alternate\n', b'DenyUsers archipelago-admin\n',
                         b'DenyUsers *@*\n', b'Include\\\n /run/alternate\n'):
            with self.subTest(baseline=baseline), self.assertRaises(ValueError):
                policy.ssh_policy(baseline, self.names)

    def test_preserves_openssh_directives_with_digits(self):
        baseline = (b'X11Forwarding yes\nX11DisplayOffset 10\nX11UseLocalhost yes\n'
                    b'PermitRootLogin no\nPubkeyAuthentication yes\n')
        post = policy.ssh_policy(baseline, self.names)
        self.assertEqual(post, b'DenyUsers archipelago c01-manager c01-authority\n' + baseline)
        self.assertEqual(policy.ssh_baseline(post, self.names, policy.sha(baseline)), baseline)

    def test_cannot_add_admin_to_generated_deny_list(self):
        baseline = b'PubkeyAuthentication yes\n'
        post = policy.ssh_policy(baseline, self.names + ['archipelago-admin'])
        with self.assertRaises(ValueError):
            policy.ssh_baseline(post, self.names, policy.sha(baseline))

    def test_remote_effective_semantics_not_only_loopback(self):
        baseline = b'PubkeyAuthentication yes\n'
        obj = types.SimpleNamespace(names=self.names, value={
            'operator': {'name': 'archipelago-admin'}, 'files': {'sshd': {'before': policy.sha(baseline)}}})
        contexts = []
        def command(argv):
            if '-T' not in argv:
                return 0, ''
            contexts.append(argv[-1])
            if 'addr=198.51.100.23' in argv[-1]:
                return 0, 'pubkeyauthentication yes\n'  # A broken/global claim must refuse.
            return 0, 'denyusers ' + ' '.join(self.names) + '\npubkeyauthentication yes\n'
        with patch.object(policy, 'read', return_value=policy.ssh_policy(baseline, self.names)), \
                patch.object(access, 'fixed_command', side_effect=command), self.assertRaises(ValueError):
            access.ssh_checks(obj)
        self.assertTrue(any('addr=198.51.100.23' in text for text in contexts))

    def test_staged_effective_operator_route_must_match(self):
        baseline = b'PubkeyAuthentication yes\n'
        obj = types.SimpleNamespace(names=self.names, value={
            'operator': {'name': 'archipelago-admin'}, 'files': {'sshd': {'before': policy.sha(baseline)}}})
        def command(argv):
            if '-T' not in argv:
                return 0, ''
            if str(policy.TARGETS['sshd']) in argv:
                return 0, 'pubkeyauthentication yes\n'
            return 0, 'denyusers ' + ' '.join(self.names) + '\npubkeyauthentication no\n'
        with patch.object(policy, 'read', return_value=policy.ssh_policy(baseline, self.names)), \
                patch.object(access, 'fixed_command', side_effect=command), self.assertRaises(ValueError):
            access.ssh_checks(obj, staged=True)


class EvidenceTests(unittest.TestCase):
    def test_late_pulse_does_not_extend_absolute_expiry(self):
        expiry = 10_000_000_000
        packet = access.watch_packet(b'PULSE', 4, expiry)
        self.assertEqual(access.watch_evidence(packet, 4, expiry, expiry - 250_000_000)[2], expiry)
        for stamp in (expiry, expiry + 1):
            with self.subTest(stamp=stamp), self.assertRaises(ValueError):
                access.watch_evidence(packet, 4, expiry, stamp)

    def test_expired_hold_and_regressed_generation_refuse(self):
        for packet in (access.watch_packet(b'HOLD', 3, 100), access.watch_packet(b'HOLD', 2, 200),
                       access.watch_packet(b'PULSE', 3, 150)):
            with self.subTest(packet=packet), self.assertRaises(ValueError):
                access.watch_evidence(packet, 3, 160, 100)

    def test_obsolete_failure_cannot_poison_candidate_or_refresh_evidence(self):
        leases = access.Leases.__new__(access.Leases)
        leases.generation = 2
        leases.last_observed = 123
        self.assertFalse(leases.checked({'generation': 1, 'failed': 'ValueError', 'reason': 'old scan'}, None))
        self.assertEqual(leases.last_observed, 123)
        with self.assertRaisesRegex(ValueError, 'current orphan'):
            leases.checked({'generation': 2, 'failed': 'ValueError', 'reason': 'current orphan'}, None)

    def test_late_success_cannot_resurrect_expired_lifetime(self):
        leases = access.Leases.__new__(access.Leases)
        leases.generation = 1
        leases.last_observed = 1
        leases.failed = False
        with patch.object(access.kernel, 'now', return_value=access.CHECK_NS + 1), self.assertRaises(ValueError):
            leases.checked({'generation': 1, 'observed': access.CHECK_NS}, None)


class RecoveryTests(unittest.TestCase):
    def setUp(self):
        self.policy = types.SimpleNamespace(digest='reviewed', value={
            'version': 1,
            'cohorts': [{'path': '/sys/fs/cgroup/old.service', 'members': {'42': 789}}],
            'copies': [{'disposition': 'quarantine-known-copy', 'identity': [1, 2, 1000, 1000, 384]}]})
        self.record = {'state': 'retired', 'configuration_sha256': 'reviewed', 'boot': 'previous', 'evidence': {
            'operations': access.OPERATIONS.copy(),
            'writers': [{'pid': 42, 'start': 789, 'group': '/sys/fs/cgroup/old.service/child'}],
            'copies': [{'index': 0, 'operation': 'quarantined-known-copy', 'identity': [1, 2, 1000, 1000, 384]}]}}

    def test_completed_retired_and_active_have_validated_provenance(self):
        for state in ('retired', 'owner-active'):
            record = dict(self.record, state=state)
            access.completed_history(record, self.policy)

    def test_partial_states_and_forged_completed_evidence_refuse(self):
        for state in ('transition-pending', 'operations-verified', 'requalification-pending'):
            with self.subTest(state=state), self.assertRaises(ValueError):
                access.completed_history(dict(self.record, state=state), self.policy)
        for field in ('writers', 'copies', 'operations'):
            record = copy.deepcopy(self.record)
            record['evidence'][field] = []
            with self.subTest(field=field), self.assertRaises(ValueError):
                access.completed_history(record, self.policy)


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