extends SceneTree
## LIVE probe for PeerIdentity (hyper-games #89 §3.3): TWO real keet-daemons on
## scratch storage over TCP (keet-daemon test/harness-tcp-scratch.mjs, twice).
## Peer A's PeerIdentity attests A's mesh key with daemon A; peer B's PeerIdentity
## verifies that claim with daemon B — real proofs, real crypto, no stub. Then a
## forged claim (B's own proof claiming to be A, and A's proof from another key).
##
## LIVE_A=127.0.0.1:<portA>,<tokenA> LIVE_B=127.0.0.1:<portB>,<tokenB>
const PeerIdentity := preload("res://addons/keet_social/keet_peer_identity.gd")
const PK_A := "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1"
const PK_B := "b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2"
const PK_EVE := "e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3e3"
var ok := true
var _sent_a: Array = []
var _sent_b: Array = []
func _p(c: bool, m: String) -> void:
print(("ok " if c else "FAIL: ") + m)
if not c:
ok = false
func _init() -> void:
_main.call_deferred()
func _client_for(env: String) -> Node:
var parts := OS.get_environment(env).split(",")
OS.set_environment("HYPERTPS_DAEMON_TCP", parts[0])
OS.set_environment("HYPERTPS_DAEMON_TOKEN", parts[1])
var c: Node = KeetUdsClient.new()
root.add_child(c) # _ready reads the env NOW
return c
func _record_a(id: int, m: String, p: String) -> void:
_sent_a.append([id, m, p])
func _record_b(id: int, m: String, p: String) -> void:
_sent_b.append([id, m, p])
func _connected(c: Node) -> bool:
return bool(c.get("is_connected"))
func _wait(cond: Callable, seconds: float) -> bool:
var end := Time.get_ticks_msec() + int(seconds * 1000.0)
while Time.get_ticks_msec() < end:
if cond.call():
return true
await process_frame
return cond.call()
func _main() -> void:
var ca := _client_for("LIVE_A")
var cb := _client_for("LIVE_B")
_p(await _wait(_connected.bind(ca), 20.0) and await _wait(_connected.bind(cb), 20.0), "both scratch daemons connected over TCP")
var pa: Node = PeerIdentity.new()
pa.client = ca
pa.send_claim = _record_a
root.add_child(pa)
var pb: Node = PeerIdentity.new()
pb.client = cb
pb.send_claim = _record_b
root.add_child(pb)
# Each side attests its own mesh key with its own daemon.
pa.set_my_dht_key(PK_A)
pb.set_my_dht_key(PK_B)
_p(await _wait(func(): return not pa._my_claim.is_empty() and not pb._my_claim.is_empty(), 60.0), "both attested their mesh keys")
var claim_a: Dictionary = pa._my_claim
var claim_b: Dictionary = pb._my_claim
_p(claim_a.memberId != claim_b.memberId, "two different keet members (%s / %s)" % [String(claim_a.memberId).left(8), String(claim_b.memberId).left(8)])
pa._on_peer_connected(2)
_p(_sent_a.size() == 1 and _sent_a[0][1] == claim_a.memberId, "A sends its claim to a new connection")
# B receives A's claim on peer id 1, whose authenticated mesh key is PK_A.
pb.set_roster([{"memberId": claim_a.memberId, "name": "alice-live"}])
pb.set_peer_key(1, PK_A)
pb.receive_claim(1, claim_a.memberId, claim_a.proof)
_p(await _wait(func(): return pb.member_of(1) == claim_a.memberId, 30.0), "B verified A's REAL proof with ITS OWN daemon")
_p(pb.name_for_peer(1) == "alice-live", "B's label for A is A's roster name (got %s)" % pb.name_for_peer(1))
# Forgeries, verified by daemon B with real crypto.
pb.set_peer_key(3, PK_EVE)
pb.receive_claim(3, claim_a.memberId, claim_a.proof)
await _wait(func(): return false, 2.0)
_p(pb.member_of(3) == "", "A's real proof replayed from another mesh key is refused")
pb.set_peer_key(4, PK_B)
pb.receive_claim(4, claim_a.memberId, claim_b.proof)
await _wait(func(): return false, 2.0)
_p(pb.member_of(4) == "", "B's real proof claiming to be A is refused")
_p(pb.name_for_peer(4).begins_with("anon-"), "the forger stays anon (got %s)" % pb.name_for_peer(4))
print("PROBE peer_identity_live: %s" % ("PASS" if ok else "FAIL"))
quit(0 if ok else 1)