R5: consent-gated recruitment + per-hop X25519 sealed credentials
Add the in-band, signature-gated consent handshake that governs whether a
node may be recruited into a measured SOR circuit, plus a per-recipient
sealed box so a hop credential decrypts only with the host's X25519 key.
- crypto.rs: seal_to_pubkey/open_sealed (ephemeral X25519 -> HKDF-SHA256 ->
Fernet AEAD), reusing the existing fernet crate (no new symmetric primitive).
- sor/consent.rs: signed ConsentRequest (Ed25519 persona), node_evaluate
(reject unless the signature verifies), CircuitBuilder that recruits a hop
only on an accept it can open; parse_sor_frame with never-panic proptests.
- net.rs: {"_sor":...} control frames are live-only and classified out-of-band
by the SOR layer (parse_sor), never surfaced as chat/app events.
- cmd_chat/sor/consent.py + bridge _sor case: bit-compatible Python mirror;
the agent only observes/classifies consent frames — it never auto-accepts
and stands up no forwarder (forwarding is R4, isolated-engine-only).
Acceptance check (roadmap R5) green both languages: host recruits a hop only
after an explicit accept; a reject leaves no entry; the credential opens only
with the host key (third party cannot); unsigned/forged request -> rejected.
Cross-language KDF parity pinned by a shared known-answer vector.
Rust 75 passed; Python SOR suite 37 passed.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,141 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user