tegwick
3e537d117f
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a0e6f0-a6f6-7b93-9124-55ef584930e2
118 lines
5.1 KiB
Python
118 lines
5.1 KiB
Python
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
|
|
|
|
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)
|