516 lines
22 KiB
Python
516 lines
22 KiB
Python
import struct
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
import RNS
|
|
from RNS.vendor import umsgpack
|
|
|
|
from presence_bridge import (
|
|
_MAX_ENCRYPTED_WIRE_BYTES,
|
|
_decode_qortalland_proximity_discovery,
|
|
_encode_qortalland_proximity_discovery,
|
|
)
|
|
from qortalland_games import _b58encode, canonical_bytes, derive_qortal_address, verify_signature, _b58decode
|
|
from qortalland_proximity import (
|
|
CONTROL_MAGIC,
|
|
LINK_MAGIC,
|
|
LOCAL_AUDIO_HEADER,
|
|
LOCAL_AUDIO_MAGIC,
|
|
QortalLandProximityVoiceManager,
|
|
)
|
|
|
|
|
|
class ProximityVoiceManagerTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.events = []
|
|
self.discovery = []
|
|
self.wallet = RNS.Cryptography.Ed25519PrivateKey.generate()
|
|
self.public_key = _b58encode(self.wallet.public_key().public_bytes())
|
|
self.address = derive_qortal_address(self.public_key)
|
|
self.manager = QortalLandProximityVoiceManager(
|
|
emit=lambda event, payload: self.events.append((event, payload)),
|
|
send_binary=lambda _frame, _source: True,
|
|
log=lambda _message: None,
|
|
resolve_peer=lambda _address, preferred="": preferred or None,
|
|
resolve_identity=lambda _peer: None,
|
|
build_destination=lambda identity: identity,
|
|
link_id_bytes=lambda _link: b"\0" * 16,
|
|
enqueue=lambda fn, args: bool(fn(*args) is not False),
|
|
broadcast_discovery=self.discovery.append,
|
|
verify_wallet=verify_signature,
|
|
derive_address=derive_qortal_address,
|
|
decode_base58=_b58decode,
|
|
identify_link=lambda _link: None,
|
|
)
|
|
self.manager.set_context({
|
|
"address": self.address,
|
|
"publicKey": self.public_key,
|
|
"groupId": "7",
|
|
"landSessionId": "land-1",
|
|
"roomId": "club",
|
|
"localDestinationHash": "aa" * 16,
|
|
"instanceId": "00112233-4455-4677-8899-aabbccddeeff",
|
|
})
|
|
|
|
def authorize(self):
|
|
self.manager._enable({"mode": "push-to-talk"})
|
|
fields = self.manager.pending_fields
|
|
signature = _b58encode(self.wallet.sign(canonical_bytes(fields)))
|
|
self.manager._submit_signature({"signature": signature, "publicKey": self.public_key})
|
|
|
|
def test_tick_is_idle_until_land_context_exists(self):
|
|
self.manager.context = None
|
|
self.manager._reconcile = lambda: self.fail("reconcile ran without land context")
|
|
|
|
self.manager.tick()
|
|
|
|
def test_reconcile_is_idle_without_land_context(self):
|
|
self.manager.context = None
|
|
self.manager._eligible = lambda: self.fail("eligibility ran without land context")
|
|
|
|
self.manager._reconcile()
|
|
|
|
def test_tick_reconciles_when_land_context_exists(self):
|
|
reconciled = []
|
|
self.manager._reconcile = lambda: reconciled.append(True)
|
|
|
|
self.manager.tick()
|
|
|
|
self.assertEqual(reconciled, [True])
|
|
|
|
def test_enable_requires_wallet_signature_and_disable_clears_secrets(self):
|
|
self.authorize()
|
|
self.assertTrue(self.manager.enabled)
|
|
self.assertEqual(len(self.manager.capability_hash), 32)
|
|
self.manager.disable("test")
|
|
self.assertFalse(self.manager.enabled)
|
|
self.assertIsNone(self.manager.ephemeral_private)
|
|
self.assertEqual(self.manager.capability_signature, "")
|
|
|
|
def test_outbound_link_identifies_before_classifier(self):
|
|
self.authorize()
|
|
self.manager._update_position({
|
|
"landSessionId": "land-1", "sequence": 1,
|
|
"roomId": "club", "x": 10, "y": 10,
|
|
})
|
|
order = []
|
|
self.manager.identify_link = lambda _link: order.append("identify")
|
|
|
|
class Link:
|
|
def set_packet_callback(self, _callback):
|
|
order.append("callback")
|
|
|
|
link = Link()
|
|
peer_key = "peer:session"
|
|
self.manager.links[peer_key] = {
|
|
"peerKey": peer_key,
|
|
"address": "Q-peer",
|
|
"sessionId": "peer-session",
|
|
"link": link,
|
|
"phase": "opening",
|
|
}
|
|
self.manager.links_by_object[id(link)] = peer_key
|
|
self.manager._send_packet = lambda _state, _raw: order.append("classifier") or True
|
|
|
|
self.manager._outbound_established(link)
|
|
|
|
self.assertEqual(order[0], "identify")
|
|
self.assertLess(order.index("identify"), order.index("classifier"))
|
|
|
|
def test_rejects_tampered_session_signature(self):
|
|
self.manager._enable({"mode": "push-to-talk"})
|
|
with self.assertRaisesRegex(ValueError, "invalid_proximity_signature"):
|
|
self.manager._submit_signature({
|
|
"signature": _b58encode(b"x" * 64),
|
|
"publicKey": self.public_key,
|
|
})
|
|
|
|
def test_local_binary_audio_is_unsigned_and_gated_by_transmit(self):
|
|
self.authorize()
|
|
self.manager._update_position({
|
|
"landSessionId": "land-1", "sequence": 1,
|
|
"roomId": "club", "x": 10, "y": 10,
|
|
})
|
|
payload = b"opus"
|
|
frame = LOCAL_AUDIO_HEADER.pack(
|
|
LOCAL_AUDIO_MAGIC, 1, 0, 0, self.manager.stream_generation,
|
|
1, int(time.time() * 1000), len(payload),
|
|
) + payload
|
|
self.assertTrue(self.manager.handle_local_audio(frame))
|
|
self.assertEqual(self.manager.stats["localFrames"], 0)
|
|
self.manager._set_transmit(True)
|
|
frame = LOCAL_AUDIO_HEADER.pack(
|
|
LOCAL_AUDIO_MAGIC, 1, 0, 0, self.manager.stream_generation,
|
|
2, int(time.time() * 1000), len(payload),
|
|
) + payload
|
|
self.assertTrue(self.manager.handle_local_audio(frame))
|
|
self.assertEqual(self.manager.stats["localFrames"], 1)
|
|
|
|
def test_media_drain_consumes_a_short_burst_in_one_scheduler_turn(self):
|
|
self.authorize()
|
|
self.manager._set_transmit(True)
|
|
scheduled = []
|
|
self.manager.enqueue_media = lambda fn, args: bool(scheduled.append((fn, args)) or True)
|
|
captured_at = int(time.time() * 1000)
|
|
for sequence in range(1, 13):
|
|
payload = f"opus-{sequence}".encode()
|
|
frame = LOCAL_AUDIO_HEADER.pack(
|
|
LOCAL_AUDIO_MAGIC, 1, 0, 0, self.manager.stream_generation,
|
|
sequence, captured_at, len(payload),
|
|
) + payload
|
|
self.assertTrue(self.manager.queue_local_audio(frame))
|
|
|
|
self.assertEqual(len(scheduled), 1)
|
|
fn, args = scheduled.pop()
|
|
fn(*args)
|
|
self.assertEqual(self.manager.stats["localFrames"], 12)
|
|
self.assertTrue(self.manager.local_audio_queue.empty())
|
|
self.assertEqual(scheduled, [])
|
|
|
|
def test_local_media_queue_reports_its_own_overflow_counter(self):
|
|
scheduled = []
|
|
self.manager.enqueue_media = lambda fn, args: bool(scheduled.append((fn, args)) or True)
|
|
captured_at = int(time.time() * 1000)
|
|
for sequence in range(1, 34):
|
|
payload = b"opus"
|
|
frame = LOCAL_AUDIO_HEADER.pack(
|
|
LOCAL_AUDIO_MAGIC, 1, 0, 0, self.manager.stream_generation,
|
|
sequence, captured_at, len(payload),
|
|
) + payload
|
|
self.assertTrue(self.manager.queue_local_audio(frame))
|
|
|
|
self.assertEqual(self.manager.stats["queueDrops"], 1)
|
|
self.assertEqual(self.manager.stats["localQueueDrops"], 1)
|
|
self.assertEqual(self.manager.stats["rendererQueueDrops"], 0)
|
|
|
|
def test_distance_gain_has_full_fade_and_silence_boundaries(self):
|
|
self.assertEqual(self.manager._gain(50), 1.0)
|
|
self.assertGreater(self.manager._gain(250), 0.0)
|
|
self.assertLess(self.manager._gain(250), 1.0)
|
|
self.assertEqual(self.manager._gain(400), 0.0)
|
|
|
|
def test_command_schema_rejects_extra_fields(self):
|
|
results = []
|
|
self.manager.handle_command(
|
|
{"type": "GET_PROXIMITY_STATE", "requestId": "one", "unsafe": True},
|
|
lambda *args, **kwargs: results.append((args, kwargs)),
|
|
)
|
|
self.assertFalse(results[0][0][1])
|
|
self.assertIn("schema", results[0][0][2])
|
|
|
|
def test_renderer_replacement_rotates_media_generation_and_source_ids(self):
|
|
self.authorize()
|
|
first_generation = self.manager.stream_generation
|
|
self.manager.source_ids["peer"] = 41
|
|
self.manager.renderer_connected()
|
|
self.assertNotEqual(self.manager.stream_generation, first_generation)
|
|
self.assertEqual(self.manager.source_ids, {})
|
|
|
|
def test_authoritative_shared_path_opens_peer_when_local_table_misses(self):
|
|
peer_hash = "11" * 16
|
|
peer_key = "Q-remote:land-remote"
|
|
|
|
class Destination:
|
|
hash = bytes.fromhex(peer_hash)
|
|
|
|
self.manager.remote_capabilities[peer_key] = {
|
|
"address": "Q-remote",
|
|
"sessionId": "land-remote",
|
|
"fields": {"destinationHash": peer_hash},
|
|
}
|
|
self.manager.resolve_peer = lambda _address, preferred="": preferred
|
|
self.manager.resolve_identity = lambda _peer: object()
|
|
self.manager.build_destination = lambda _identity: Destination()
|
|
self.manager.path_available = lambda _destination_hash: True
|
|
link = object()
|
|
|
|
with patch.object(RNS.Transport, "has_path", return_value=False), patch.object(
|
|
RNS, "Link", return_value=link
|
|
) as open_link, patch.object(RNS.Transport, "request_path") as request_path:
|
|
self.manager._open_peer(peer_key, 10.0)
|
|
|
|
open_link.assert_called_once()
|
|
request_path.assert_not_called()
|
|
self.assertEqual(self.manager.links[peer_key]["link"], link)
|
|
|
|
def test_malformed_proximity_classifier_is_consumed_and_closed(self):
|
|
class Link:
|
|
closed = False
|
|
|
|
def teardown(self):
|
|
self.closed = True
|
|
|
|
link = Link()
|
|
self.assertTrue(self.manager.handle_classifier(link, LINK_MAGIC + b"not-msgpack"))
|
|
self.assertTrue(link.closed)
|
|
|
|
def test_compact_discovery_fits_reticulum_and_round_trips_maximum_fields(self):
|
|
self.manager.set_context({
|
|
"address": self.address,
|
|
"publicKey": self.public_key,
|
|
"groupId": str(0x7FFFFFFF),
|
|
"landSessionId": "s" * 24,
|
|
"roomId": "r" * 64,
|
|
"localDestinationHash": "aa" * 16,
|
|
"instanceId": "00112233-4455-4677-8899-aabbccddeeff",
|
|
})
|
|
self.authorize()
|
|
wire = self.discovery[-1]
|
|
encoded = _encode_qortalland_proximity_discovery(wire)
|
|
self.assertIsNotNone(encoded)
|
|
self.assertLessEqual(len(encoded), _MAX_ENCRYPTED_WIRE_BYTES)
|
|
self.assertEqual(_decode_qortalland_proximity_discovery(encoded), wire)
|
|
|
|
def test_discovery_announcement_signature_rejects_changed_room(self):
|
|
remote_wallet = RNS.Cryptography.Ed25519PrivateKey.generate()
|
|
remote_public_key = _b58encode(remote_wallet.public_key().public_bytes())
|
|
remote_address = derive_qortal_address(remote_public_key)
|
|
remote_discovery = []
|
|
remote = QortalLandProximityVoiceManager(
|
|
emit=lambda *_args: None,
|
|
send_binary=lambda *_args: True,
|
|
log=lambda _message: None,
|
|
resolve_peer=lambda _address, preferred="": preferred or None,
|
|
resolve_identity=lambda _peer: None,
|
|
build_destination=lambda identity: identity,
|
|
link_id_bytes=lambda _link: b"\0" * 16,
|
|
enqueue=lambda fn, args: bool(fn(*args) is not False),
|
|
broadcast_discovery=remote_discovery.append,
|
|
verify_wallet=verify_signature,
|
|
derive_address=derive_qortal_address,
|
|
decode_base58=_b58decode,
|
|
)
|
|
remote.set_context({
|
|
"address": remote_address,
|
|
"publicKey": remote_public_key,
|
|
"groupId": "7",
|
|
"landSessionId": "land-1",
|
|
"roomId": "club",
|
|
"localDestinationHash": "bb" * 16,
|
|
"instanceId": "11112233-4455-4677-8899-aabbccddeeff",
|
|
})
|
|
remote._enable({"mode": "push-to-talk"})
|
|
remote._submit_signature({
|
|
"signature": _b58encode(remote_wallet.sign(canonical_bytes(remote.pending_fields))),
|
|
"publicKey": remote_public_key,
|
|
})
|
|
# A freshly signed Land capability must not depend on the separate
|
|
# presence-lease cache having caught up yet. The exact RNS endpoint is
|
|
# still authenticated when the private link handshake completes.
|
|
self.manager.resolve_peer = lambda _address, preferred="": None
|
|
encoded = _encode_qortalland_proximity_discovery(remote_discovery[-1])
|
|
self.assertIsNotNone(encoded)
|
|
decoded = _decode_qortalland_proximity_discovery(encoded)
|
|
self.assertIsNotNone(decoded)
|
|
tampered = {**decoded, "u": "another-room"}
|
|
self.assertTrue(self.manager.on_discovery(tampered, "cd" * 16))
|
|
remote_key = self.manager._peer_key(remote_address, "land-1")
|
|
self.assertNotIn(remote_key, self.manager.remote_capabilities)
|
|
self.assertTrue(self.manager.on_discovery(decoded, "cd" * 16))
|
|
self.assertIn(remote_key, self.manager.remote_capabilities)
|
|
|
|
def test_same_account_land_sessions_keep_independent_routes_and_audio_sources(self):
|
|
self.authorize()
|
|
self.manager._update_position({
|
|
"landSessionId": "land-1", "sequence": 1,
|
|
"roomId": "club", "x": 0, "y": 0,
|
|
})
|
|
resolved = []
|
|
self.manager.resolve_peer = lambda address, preferred="": (
|
|
resolved.append((address, preferred)) or preferred
|
|
)
|
|
|
|
for session_id, destination_hash, x in (
|
|
("land-2", "bb" * 16, 20),
|
|
("land-3", "cc" * 16, 40),
|
|
):
|
|
discovery = []
|
|
remote = QortalLandProximityVoiceManager(
|
|
emit=lambda *_args: None,
|
|
send_binary=lambda *_args: True,
|
|
log=lambda _message: None,
|
|
resolve_peer=lambda _address, preferred="": preferred or None,
|
|
resolve_identity=lambda _peer: None,
|
|
build_destination=lambda identity: identity,
|
|
link_id_bytes=lambda _link: b"\0" * 16,
|
|
enqueue=lambda fn, args: bool(fn(*args) is not False),
|
|
broadcast_discovery=discovery.append,
|
|
verify_wallet=verify_signature,
|
|
derive_address=derive_qortal_address,
|
|
decode_base58=_b58decode,
|
|
)
|
|
remote.set_context({
|
|
"address": self.address,
|
|
"publicKey": self.public_key,
|
|
"groupId": "7",
|
|
"landSessionId": session_id,
|
|
"roomId": "club",
|
|
"localDestinationHash": destination_hash,
|
|
"instanceId": f"{x:08x}-4455-4677-8899-aabbccddeeff",
|
|
})
|
|
remote._enable({"mode": "push-to-talk"})
|
|
remote._submit_signature({
|
|
"signature": _b58encode(self.wallet.sign(canonical_bytes(remote.pending_fields))),
|
|
"publicKey": self.public_key,
|
|
})
|
|
decoded = _decode_qortalland_proximity_discovery(
|
|
_encode_qortalland_proximity_discovery(discovery[-1])
|
|
)
|
|
self.assertTrue(self.manager.on_discovery(decoded, "dd" * 16))
|
|
self.manager.on_land_state({
|
|
"a": self.address, "s": session_id, "g": "7", "u": "club",
|
|
"x": x, "y": 0, "ts": int(time.time() * 1000),
|
|
}, "dd" * 16)
|
|
|
|
keys = [
|
|
self.manager._peer_key(self.address, "land-2"),
|
|
self.manager._peer_key(self.address, "land-3"),
|
|
]
|
|
self.assertEqual(set(self.manager.remote_capabilities), set(keys))
|
|
self.assertEqual(set(self.manager.remote_positions), set(keys))
|
|
self.assertEqual({peer_key for _distance, peer_key in self.manager._eligible()}, set(keys))
|
|
self.assertNotEqual(self.manager._source_id(keys[0]), self.manager._source_id(keys[1]))
|
|
self.assertIn((self.address, "bb" * 16), resolved)
|
|
self.assertIn((self.address, "cc" * 16), resolved)
|
|
|
|
def test_link_hello_is_bound_to_both_land_sessions_and_authenticated_endpoint(self):
|
|
self.authorize()
|
|
self.manager._update_position({
|
|
"landSessionId": "land-1", "sequence": 1,
|
|
"roomId": "club", "x": 0, "y": 0,
|
|
})
|
|
destination_hash = "bb" * 16
|
|
discovery = []
|
|
remote = QortalLandProximityVoiceManager(
|
|
emit=lambda *_args: None,
|
|
send_binary=lambda *_args: True,
|
|
log=lambda _message: None,
|
|
resolve_peer=lambda _address, preferred="": preferred or None,
|
|
resolve_identity=lambda _peer: None,
|
|
build_destination=lambda identity: identity,
|
|
link_id_bytes=lambda _link: b"l" * 16,
|
|
enqueue=lambda fn, args: bool(fn(*args) is not False),
|
|
broadcast_discovery=discovery.append,
|
|
verify_wallet=verify_signature,
|
|
derive_address=derive_qortal_address,
|
|
decode_base58=_b58decode,
|
|
)
|
|
remote.set_context({
|
|
"address": self.address, "publicKey": self.public_key,
|
|
"groupId": "7", "landSessionId": "land-0", "roomId": "club",
|
|
"localDestinationHash": destination_hash,
|
|
"instanceId": "11112233-4455-4677-8899-aabbccddeeff",
|
|
})
|
|
remote._enable({"mode": "push-to-talk"})
|
|
remote._submit_signature({
|
|
"signature": _b58encode(self.wallet.sign(canonical_bytes(remote.pending_fields))),
|
|
"publicKey": self.public_key,
|
|
})
|
|
decoded = _decode_qortalland_proximity_discovery(
|
|
_encode_qortalland_proximity_discovery(discovery[-1])
|
|
)
|
|
self.manager.resolve_peer = lambda address, preferred="": (
|
|
preferred if address == self.address and preferred == destination_hash else None
|
|
)
|
|
self.manager.resolve_link_peer_hash = lambda _link: destination_hash
|
|
self.assertTrue(self.manager.on_discovery(decoded, "dd" * 16))
|
|
self.manager.on_land_state({
|
|
"a": self.address, "s": "land-0", "g": "7", "u": "club",
|
|
"x": 20, "y": 0, "ts": int(time.time() * 1000),
|
|
}, "dd" * 16)
|
|
peer_key = self.manager._peer_key(self.address, "land-0")
|
|
capability = self.manager.remote_capabilities[peer_key]
|
|
link = object()
|
|
hello = {
|
|
"v": 1, "f": self.address, "t": self.address, "g": "7",
|
|
"s": "land-0", "o": "land-1", "r": "club",
|
|
"c": capability["hash"], "l": b"\0" * 16,
|
|
"n": b"n" * 16, "ts": int(time.time() * 1000),
|
|
}
|
|
hello["z"] = remote.ephemeral_private.sign(umsgpack.packb(hello))
|
|
self.assertTrue(self.manager._verify_link_hello(link, hello))
|
|
|
|
wrong_target = {**hello, "o": "land-other", "n": b"o" * 16}
|
|
signed_wrong_target = dict(wrong_target)
|
|
signed_wrong_target.pop("z")
|
|
wrong_target["z"] = remote.ephemeral_private.sign(umsgpack.packb(signed_wrong_target))
|
|
self.assertFalse(self.manager._verify_link_hello(link, wrong_target))
|
|
|
|
self.manager.resolve_link_peer_hash = lambda _link: "cc" * 16
|
|
wrong_link = {**hello, "n": b"p" * 16}
|
|
signed_wrong_link = dict(wrong_link)
|
|
signed_wrong_link.pop("z")
|
|
wrong_link["z"] = remote.ephemeral_private.sign(umsgpack.packb(signed_wrong_link))
|
|
self.assertFalse(self.manager._verify_link_hello(link, wrong_link))
|
|
|
|
def test_inbound_link_retries_accept_until_optional_auth_ack(self):
|
|
self.authorize()
|
|
link = object()
|
|
address = "Q" + "p" * 33
|
|
link_id = b"l" * 16
|
|
nonce = b"n" * 16
|
|
accept = {"c": "accept", "marker": "test"}
|
|
state = {
|
|
"peerKey": f"{address}:land-2", "address": address, "sessionId": "land-2",
|
|
"link": link,
|
|
"linkId": link_id,
|
|
"nonce": nonce,
|
|
"phase": "connected",
|
|
"authenticated": True,
|
|
"authAccept": accept,
|
|
"lastAuthAccept": time.time() - 2,
|
|
"authAcceptAttempts": 1,
|
|
"createdAt": time.time(),
|
|
"lastActivity": time.time(),
|
|
"sourceId": 1,
|
|
}
|
|
self.manager.links[state["peerKey"]] = state
|
|
self.manager.links_by_object[id(link)] = state["peerKey"]
|
|
sent = []
|
|
self.manager._send_control = lambda _state, payload: sent.append(payload)
|
|
self.manager._reconcile = lambda: None
|
|
|
|
self.manager.tick()
|
|
self.assertIn(accept, sent)
|
|
self.assertTrue(state["authenticated"])
|
|
|
|
packet = type("Packet", (), {"link": link})()
|
|
ack = {
|
|
"v": 1, "ts": int(time.time() * 1000), "c": "auth_ack",
|
|
"l": link_id, "n": nonce,
|
|
}
|
|
self.manager._on_packet(CONTROL_MAGIC + umsgpack.packb(ack), packet)
|
|
self.assertTrue(state["authenticated"])
|
|
self.assertEqual(state["phase"], "connected")
|
|
self.assertNotIn("authAccept", state)
|
|
|
|
def test_signed_control_preserves_messagepack_key_order(self):
|
|
payload = {
|
|
"v": 1, "a": self.address, "c": "accept", "f": self.address,
|
|
"t": self.address, "h": b"h" * 32, "q": b"q" * 32,
|
|
"l": b"l" * 16, "n": b"n" * 16, "r": b"r" * 16,
|
|
"ts": int(time.time() * 1000), "z": b"z" * 64,
|
|
}
|
|
captured = []
|
|
|
|
class Packet:
|
|
def __init__(self, _link, raw):
|
|
captured.append(raw)
|
|
|
|
def send(self):
|
|
return None
|
|
|
|
with patch("qortalland_proximity.RNS.Packet", Packet):
|
|
self.manager._send_control({"link": object()}, payload)
|
|
|
|
decoded = umsgpack.unpackb(captured[0][len(CONTROL_MAGIC):])
|
|
self.assertEqual(list(decoded.keys()), list(payload.keys()))
|
|
signed = dict(decoded)
|
|
signed.pop("z")
|
|
self.assertEqual(umsgpack.packb(signed), umsgpack.packb({key: value for key, value in payload.items() if key != "z"}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|