180 lines
9.8 KiB
Python
180 lines
9.8 KiB
Python
"""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()
|