tegwick
9892c81ed6
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09cbb-87c6-7900-a145-4ce53ba9f1a6
65 lines
3.7 KiB
Python
65 lines
3.7 KiB
Python
import json
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
from urllib.request import Request, urlopen
|
|
from urllib.error import HTTPError
|
|
import pytest
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.hazmat.primitives import serialization
|
|
from railiance_clock.authority import Authority, Server
|
|
from railiance_clock.protocol import PROFILE, encode64, verify
|
|
from railiance_clock.core import ClockError
|
|
from railiance_clock.health import assess
|
|
|
|
@pytest.fixture
|
|
def authority(tmp_path):
|
|
key=ec.generate_private_key(ec.SECP256R1())
|
|
kp=tmp_path/'key.pem';kp.write_bytes(key.private_bytes(serialization.Encoding.PEM,serialization.PrivateFormat.PKCS8,serialization.NoEncryption()));kp.chmod(0o600)
|
|
hp=tmp_path/'health.json'
|
|
hp.write_text(json.dumps({'quality':'bounded','boot_id':Path('/proc/sys/kernel/random/boot_id').read_text().strip(),'policy_id':'test-v1','observed_monotonic_ns':1_000_000_000,'error_ns':1000}));hp.chmod(0o644)
|
|
cfg={'private_key_file':str(kp),'health_file':str(hp),'health_owner_uid':os.getuid(),'policy_id':'test-v1','authority_id':'test-authority','environment':'test','kid':'k1','max_health_age_ns':1_000_000_000,'host_drift_ppm':1000,'max_error_ns':100_000_000,'max_continuity_error_ns':10_000_000,'max_sample_age_ns':5_000_000_000}
|
|
clocks={'wall':100_000_000_000,'mono':1_000_000_000}
|
|
return Authority(cfg,wall=lambda:clocks['wall'],monotonic=lambda:clocks['mono']),hp,clocks
|
|
|
|
def request():
|
|
return dict(profile=PROFILE,authority_id='test-authority',environment='test',nonce=encode64(os.urandom(32)))
|
|
|
|
def test_live_http_sample_and_health_refusal(authority):
|
|
a,hp,clocks=authority
|
|
server=Server(('127.0.0.1',0),a);thread=threading.Thread(target=server.serve_forever);thread.start()
|
|
try:
|
|
url='http://127.0.0.1:'+str(server.server_port)+'/v1/time-samples'
|
|
req=Request(url,data=json.dumps(request()).encode(),headers={'Content-Type':'application/json'})
|
|
with urlopen(req) as r:
|
|
assert r.status==200
|
|
assert verify(r.read(),a.key.public_key(),'k1')['epoch']==a.epoch
|
|
hp.write_text('{"quality":"unusable"}')
|
|
with pytest.raises(HTTPError) as exc:urlopen(req)
|
|
assert exc.value.code==503
|
|
finally:
|
|
server.shutdown();server.server_close();thread.join()
|
|
|
|
@pytest.mark.parametrize('change',['stale','step','unsafe_health','wrong_boot','unknown_policy'])
|
|
def test_authority_refuses_unusable_health(authority,change):
|
|
a,hp,clocks=authority
|
|
if change=='stale':clocks['mono']+=2_000_000_000;clocks['wall']+=2_000_000_000
|
|
if change=='step':clocks['wall']-=1_000_000_000
|
|
if change=='unsafe_health':hp.chmod(0o666)
|
|
if change in ('wrong_boot','unknown_policy'):
|
|
d=json.loads(hp.read_text());d['boot_id' if change=='wrong_boot' else 'policy_id']='other';hp.write_text(json.dumps(d))
|
|
with pytest.raises(ClockError):a.sample(json.dumps(request()).encode())
|
|
|
|
def test_restart_changes_epoch(authority):
|
|
a,_,clocks=authority
|
|
b=Authority(a.config,wall=lambda:clocks['wall'],monotonic=lambda:clocks['mono'])
|
|
assert a.epoch!=b.epoch
|
|
|
|
def test_health_model_refusals():
|
|
p={'upstreams':['ntp.ubuntu.com'],'measurement_margin_ns':1000,'max_error_ns':100_000_000}
|
|
k={'state':0,'status':0,'maxerror_us':1000,'esterror_us':100,'tolerance_scaled_ppm':32768000}
|
|
c={'NTP':'yes','NTPSynchronized':'yes'};u={'ServerName':'ntp.ubuntu.com','NTPMessage':'{ Leap=0, Ignored=no, }'}
|
|
assert assess(p,k,c,u)==1001000
|
|
for changed in [{**k,'state':5},{**k,'status':16},{**k,'maxerror_us':200_000},{**k,'maxerror_us':-1}]:
|
|
with pytest.raises(ClockError):assess(p,changed,c,u)
|
|
with pytest.raises(ClockError):assess(p,k,c,{**u,'ServerName':'unadmitted'})
|