hack-house/tests/test_sor_consent.py

"""R5 — consent handshake + X25519 hop-credential acceptance checks (Python).

Acceptance predicate (roadmap R5): a host recruits a hop only after an explicit
accept; a reject leaves no circuit entry; the hop credential decrypts only with
the host's X25519 key (a third party cannot open it); signature verification
gates acceptance (unsigned/forged request -> rejected). Mirrors the Rust checks
in hh/src/sor/consent.rs, and pins cross-language KDF parity with crypto.rs.
"""

import base64

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from cmd_chat.sor.consent import (
    CircuitBuilder,
    ConsentAccept,
    ConsentReject,
    ConsentRequest,
    _seal_key,
    node_evaluate,
    open_sealed,
    parse_sor_frame,
    persona_sign,
    seal_to_pubkey,
    x25519_keypair,
)


def _ed_signer(seed: int):
    raw = bytes([seed]) * 32
    sk = Ed25519PrivateKey.from_private_bytes(raw)
    pub_b64 = base64.standard_b64encode(sk.public_key().public_bytes_raw()).decode()
    return raw, pub_b64


def _signed_request(host_ed_raw, host_ed_pub, host_x_pub, circuit="c0", hop=1):
    req = ConsentRequest(host_ed_pub, host_x_pub, circuit, hop, "deadbeef", "")
    req.sig = persona_sign(host_ed_raw, req.canonical())
    return req


def test_accept_recruits_a_hop_only_for_the_host():
    host_ed_raw, host_ed_pub = _ed_signer(1)
    host_x_sk, host_x_pub = x25519_keypair()
    _, node_ed_pub = _ed_signer(2)

    req = _signed_request(host_ed_raw, host_ed_pub, host_x_pub)
    assert req.signature_ok()

    decision = node_evaluate(req, node_ed_pub, True, b"ephemeral-hop-key")
    assert isinstance(decision, ConsentAccept)

    cb = CircuitBuilder("c0", host_x_sk)
    assert cb.recruit(decision), "accept must recruit a hop"
    assert len(cb.hops) == 1
    assert cb.hops[0].cred == b"ephemeral-hop-key"

    # A different host (wrong X25519 secret) cannot open the cred -> no hop.
    other_x_sk, _ = x25519_keypair()
    evil = CircuitBuilder("c0", other_x_sk)
    assert not evil.recruit(decision), "third party must not recruit"
    assert evil.hops == []


def test_forged_or_unsigned_request_is_rejected():
    _, host_ed_pub = _ed_signer(1)
    _, host_x_pub = x25519_keypair()
    _, node_ed_pub = _ed_signer(2)

    unsigned = ConsentRequest(host_ed_pub, host_x_pub, "c0", 1, "n", "")
    assert not unsigned.signature_ok()
    assert isinstance(node_evaluate(unsigned, node_ed_pub, True, b"k"), ConsentReject)

    # Forged: signed by the WRONG key but claiming host_ed_pub.
    wrong_raw, _ = _ed_signer(9)
    unsigned.sig = persona_sign(wrong_raw, unsigned.canonical())
    assert not unsigned.signature_ok(), "forged signature must not verify"
    assert isinstance(node_evaluate(unsigned, node_ed_pub, True, b"k"), ConsentReject)


def test_tampering_a_bound_field_breaks_signature():
    host_ed_raw, host_ed_pub = _ed_signer(1)
    _, host_x_pub = x25519_keypair()
    req = _signed_request(host_ed_raw, host_ed_pub, host_x_pub)
    req.hop_index = 2  # was 1 when signed
    assert not req.signature_ok()


def test_reject_leaves_no_entry():
    host_ed_raw, host_ed_pub = _ed_signer(1)
    host_x_sk, host_x_pub = x25519_keypair()
    _, node_ed_pub = _ed_signer(2)
    req = _signed_request(host_ed_raw, host_ed_pub, host_x_pub)

    decision = node_evaluate(req, node_ed_pub, False, b"k")
    assert isinstance(decision, ConsentReject)
    cb = CircuitBuilder("c0", host_x_sk)
    assert not cb.recruit(decision)
    assert cb.hops == [], "reject must leave no circuit entry"


def test_sealed_cred_only_host_opens():
    host_x_sk, host_x_pub = x25519_keypair()
    sealed = seal_to_pubkey(host_x_pub, b"hop-credential")
    assert open_sealed(host_x_sk, sealed) == b"hop-credential"
    other_sk, _ = x25519_keypair()
    with pytest.raises(ValueError):
        open_sealed(other_sk, sealed)


def test_seal_key_matches_rust_vector():
    # Same fixed (shared, epk, recipient_pub) triple as
    # hh/src/crypto.rs::sealed_box::seal_key_matches_python_vector.
    shared = bytes(range(32))
    epk = bytes(range(32, 64))
    pub = bytes(range(64, 96))
    assert (
        _seal_key(shared, epk, pub).hex()
        == "7708606d3ae3f6c1fb103975302a77b65fc14d5363ee737a7eb8bdd0ac246e81"
    )


def test_parse_request_roundtrip():
    frame = (
        '{"_sor":{"op":"request","host_ed":"E","host_x":"X","cid":"c1",'
        '"hop":3,"nonce":"nn","sig":"ss"}}'
    )
    parsed = parse_sor_frame(frame)
    assert parsed is not None and parsed["kind"] == "request"
    r = parsed["req"]
    assert (r.host_ed_pub, r.host_x_pub, r.circuit_id, r.hop_index) == ("E", "X", "c1", 3)


@pytest.mark.parametrize(
    "text",
    ["", "not json", "{}", '{"_sor":123}', '{"_sor":{"op":42}}',
     '{"_sor":{"op":"peer"}}', '{"_sor":{"op":"unknown"}}', "[]", "null"],
)
def test_parse_never_raises(text):
    parse_sor_frame(text)  # must not raise