197 lines
10 KiB
Python
197 lines
10 KiB
Python
"""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()
|