import json import os from datetime import datetime, timezone import pytest from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives import serialization from railiance_clock.client import Clock, FileTrust from railiance_clock.core import ClockError, Tick, Window from railiance_clock.protocol import sign, strict_json, encode64, verify class Timer: def __init__(self): self.n=0; self.boot=0; self.domain='boot1' def read(self): return Tick(self.n, self.n+self.boot, self.domain) @pytest.fixture def setup(tmp_path): key=ec.generate_private_key(ec.SECP256R1()) data={'status':'admitted','authority_id':'railiance01','environment':'test','epoch':'epoch1','policy_id':'test-v1','kid':'test-key', 'endpoint':'https://clock.invalid/v1/time-samples', 'public_key_pem':key.public_key().public_bytes(serialization.Encoding.PEM,serialization.PublicFormat.SubjectPublicKeyInfo).decode(), 'limits':{'max_age_ns':5_000_000_000,'max_rtt_ns':2_000_000_000,'max_width_ns':3_000_000_000,'max_server_error_ns':50_000_000,'timer_ppm':1000,'timer_resolution_ns':1000,'suspend_tolerance_ns':5_000_000,'trust_session_ns':60_000_000_000}} tmp_path.chmod(0o700) state=tmp_path/'state.json';state.write_text(json.dumps({'authority_id':'railiance01','epoch':'epoch1','lower_ns':0}));state.chmod(0o600) data.update(client_boot_id='boot1',trust_not_after_boottime_ns=60_000_000_000,rollback_state_file=str(state)) path=tmp_path/'trust.json';path.write_text(json.dumps(data));path.chmod(0o600) timer=Timer() def transport(request): timer.n+=10_000_000 return sign(dict(request,epoch='epoch1',policy_id='test-v1',quality='bounded',receive_unix_ns='100000000000',send_unix_ns='100001000000',error_ns='1000000',max_age_ns='5000000000'),key,'test-key') return path,key,timer,transport def test_interval_contains_send_with_asymmetric_delay(setup, monkeypatch): path,key,timer,transport=setup import time monkeypatch.setattr(time,'time',lambda: -123456789) w=Clock(FileTrust(path),timer=timer,transport=transport).read() assert w.lower_ns<=100001000000<=w.upper_ns assert w.upper_ns>=100011000000 @pytest.mark.parametrize('change',['nonce','epoch','policy_id','authority_id','environment']) def test_signed_wrong_binding_refuses(setup,change): path,key,timer,transport=setup def changed(request): payload=verify(transport(request),key.public_key(),'test-key') payload[change]=encode64(os.urandom(32)) if change=='nonce' else 'wrong' return sign(payload,key,'test-key') with pytest.raises(ClockError):Clock(FileTrust(path),timer=timer,transport=changed).read() def test_bad_signature(setup): path,key,timer,transport=setup with pytest.raises(ClockError): Clock(FileTrust(path),timer=timer,transport=lambda r:transport(r)[:-10]+b'AAAAAAAAAA').read() def test_replay_refused(setup): path,key,timer,transport=setup old=[] def replay(request): if not old:old.append(transport(request)) return old[0] clock=Clock(FileTrust(path),timer=timer,transport=replay) clock.read() with pytest.raises(ClockError):clock.read() @pytest.mark.parametrize('failure',['delay','suspend','domain','backward','trust']) def test_lifecycle_during_exchange_refuses(setup,failure): path,key,timer,transport=setup def disrupted(request): result=transport(request) if failure=='delay':timer.n+=6_000_000_000 if failure=='suspend':timer.boot+=1_000_000_000 if failure=='domain':timer.domain='boot2' if failure=='backward':timer.n=-1 if failure=='trust':path.write_text('{}') return result with pytest.raises(ClockError):Clock(FileTrust(path),timer=timer,transport=disrupted).read() def test_expired_session(setup): path,key,timer,transport=setup clock=Clock(FileTrust(path),timer=timer,transport=transport);timer.n=61_000_000_000 with pytest.raises(ClockError):clock.read() @pytest.mark.parametrize('raw',[b'{"a":1,"a":2}',b'{"a":NaN}',b'\xef\xbb\xbf{}',b'{}{}',b'{"a":{"b":{"c":{"d":{"e":1}}}}}']) def test_strict_json(raw): with pytest.raises(ClockError):strict_json(raw) def test_validity_uses_entire_interval(): w=Window(10_000_000_000,12_000_000_000,'a','e','p','s') dt=lambda n:datetime.fromtimestamp(n,timezone.utc) w.require_valid(dt(10),dt(13)) with pytest.raises(ClockError):w.require_valid(dt(11),dt(13)) with pytest.raises(ClockError):w.require_valid(dt(10),dt(12)) def test_new_client_cannot_reset_expired_trust(setup): path,key,timer,transport=setup timer.n=61_000_000_000 with pytest.raises(ClockError):Clock(FileTrust(path),timer=timer,transport=transport) def test_missing_state_refuses_without_sending(setup): path,key,timer,transport=setup from pathlib import Path Path(json.loads(path.read_text())['rollback_state_file']).unlink() called=[] with pytest.raises(ClockError):Clock(FileTrust(path),timer=timer,transport=lambda r:called.append(r)).read() assert not called def test_restart_preserves_floor(setup): path,key,timer,transport=setup Clock(FileTrust(path),timer=timer,transport=transport).read() def older(request): payload=verify(transport(request),key.public_key(),'test-key') payload.update(receive_unix_ns='50000000000',send_unix_ns='50001000000') return sign(payload,key,'test-key') with pytest.raises(ClockError):Clock(FileTrust(path),timer=timer,transport=older).read() def test_rotation_requires_fresh_client_and_matching_signer(setup): path, key, timer, transport = setup old = Clock(FileTrust(path), timer=timer, transport=transport) old.read() replacement = ec.generate_private_key(ec.SECP256R1()) data = json.loads(path.read_text()) data['kid'] = 'rotated-key' data['public_key_pem'] = replacement.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo).decode() path.write_text(json.dumps(data)) with pytest.raises(ClockError): old.read() # Loading new trust must still refuse the old signer. with pytest.raises(ClockError): Clock(FileTrust(path), timer=timer, transport=transport).read() def rotated(request): payload = verify(transport(request), key.public_key(), 'test-key') return sign(payload, replacement, 'rotated-key') assert Clock(FileTrust(path), timer=timer, transport=rotated).read().authority_id == 'railiance01' data['status'] = 'revoked' path.write_text(json.dumps(data)) with pytest.raises(ClockError): FileTrust(path)