railiance-clock/tests/test_clock.py

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)