railiance-platform/tests/test_flex_auth_signer_custody.py

"""Actual disposable KV CAS/readback; no estate key, identity or grants."""
from dataclasses import replace
import base64
import json
from pathlib import Path
import sys

import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
sys.path.insert(0,str(Path(__file__).resolve().parents[1]/'scripts'))
from test_task_execution_native_fixture import bao_dev
from flex_auth_signer_custody import SignerWriter,FixtureWriterAdmission,BINDINGS,PATH
from task_execution_journal import Refused


@pytest.fixture
def fixture(bao_dev,tmp_path):
    origin,root,call,process=bao_dev
    assert call('POST','sys/mounts/platform',{'type':'kv','options':{'version':'2'}})[0] in {200,204}
    directory=tmp_path/'receipts';directory.mkdir(mode=0o700)
    admission=FixtureWriterAdmission(int(origin.rsplit(':',1)[1]),root,process,True)
    def writer(operation='first',version=0,**kwargs):
        return SignerWriter(admission,directory,operation,kid=kwargs.pop('kid','synthetic-key-'+str(version)),
                            expected_version=version,bindings={key:'sha256:'+'a'*64 for key in BINDINGS},**kwargs)
    return writer,call,directory,admission


def test_actual_bootstrap_rotation_public_correspondence_and_cas_conflict(fixture):
    writer,call,directory,admission=fixture
    first=writer();receipt=first.write()
    assert receipt['status']=='custody-verified-pending-independent-public-admission'
    assert receipt['kv_version']==1 and receipt['exact_readback']
    document=call('GET',PATH)[1]['data']['data']
    seed=base64.urlsafe_b64decode(document['ED25519_PRIVATE_KEY']+'=')
    public=Ed25519PrivateKey.from_private_bytes(seed).public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.Raw)
    assert len(seed)==32
    assert base64.urlsafe_b64encode(public).decode().rstrip('=')==receipt['public_key']==document['ED25519_PUBLIC_KEY']
    raw=(directory/'first.json').read_bytes()
    assert document['ED25519_PRIVATE_KEY'].encode() not in raw and admission.owner_token.encode() not in raw
    assert (directory/'first.json').stat().st_mode&0o777==0o600
    with pytest.raises(Refused):first.write()
    with pytest.raises(FileExistsError):writer()
    conflict=writer('conflict',0)
    with pytest.raises(Refused,match='outcome unconfirmed'):conflict.write()
    assert json.loads((directory/'conflict.json').read_bytes())['status']=='write-intent-unknown'
    assert call('GET',PATH)[1]['data']['metadata']['version']==1
    second=writer('rotation',1,kid='synthetic-successor').write()
    assert second['kv_version']==2 and second['public_key']!=receipt['public_key']
    assert call('GET',PATH+'?version=1')[1]['data']['data']==document
    assert call('GET',PATH)[1]['data']['data']['kid']=='synthetic-successor'


@pytest.mark.parametrize('case',['lost-response','readback-mismatch','read-denied','write-denied'])
def test_unknown_is_durable_and_no_retry_or_private_output(fixture,monkeypatch,case):
    factory,call,directory,admission=fixture
    writer=factory();original=writer._http;seen=[]
    def fault(method,path,body=None):
        seen.append((method,path))
        if case=='write-denied' and method=='POST':return 403,{'errors':['synthetic secret backend detail']}
        if case=='read-denied' and method=='GET':return 403,{'errors':['synthetic secret backend detail']}
        result=original(method,path,body)
        if case=='lost-response' and method=='POST':raise OSError('synthetic secret backend detail')
        if case=='readback-mismatch' and method=='GET':result[1]['data']['data']['kid']='different'
        return result
    monkeypatch.setattr(writer,'_http',fault)
    with pytest.raises(Refused,match='^signer custody outcome unconfirmed$'):writer.write()
    raw=(directory/'first.json').read_bytes();receipt=json.loads(raw)
    assert receipt['status']=='write-intent-unknown' and receipt['exact_readback'] is False
    assert b'synthetic secret backend detail' not in raw and admission.owner_token.encode() not in raw
    if case!='write-denied':
        document=call('GET',PATH)[1]['data']['data']
        assert document['ED25519_PRIVATE_KEY'].encode() not in raw
    before=list(seen)
    with pytest.raises(Refused):writer.write()
    assert seen==before
    with pytest.raises(FileExistsError):factory()


def test_durable_intent_failure_prevents_effect(fixture,monkeypatch):
    factory,call,directory,admission=fixture
    writer=factory()
    def fail(**kwargs):raise OSError('private storage failed')
    monkeypatch.setattr(writer,'_save',fail)
    with pytest.raises(OSError):writer.write()
    assert call('GET',PATH)[0]==404


@pytest.mark.parametrize('case',['unfenced','wrong-port','bool-version','indirect-directory','unsafe-directory'])
def test_invalid_custody_contract_before_generation_or_http(fixture,case):
    factory,call,directory,admission=fixture
    expected=0
    if case=='unfenced':admission=replace(admission,fenced=False)
    elif case=='wrong-port':admission=replace(admission,port=admission.port+1)
    elif case=='bool-version':expected=True
    elif case=='indirect-directory':
        link=directory.parent/'indirect';link.symlink_to(directory,target_is_directory=True);directory=link
    elif case=='unsafe-directory':directory.chmod(0o755)
    with pytest.raises(Refused):SignerWriter(admission,directory,'bad',kid='synthetic',expected_version=expected,
        bindings={key:'sha256:'+'a'*64 for key in BINDINGS})
    assert call('GET',PATH)[0]==404


def test_failed_success_receipt_retains_unknown_after_actual_write(fixture,monkeypatch):
    factory,call,directory,admission=fixture
    writer=factory();original=writer._save
    def fail_success_only(**kwargs):
        if writer.receipt['status']=='custody-verified-pending-independent-public-admission':
            raise OSError('synthetic receipt failure')
        return original(**kwargs)
    monkeypatch.setattr(writer,'_save',fail_success_only)
    with pytest.raises(Refused,match='outcome unconfirmed'):writer.write()
    assert call('GET',PATH)[1]['data']['metadata']['version']==1
    assert json.loads((directory/'first.json').read_bytes())['status']=='write-intent-unknown'
    with pytest.raises(Refused):writer.write()