33 lines
1.7 KiB
Python
33 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Synthetic wire oracle for Rust boundary tests, never a production authorizer."""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import sys
|
|
from unittest.mock import patch
|
|
|
|
assert os.environ.get('ARCHY_TEST_ISOLATED') == '1'
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'scripts'))
|
|
import c01_nts as nts
|
|
import c01_recovery as recovery
|
|
|
|
binding = {'local_identity': [1] * 32, 'incarnation': [2] * 32, 'reader_schema': 2, 'writer_schema': 2}
|
|
observation = nts.Observation(None, '06' * 16, int(sys.argv[1]), 10_000_000_000, 0, 100_000_000, 10_000_000_000)
|
|
qualification = recovery.PlatformQualification('fixture', 'fixture', '12' * 32, 5_000_000, 100_000_000, 10)
|
|
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):
|
|
approval = recovery.prepare(object(), qualification, binding, bytes([7] * 16))
|
|
client, peer = socket.socketpair(socket.AF_UNIX, socket.SOCK_SEQPACKET)
|
|
with client, peer, patch.object(recovery, 'check_administrator'), \
|
|
patch.object(recovery, 'check_peer'), \
|
|
patch.object(recovery.time, 'clock_gettime_ns', return_value=10_000_000_001):
|
|
peer.send(bytes([42] * 32))
|
|
approval.send_once(client, recovery.Peer(123, 456, 456))
|
|
packet = peer.recv(8192)
|
|
assert json.loads(packet)['recovery']['uncertainty'] == {'secs': 0, 'nanos': 100_000_000}
|
|
sys.stdout.write(packet.decode('ascii'))
|