"""Automatic issuer wire tests; fixture credentials/time are not qualification.""" import copy from dataclasses import replace from decimal import Decimal import json from pathlib import Path 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 dict(local_identity=[1]*32, incarnation=[2]*32, reader_schema=2, writer_schema=2) def challenge(): return dict(version=3, binding=binding(), delegation_id=[3]*32, application_owner=[4]*32, epoch=1, cause='Restart', boot_id=[6]*16, not_before=recovery.duration(0), max_awake=recovery.duration(10**9), characterized=None, barrier_raw=None, domain=list(bytes.fromhex('513ae7b62396dedc13f0f76203236b2c')), nonce=[5]*32) def qualification(): return recovery.PlatformQualification('fixture', 'fixture', '12'*32, 5000, 2000, 10, 10**9) def observation(): now = time.clock_gettime_ns(time.CLOCK_BOOTTIME) wall = time.time_ns() assessment = nts.Assessment('fixture', '127.0.0.1', Decimal(wall)/nts.NANO, Decimal(0), 1000, (), ()) return nts.Observation(assessment, '06'*16, wall, now, 0, 1000, now) class EpochIssuerTests(unittest.TestCase): def test_missing_awake_bound_refuses_before_network_collection(self): q=replace(qualification(),max_awake_ns=None) with patch.object(recovery.os,'getresuid',return_value=(0,0,0)), \ patch.object(recovery.os,'getresgid',return_value=(0,0,0)), \ patch.object(recovery.PlatformQualification,'validate'), \ patch.object(nts,'collect') as collect, self.assertRaisesRegex(nts.Unavailable,'awake lifetime'): recovery.prepare_epoch(object(),q,binding(),challenge(), recovery.EpochIssuerDelegation(bytes([3])*32,True),10**9,time.clock_gettime_ns(time.CLOCK_BOOTTIME)+10**9) collect.assert_not_called() def test_waits_for_new_reference_without_reissuing_challenge_or_reanchoring(self): o=observation(); c=challenge();c['not_before']=recovery.duration(o.boottime_ns-10**9) stale=replace(o,assessment=replace(o.assessment,reference_time=Decimal(o.utc_ns-60*10**9)/nts.NANO)) q=replace(qualification(),max_awake_ns=10**9) with patch.object(recovery.os,'getresuid',return_value=(0,0,0)), \ patch.object(recovery.os,'getresgid',return_value=(0,0,0)), \ patch.object(recovery.PlatformQualification,'validate'), \ patch.object(recovery,'boot_id',return_value='06'*16), \ patch.object(recovery.time,'sleep'), patch.object(nts,'collect',side_effect=[stale,o]) as collect: prepared=recovery.prepare_epoch(object(),q,binding(),c, recovery.EpochIssuerDelegation(bytes([3])*32,True),10**9,o.boottime_ns+10**9) self.assertEqual(collect.call_count,2) self.assertIs(prepared._observation,o) c['application_owner']=[9]*32 self.assertEqual(prepared._epoch_challenge['application_owner'],[4]*32) def test_collected_epoch_is_not_signed_for_a_different_appointed_session(self): prepared=self.prepared();prepared._epoch_challenge=challenge() client,server=socket.socketpair(socket.AF_UNIX,socket.SOCK_SEQPACKET) self.addCleanup(server.close);server.settimeout(1) other=challenge();other['application_owner']=[9]*32 server.send(json.dumps(other).encode()) with patch.object(recovery,'check_administrator'),patch.object(recovery,'check_peer'), \ patch.object(recovery.PlatformQualification,'validate'),patch.object(recovery,'boot_id',return_value='06'*16), \ self.assertRaisesRegex(nts.Unavailable,'different appointed epoch/session'): prepared.send_epoch_once(client,recovery.Peer(123,456,456),recovery.EpochIssuerDelegation(bytes([3])*32,True)) self.assertTrue(prepared._used) self.assertEqual(server.recv(8192),b'') def packet(self, c, wake=False, o=None): return recovery.epoch_packet(o or observation(), recovery.binding_bytes(binding()), qualification(), c, recovery.EpochIssuerDelegation(bytes([3])*32, wake)) def test_cross_language_domain_vector_and_exact_original_observation(self): o = observation() p = self.packet(challenge(), o=o) self.assertEqual(bytes(p['recovery']['domain']).hex(), '513ae7b62396dedc13f0f76203236b2c') self.assertEqual(p['recovery']['wall'], recovery.duration(o.utc_ns)) self.assertEqual(p['recovery']['uncertainty'], recovery.duration(o.observation_error_ns)) self.assertEqual(p['challenge']['application_owner'], [4]*32) self.assertEqual(p['expires_at'], recovery.duration(o.boottime_ns+10**9)) def test_immutable_expiry_refuses_missing_legacy_mismatched_and_overflow_lifetimes(self): for mutate in (lambda c:c.pop('max_awake'), lambda c:c.update(version=1), lambda c:c.update(max_awake=recovery.duration(0)), lambda c:c.update(max_awake=recovery.duration(2*10**9))): c=challenge();mutate(c) with self.assertRaises(nts.Unavailable):self.packet(c) o=replace(observation(),boottime_ns=(2**64-1)*10**9) with self.assertRaisesRegex(nts.Unavailable,'duration'):self.packet(challenge(),o=o) def test_expiry_is_not_extended_when_same_observation_is_encoded_later(self): o=observation();first=self.packet(challenge(),o=o) with patch.object(recovery.time,'clock_gettime_ns',return_value=o.boottime_ns+500_000_000): second=self.packet(challenge(),o=o) self.assertEqual(first,second) def test_automatic_wake_is_allowed_only_by_matching_protected_delegation(self): c = challenge(); c['cause'] = 'Wake' with self.assertRaisesRegex(nts.Unavailable, 'does not cover'): self.packet(c) self.assertEqual(self.packet(c, wake=True)['challenge'], c) def test_old_domain_boot_delegation_pre_wake_and_malformed_challenges_refuse(self): for mutate in ( lambda c: c.update(epoch=2), lambda c: c.update(epoch=True), lambda c: c.update(domain=[8]*16), lambda c: c.update(boot_id=[7]*16), lambda c: c.update(delegation_id=[8]*32), lambda c: c.update(application_owner=[0]*32), lambda c: c.update(nonce=[True]*32), lambda c: c.update(cause='ManualBypass'), lambda c: c.update(extra=1), lambda c: c['binding'].update(incarnation=[8]*32), lambda c: c.update(not_before=recovery.duration(observation().boottime_ns+10**9)), ): c = copy.deepcopy(challenge()); mutate(c) with self.subTest(challenge=c), self.assertRaises(nts.Unavailable): self.packet(c, wake=True) def prepared(self): with patch.object(recovery.os, 'getresuid', return_value=(0,0,0)), \ patch.object(recovery.os, 'getresgid', return_value=(0,0,0)), \ patch.object(recovery.PlatformQualification, 'validate'), \ patch.object(recovery, 'boot_id', return_value='06'*16), \ patch.object(nts, 'collect', return_value=observation()): return recovery.prepare(object(), qualification(), binding(), bytes([7])*16) def test_socket_automatic_issuer_consumes_across_both_protocols(self): p = self.prepared() client, server = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) self.addCleanup(server.close); self.addCleanup(client.close) server.settimeout(1); server.send(json.dumps(challenge()).encode()) with patch.object(recovery, 'check_administrator'), patch.object(recovery, 'check_peer'), \ patch.object(recovery.PlatformQualification, 'validate'), \ patch.object(recovery, 'boot_id', return_value='06'*16): p.send_epoch_once(client, recovery.Peer(123,456,456), recovery.EpochIssuerDelegation(bytes([3])*32, False)) packet = json.loads(server.recv(8192)) self.assertEqual(packet['challenge'], challenge()) self.assertEqual(packet['version'], 3) again, other = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) self.addCleanup(other.close) with self.assertRaisesRegex(nts.Unavailable, 'already consumed'): p.send_once(again, recovery.Peer(123,456,456)) self.assertEqual(again.fileno(), -1) def test_bad_wake_scope_consumes_without_emitting_a_grant(self): p = self.prepared(); client, server = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET) self.addCleanup(server.close); server.settimeout(1) c = challenge(); c['cause']='Wake'; server.send(json.dumps(c).encode()) with patch.object(recovery, 'check_administrator'), patch.object(recovery, 'check_peer'), \ patch.object(recovery.PlatformQualification, 'validate'), \ patch.object(recovery, 'boot_id', return_value='06'*16), self.assertRaises(nts.Unavailable): p.send_epoch_once(client, recovery.Peer(123,456,456), recovery.EpochIssuerDelegation(bytes([3])*32,False)) self.assertEqual(server.recv(8192), b'') self.assertTrue(p._used) def test_post_wake_collection_does_not_reuse_pre_wake_nts_reference(self): o = observation() c = challenge(); c['cause'] = 'Wake' c['not_before'] = recovery.duration(o.boottime_ns - 1_000_000) old = replace(o, assessment=replace(o.assessment, reference_time=Decimal(o.utc_ns - 2_000_000)/nts.NANO)) with self.assertRaisesRegex(nts.Unavailable, 'authenticated reference'): self.packet(c, wake=True, o=old) self.assertEqual(self.packet(c, wake=True, o=o)['reference_wall'], recovery.duration(o.utc_ns)) if __name__ == '__main__': unittest.main()