#!/usr/bin/env python3 """Deterministic handoff checks; credentials/time are fixtures, not acceptance.""" from dataclasses import replace import json from pathlib import Path import pickle import socket import sys import time import unittest from unittest.mock import patch sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import c01_nts as nts import c01_recovery as recovery def binding(): return {'local_identity': [1] * 32, 'incarnation': [2] * 32, 'reader_schema': 2, 'writer_schema': 2} def observation(): now = time.clock_gettime_ns(time.CLOCK_BOOTTIME) return nts.Observation(None, 'ab' * 16, 1_700_000_000 * recovery.NANO, now, 0, 1_000, now) def qualification(): return recovery.PlatformQualification('fixture-kernel', 'fixture-clock', '12' * 32, 5_000, 2_000, 10) class HandoffTests(unittest.TestCase): def setUp(self): self.platform = patch.object(recovery.PlatformQualification, 'validate') self.platform.start(); self.addCleanup(self.platform.stop) self.boot = patch.object(recovery, 'boot_id', return_value='ab' * 16) self.boot.start(); self.addCleanup(self.boot.stop) def prepare(self, value=None, state=None): with patch.object(recovery.os, 'getresuid', return_value=(0, 0, 0)), \ patch.object(recovery.os, 'getresgid', return_value=(0, 0, 0)), \ patch.object(nts, 'collect', return_value=value or observation()) as collect: result = recovery.prepare(object(), qualification(), state or binding(), bytes([7] * 16)) collect.assert_called_once() return result def send(self, prepared, client): with patch.object(recovery, 'check_administrator'), patch.object(recovery, 'check_peer'): prepared.send_once(client, recovery.Peer(123, 456, 456)) def pair(self): client, server = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) self.addCleanup(client.close); self.addCleanup(server.close) server.settimeout(1) return client, server def test_live_collection_freezes_exact_binding_mapping_and_challenge(self): original = observation() state = binding() prepared = self.prepare(original, state) state['incarnation'][0] = 99 client, server = self.pair() challenge = bytes(range(32)) server.send(challenge) self.send(prepared, client) packet = json.loads(server.recv(8192)) self.assertEqual(packet['challenge'], list(challenge)) self.assertEqual(packet['binding'], binding()) self.assertEqual(packet['version'], 4) self.assertEqual(packet['recovery'], { 'domain': [7] * 16, 'boot_id': [171] * 16, 'wall': recovery.duration(original.utc_ns), 'monotonic': recovery.duration(original.boottime_ns), 'max_bracket': recovery.duration(qualification().max_bracket_ns), 'uncertainty': recovery.duration(original.observation_error_ns), 'max_elapsed_error_ppb': qualification().max_elapsed_error_ppb}) with self.assertRaisesRegex(nts.Unavailable, 'already consumed'): self.send(prepared, client) self.assertEqual(client.fileno(), -1) fresh, peer = self.pair() with self.assertRaisesRegex(nts.Unavailable, 'already consumed'): self.send(prepared, fresh) self.assertEqual(fresh.fileno(), -1) self.assertEqual(peer.recv(8192), b'') def test_collection_failure_never_returns_an_approval(self): with patch.object(recovery.os, 'getresuid', return_value=(0, 0, 0)), \ patch.object(recovery.os, 'getresgid', return_value=(0, 0, 0)), \ patch.object(nts, 'collect', side_effect=nts.Unavailable('no authenticated time')): with self.assertRaisesRegex(nts.Unavailable, 'no authenticated time'): recovery.prepare(object(), qualification(), binding(), bytes([7] * 16)) def test_non_root_and_saved_non_root_preparation_refuse_before_collection(self): for ids in ((1, 1, 1), (0, 0, 1)): with self.subTest(ids=ids), patch.object(recovery.os, 'getresuid', return_value=ids), \ patch.object(nts, 'collect') as collect: with self.assertRaises(nts.Unavailable): recovery.prepare(object(), qualification(), binding(), bytes([7] * 16)) collect.assert_not_called() def test_platform_observation_bounds_and_boot_are_required(self): original = observation() for invalid in (replace(original, boot_id='cd' * 16), replace(original, observation_error_ns=2_001), replace(original, sampling_bracket_ns=5_001), replace(original, collected_at_ns=original.collected_at_ns + 1)): with self.subTest(invalid=invalid), self.assertRaises(nts.Unavailable): self.prepare(invalid) def test_stale_boot_changed_or_regressed_observation_is_consumed_without_send(self): for case in ('stale', 'future', 'boot'): original = observation() prepared = self.prepare(original) now = original.collected_at_ns if case == 'stale': now += recovery.IO_BUDGET_NS elif case == 'future': now -= 1 client, server = self.pair() server.send(bytes(32)) with patch.object(recovery.time, 'clock_gettime_ns', return_value=now), \ patch.object(recovery, 'boot_id', return_value='cd' * 16 if case == 'boot' else 'ab' * 16): with self.assertRaises(nts.Unavailable): self.send(prepared, client) self.assertTrue(prepared._used) try: self.assertEqual(server.recv(8192), b'') except ConnectionResetError: # Closing with the unread challenge queued may reset seqpacket. # Neither EOF nor reset delivered an authorization packet. pass def test_wrong_challenge_size_refuses_and_cannot_retry(self): for size in (0, 31, 33, 8193): with self.subTest(size=size): prepared = self.prepare() client, server = self.pair() server.send(bytes(size)) with self.assertRaisesRegex(nts.Unavailable, 'invalid recovery challenge'): self.send(prepared, client) self.assertTrue(prepared._used) self.assertEqual(client.fileno(), -1) def test_peer_or_privilege_refusal_consumes_approval(self): for name in ('check_administrator', 'check_peer'): prepared = self.prepare() client, _ = self.pair() with patch.object(recovery, 'check_administrator'), patch.object(recovery, 'check_peer'), \ patch.object(recovery, name, side_effect=nts.Unavailable('identity refused')): with self.assertRaisesRegex(nts.Unavailable, 'identity refused'): prepared.send_once(client, recovery.Peer(123, 456, 456)) self.assertTrue(prepared._used) self.assertEqual(client.fileno(), -1) def test_approval_has_no_pickle_transport(self): with self.assertRaisesRegex(TypeError, 'must not be serialized'): pickle.dumps(self.prepare()) def test_deadline_rechecked_after_poll(self): client, server = self.pair() server.send(bytes(32)) with patch.object(recovery.time, 'clock_gettime_ns', side_effect=[0, 10]): with self.assertRaisesRegex(nts.Unavailable, 'deadline expired'): recovery.wait_ready(client, recovery.select.POLLIN, 10) class BindingAndQualificationTests(unittest.TestCase): def test_binding_refuses_foreign_schema_unknown_fields_and_invalid_bytes(self): for change in ({'reader_schema': 1}, {'writer_schema': True}, {'extra': 0}, {'incarnation': [0] * 32}, {'local_identity': [256] * 32}, {'incarnation': [True] * 32}): with self.subTest(change=change), self.assertRaises(nts.Unavailable): recovery.binding_bytes({**binding(), **change}) def test_duration_retains_fraction_and_rejects_invalid_values(self): self.assertEqual(recovery.duration(1_000_000_123), {'secs': 1, 'nanos': 123}) for invalid in (-1, True, 0.5, 2 ** 64 * recovery.NANO): with self.assertRaises(nts.Unavailable): recovery.duration(invalid) def test_platform_requires_exact_kernel_clocksource_and_evidence(self): current = qualification() with patch.object(recovery.os, 'uname') as uname, \ patch.object(recovery, 'clocksource', return_value=current.clocksource): uname.return_value.release = current.kernel_release current.validate() for invalid in (replace(current, kernel_release='foreign'), replace(current, clocksource='foreign'), replace(current, evidence_sha256='0' * 64), replace(current, evidence_sha256='z' * 64), replace(current, max_bracket_ns=0), replace(current, max_observation_error_ns=True), replace(current, max_elapsed_error_ppb=-1), replace(current, max_elapsed_error_ppb=True)): with self.subTest(invalid=invalid), self.assertRaises(nts.Unavailable): invalid.validate() if __name__ == '__main__': unittest.main()