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

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)