203 lines
9.5 KiB
Python
203 lines
9.5 KiB
Python
#!/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()
|