Qortal/Qortal-Hub
1
0
Fork 1
mirror of https://github.com/Qortal/Qortal-Hub.git synced 2026-08-25 08:56:41 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
Qortal-Hub/electron/resources/qortalland_proximity.py
T

1545 lines
74 KiB
Python
Raw Permalink Blame History

"""Qortal Land proximity voice over authenticated, dedicated Reticulum links."""
from __future__ import annotations
import hashlib
import math
import queue
import secrets
import struct
import threading
import time
import uuid
from typing import Any, Callable, Dict, Optional
import RNS
from RNS.vendor import umsgpack
DISCOVERY_TYPE = "QLPV1"
LINK_MAGIC = b"QLP1"
MEDIA_MAGIC = b"PVA1"
CONTROL_MAGIC = b"PVC1"
LOCAL_AUDIO_MAGIC = b"QLA1"
PROTOCOL_VERSION = 1
MAX_PEERS = 7
MAX_OPUS_BYTES = 320
MAX_LOCAL_AUDIO_BYTES = 2 * 1024
FULL_VOLUME_DISTANCE = 100.0
AUDIBLE_DISTANCE = 400.0
AUDIBLE_EXIT_DISTANCE = 460.0
PRECONNECT_DISTANCE = 520.0
RELEASE_DISTANCE = 600.0
LAND_STATE_MAX_AGE = 5.0
DISCOVERY_MAX_AGE = 25.0
HEARTBEAT_INTERVAL = 5.0
LINK_DEAD_AFTER = 15.0
LINK_TIMEOUT = 30.0
CAPABILITY_MAX_AGE_MS = 4 * 60 * 60 * 1000
CAPABILITY_CLOCK_SKEW_MS = 2 * 60 * 1000
LOCAL_AUDIO_HEADER = struct.Struct(">4sBBHIIQH")
RNS_AUDIO_HEADER = struct.Struct(">4sBBIIIH")
MEDIA_DRAIN_MAX_FRAMES = 24
MEDIA_DRAIN_TIME_BUDGET = 0.012
PROXIMITY_COMMANDS = {
"ENABLE_PROXIMITY_VOICE",
"SUBMIT_PROXIMITY_SESSION_SIGNATURE",
"DISABLE_PROXIMITY_VOICE",
"UPDATE_PROXIMITY_POSITION",
"SET_PROXIMITY_TRANSMIT",
"SET_PROXIMITY_SUSPENDED",
"SET_PROXIMITY_PEER_POLICY",
"GET_PROXIMITY_STATE",
"GET_PROXIMITY_DIAGNOSTICS",
}
COMMAND_FIELDS = {
"ENABLE_PROXIMITY_VOICE": ({"type", "requestId", "mode"}, {"type", "requestId"}),
"SUBMIT_PROXIMITY_SESSION_SIGNATURE": ({"type", "requestId", "signature", "publicKey"}, {"type", "requestId", "signature", "publicKey"}),
"DISABLE_PROXIMITY_VOICE": ({"type", "requestId"}, {"type", "requestId"}),
"UPDATE_PROXIMITY_POSITION": ({"type", "requestId", "landSessionId", "sequence", "roomId", "x", "y"}, {"type", "requestId", "landSessionId", "sequence", "roomId", "x", "y"}),
"SET_PROXIMITY_TRANSMIT": ({"type", "requestId", "transmitting", "mode"}, {"type", "requestId", "transmitting"}),
"SET_PROXIMITY_SUSPENDED": ({"type", "requestId", "suspended"}, {"type", "requestId", "suspended"}),
"SET_PROXIMITY_PEER_POLICY": ({"type", "requestId", "address", "sessionId", "muted", "volume", "blocked"}, {"type", "requestId", "address"}),
"GET_PROXIMITY_STATE": ({"type", "requestId"}, {"type", "requestId"}),
"GET_PROXIMITY_DIAGNOSTICS": ({"type", "requestId"}, {"type", "requestId"}),
}
def _canonical(fields: Dict[str, Any]) -> bytes:
import json
return json.dumps(fields, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def _safe_close(link) -> None:
if link is None:
return
try:
link.teardown()
except Exception:
try:
link.close()
except Exception:
pass
class QortalLandProximityVoiceManager:
def __init__(
self,
emit: Callable[[str, Dict[str, Any]], None],
send_binary: Callable[[bytes, int], bool],
log: Callable[[str], None],
resolve_peer: Callable[[str, str], Optional[str]],
resolve_identity: Callable[[str], Any],
build_destination: Callable[[Any], Any],
link_id_bytes: Callable[[Any], bytes],
enqueue: Callable[[Callable[..., Any], tuple], bool],
broadcast_discovery: Callable[[Dict[str, Any]], None],
verify_wallet: Callable[[Dict[str, Any], str, str], bool],
derive_address: Callable[[str], str],
decode_base58: Callable[[str], bytes],
enqueue_media: Optional[Callable[[Callable[..., Any], tuple], bool]] = None,
resolve_link_peer_hash: Optional[Callable[[Any], str]] = None,
identify_link: Optional[Callable[[Any], None]] = None,
path_available: Optional[Callable[[bytes], bool]] = None,
refresh_path: Optional[Callable[[str, str], bool]] = None,
):
self.emit = emit
self.send_binary = send_binary
self.log = log
self.resolve_peer = resolve_peer
self.resolve_identity = resolve_identity
self.build_destination = build_destination
self.link_id_bytes = link_id_bytes
self.enqueue = enqueue
self.enqueue_media = enqueue_media or enqueue
self.broadcast_discovery = broadcast_discovery
self.verify_wallet = verify_wallet
self.derive_address = derive_address
self.decode_base58 = decode_base58
self.resolve_link_peer_hash = resolve_link_peer_hash
self.identify_link = identify_link
self.path_available = path_available or RNS.Transport.has_path
self.refresh_path = refresh_path
self.lock = threading.RLock()
self.context: Optional[Dict[str, Any]] = None
self.enabled = False
self.suspended = False
self.suspended_at: Optional[float] = None
self.transmitting = False
self.mode = "push-to-talk"
self.position: Optional[Dict[str, Any]] = None
self.position_sequence = -1
self.pending_position_command: Optional[tuple[Dict[str, Any], Callable[..., None]]] = None
self.position_drain_scheduled = False
self.ephemeral_private = None
self.pending_ephemeral_private = None
self.pending_fields: Optional[Dict[str, Any]] = None
self.capability: Optional[Dict[str, Any]] = None
self.capability_signature = ""
self.capability_hash = b""
self.remote_capabilities: Dict[str, Dict[str, Any]] = {}
self.remote_positions: Dict[str, Dict[str, Any]] = {}
self.links: Dict[str, Dict[str, Any]] = {}
self.links_by_object: Dict[int, str] = {}
self.source_ids: Dict[str, int] = {}
self.next_source_id = 1
self.stream_generation = secrets.randbits(31) or 1
self.local_media_max_sequence = -1
self.local_audio_queue: "queue.Queue[tuple[float, bytes]]" = queue.Queue(maxsize=32)
self.media_drain_scheduled = False
self.used_link_nonces: Dict[bytes, float] = {}
self.blocked_addresses: set[str] = set()
self.path_requested_at: Dict[str, float] = {}
self.link_retry: Dict[str, Dict[str, float]] = {}
self.replacement_since: Dict[str, float] = {}
self.visible_capacity_peers: set[str] = set()
self.last_discovery_at = 0.0
self.last_stats_at = 0.0
self.last_media_log_at = 0.0
self.capacity_reduced_until = 0.0
self.renderer_lost_at: Optional[float] = None
self.diagnostic_last: Dict[str, float] = {}
self.stats = {
"localFrames": 0, "sentFrames": 0, "receivedFrames": 0,
"staleDrops": 0, "queueDrops": 0, "invalidFrames": 0,
"localQueueDrops": 0, "rendererQueueDrops": 0,
"duplicateDrops": 0, "sequenceSkips": 0, "linkFailures": 0,
}
def _trace(self, stage: str, address: str = "", code: str = "", throttle: float = 0.0) -> None:
key = f"{stage}:{address}:{code}"
now = time.time()
if throttle > 0 and now - self.diagnostic_last.get(key, 0.0) < throttle:
return
self.diagnostic_last[key] = now
peer = hashlib.sha256(str(address or "unknown").encode("utf-8")).hexdigest()[:10]
suffix = f" code={str(code)[:48]}" if code else ""
self.log(f"[qortalland-proximity] stage={stage} peer={peer}{suffix}")
@staticmethod
def _peer_key(address: str, session_id: str) -> str:
return f"{str(address or '').strip()}:{str(session_id or '').strip()}"
def set_context(self, context: Dict[str, Any]) -> None:
previous = self.context
if previous and (
previous.get("address") != context.get("address")
or previous.get("groupId") != context.get("groupId")
or previous.get("landSessionId") != context.get("landSessionId")
or previous.get("localDestinationHash") != context.get("localDestinationHash")
):
self.disable("land_context_changed")
self.context = dict(context)
def clear_context(self) -> None:
self.disable("land_context_cleared")
self.context = None
def handle_command(self, message: Dict[str, Any], command_result: Callable[..., None]) -> None:
command = str(message.get("type") or "")
request_id = message.get("requestId")
try:
allowed, required = COMMAND_FIELDS.get(command, (set(), set()))
if (
not isinstance(request_id, str) or not request_id or len(request_id) > 80
or not required.issubset(message.keys()) or not set(message.keys()).issubset(allowed)
):
raise ValueError("invalid_proximity_command_schema")
if command == "ENABLE_PROXIMITY_VOICE":
self._enable(message)
elif command == "SUBMIT_PROXIMITY_SESSION_SIGNATURE":
self._submit_signature(message)
elif command == "DISABLE_PROXIMITY_VOICE":
self.disable("user_disabled")
elif command == "UPDATE_PROXIMITY_POSITION":
self._update_position(message)
elif command == "SET_PROXIMITY_TRANSMIT":
if not isinstance(message.get("transmitting"), bool):
raise ValueError("invalid_proximity_transmit")
if "mode" in message:
mode = str(message.get("mode") or "")
if mode not in {"push-to-talk", "open-mic"}:
raise ValueError("invalid_proximity_mode")
self.mode = mode
self._set_transmit(message.get("transmitting") is True)
elif command == "SET_PROXIMITY_SUSPENDED":
if not isinstance(message.get("suspended"), bool):
raise ValueError("invalid_proximity_suspension")
self.suspended = message.get("suspended") is True
self.suspended_at = time.time() if self.suspended else None
if self.suspended:
self.transmitting = False
self._send_control_all({"c": "pause" if self.suspended else "resume"})
self._emit_state()
self._broadcast(True)
elif command == "SET_PROXIMITY_PEER_POLICY":
self._set_peer_policy(message)
elif command in {"GET_PROXIMITY_STATE", "GET_PROXIMITY_DIAGNOSTICS"}:
self._emit_snapshot(diagnostics=command.endswith("DIAGNOSTICS"))
else:
raise ValueError("invalid_proximity_command")
command_result(request_id, True)
except Exception as exc:
command_result(request_id, False, str(exc)[:160])
def queue_position_command(self, message: Dict[str, Any], command_result: Callable[..., None]) -> bool:
with self.lock:
previous = self.pending_position_command
self.pending_position_command = (dict(message), command_result)
if previous is not None:
previous_message, previous_result = previous
previous_result(previous_message.get("requestId"), True, payload={"coalesced": True})
if self.position_drain_scheduled:
return True
self.position_drain_scheduled = True
if self.enqueue(self._drain_position_command, ()):
return True
with self.lock:
pending = self.pending_position_command
self.pending_position_command = None
self.position_drain_scheduled = False
if pending is not None:
pending[1](pending[0].get("requestId"), False, "command_queue_full")
return False
def _drain_position_command(self) -> None:
with self.lock:
pending = self.pending_position_command
self.pending_position_command = None
self.position_drain_scheduled = False
if pending is not None:
self.handle_command(pending[0], pending[1])
def _enable(self, message: Dict[str, Any]) -> None:
if not self.context:
raise ValueError("land_context_required")
try:
group_id = int(self.context.get("groupId") or 0)
except (TypeError, ValueError):
group_id = 0
if not 0 < group_id <= 0x7FFFFFFF:
raise ValueError("invalid_proximity_group")
mode = str(message.get("mode") or "push-to-talk")
if mode not in {"push-to-talk", "open-mic"}:
raise ValueError("invalid_proximity_mode")
self.mode = mode
self.pending_ephemeral_private = RNS.Cryptography.Ed25519PrivateKey.generate()
ephemeral_public = self.pending_ephemeral_private.public_key().public_bytes().hex()
now_ms = int(time.time() * 1000)
self.pending_fields = {
"type": "QORTAL_LAND_PROXIMITY_VOICE_SESSION",
"protocolVersion": PROTOCOL_VERSION,
"address": self.context["address"],
"signerPublicKey": self.context["publicKey"],
"ephemeralPublicKey": ephemeral_public,
"groupId": self.context["groupId"],
"landSessionId": self.context["landSessionId"],
"destinationHash": self.context["localDestinationHash"],
"instanceId": self.context.get("instanceId", ""),
# The fresh ephemeral key already provides a unique 256-bit value.
# Reuse it as the signed nonce so compact discovery does not have to
# carry a second redundant 32-byte random value.
"nonce": ephemeral_public,
"createdAt": now_ms,
"expiresAt": now_ms + CAPABILITY_MAX_AGE_MS,
}
self.emit("PROXIMITY_SIGNATURE_REQUIRED", {"fields": dict(self.pending_fields)})
self.emit("PROXIMITY_STATE", {"state": "authorizing", "mode": self.mode})
def _submit_signature(self, message: Dict[str, Any]) -> None:
fields = self.pending_fields
signature = str(message.get("signature") or "")
public_key = str(message.get("publicKey") or "")
if (
not fields or not signature or len(signature) > 128
or len(public_key) > 64 or public_key != fields.get("signerPublicKey")
):
raise ValueError("unexpected_proximity_signature")
if self.derive_address(public_key) != fields.get("address"):
raise ValueError("proximity_signer_mismatch")
if not self.verify_wallet(fields, public_key, signature):
raise ValueError("invalid_proximity_signature")
if self.pending_ephemeral_private is None:
raise ValueError("missing_proximity_ephemeral_key")
self.ephemeral_private = self.pending_ephemeral_private
self.pending_ephemeral_private = None
self.capability = dict(fields)
self.capability_signature = signature
self.capability_hash = hashlib.sha256(_canonical(fields) + self.decode_base58(signature)).digest()
self.pending_fields = None
self.enabled = True
self.suspended = False
self.suspended_at = None
self.stream_generation = (self.stream_generation + 1) & 0x7FFFFFFF or 1
self.local_media_max_sequence = -1
self._broadcast(True)
self._emit_state()
def disable(self, reason: str) -> None:
was_enabled = self.enabled
if was_enabled:
self._broadcast(False)
with self.lock:
states = list(self.links.values())
self.links.clear()
self.links_by_object.clear()
for state in states:
_safe_close(state.get("link"))
self.enabled = False
self.suspended = False
self.suspended_at = None
self.transmitting = False
self.pending_fields = None
self.pending_ephemeral_private = None
self.capability = None
self.capability_signature = ""
self.capability_hash = b""
self.ephemeral_private = None
self.position = None
self.position_sequence = -1
self.local_media_max_sequence = -1
with self.lock:
pending_position = self.pending_position_command
self.pending_position_command = None
self.position_drain_scheduled = False
if pending_position is not None:
pending_position[1](pending_position[0].get("requestId"), False, "proximity_disabled")
self.source_ids.clear()
self.next_source_id = 1
self.remote_capabilities.clear()
self.remote_positions.clear()
self.used_link_nonces.clear()
self.path_requested_at.clear()
self.link_retry.clear()
self.replacement_since.clear()
self.visible_capacity_peers.clear()
self.diagnostic_last.clear()
while True:
try:
self.local_audio_queue.get_nowait()
except queue.Empty:
break
self.media_drain_scheduled = False
self.renderer_lost_at = None
self.emit("PROXIMITY_STATE", {"state": "off", "reason": reason, "mode": self.mode})
def renderer_connected(self) -> None:
self.renderer_lost_at = None
self.stream_generation = (self.stream_generation + 1) & 0x7FFFFFFF or 1
self.local_media_max_sequence = -1
self.source_ids.clear()
self.next_source_id = 1
for peer_key, state in self.links.items():
state["sourceId"] = self._source_id(peer_key)
self._emit_peer(peer_key, state)
def renderer_lost(self) -> None:
if self.enabled and self.renderer_lost_at is None:
self.renderer_lost_at = time.time()
self.transmitting = False
self._send_control_all({"c": "talk", "a": False})
def _update_position(self, message: Dict[str, Any]) -> None:
if not self.context:
raise ValueError("land_context_required")
sequence = message.get("sequence")
x, y = message.get("x"), message.get("y")
room = str(message.get("roomId") or "")
if (
not isinstance(sequence, int) or isinstance(sequence, bool)
or sequence <= self.position_sequence
or not isinstance(x, (int, float)) or isinstance(x, bool)
or not isinstance(y, (int, float)) or isinstance(y, bool)
or not math.isfinite(float(x)) or not math.isfinite(float(y))
or not room or len(room) > 64
or str(message.get("landSessionId") or "") != self.context["landSessionId"]
):
raise ValueError("invalid_proximity_position")
old_room = self.position.get("roomId") if self.position else None
self.position_sequence = sequence
self.position = {"roomId": room, "x": float(x), "y": float(y), "at": time.time()}
if old_room != room:
self._broadcast(True)
self._reconcile()
def _set_transmit(self, transmitting: bool) -> None:
self.transmitting = bool(transmitting and self.enabled and not self.suspended)
self._send_control_all({"c": "talk", "a": self.transmitting})
self.emit("PROXIMITY_SPEAKING_STATE", {
"address": self.context.get("address") if self.context else "",
"speaking": self.transmitting,
})
def _set_peer_policy(self, message: Dict[str, Any]) -> None:
address = str(message.get("address") or "")
session_id = str(message.get("sessionId") or "")
if len(address) < 20 or len(address) > 64:
raise ValueError("invalid_peer_address")
if "blocked" in message and not isinstance(message.get("blocked"), bool):
raise ValueError("invalid_peer_block_policy")
if "muted" in message and not isinstance(message.get("muted"), bool):
raise ValueError("invalid_peer_mute_policy")
if message.get("blocked") is True:
self.blocked_addresses.add(address)
for peer_key, capability in list(self.remote_capabilities.items()):
if capability.get("address") == address:
self._drop_remote(peer_key, "blocked")
return
if message.get("blocked") is False:
self.blocked_addresses.discard(address)
volume = message.get("volume", 1.0)
if not isinstance(volume, (int, float)) or not math.isfinite(float(volume)):
raise ValueError("invalid_peer_volume")
targets = [self._peer_key(address, session_id)] if session_id else [
key for key, state in self.links.items() if state.get("address") == address
]
for peer_key in targets:
state = self.links.get(peer_key)
if not state:
continue
state["muted"] = message.get("muted") is True
state["volume"] = max(0.0, min(2.0, float(volume)))
self._emit_peer(peer_key, state)
def _broadcast(self, enabled: bool) -> None:
if not self.context:
return
wire: Dict[str, Any] = {
"t": DISCOVERY_TYPE,
"v": PROTOCOL_VERSION,
"e": bool(enabled and self.enabled),
"a": self.context["address"],
"g": self.context["groupId"],
"s": self.context["landSessionId"],
"u": self.position.get("roomId") if self.position else self.context.get("roomId", ""),
"b": bool(self.suspended),
"ts": int(time.time() * 1000),
"p": 0,
}
if self.capability:
wire["c"] = self.capability
wire["z"] = self.capability_signature
wire["h"] = self.capability_hash.hex()
if self.ephemeral_private and wire.get("h"):
signed_announcement = {
key: wire[key]
for key in ("t", "v", "e", "a", "g", "s", "u", "b", "ts", "h")
}
wire["j"] = self.ephemeral_private.sign(_canonical(signed_announcement)).hex()
self.broadcast_discovery(wire)
self.last_discovery_at = time.time()
def on_discovery(self, wire: Dict[str, Any], peer_hash: str) -> bool:
if wire.get("t") != DISCOVERY_TYPE:
return False
try:
required_wire = {"t", "v", "e", "a", "g", "s", "u", "b", "ts", "p", "c", "z", "h", "j"}
if (
set(wire.keys()) != required_wire
or wire.get("v") != PROTOCOL_VERSION
or not isinstance(wire.get("e"), bool)
or not isinstance(wire.get("b"), bool)
or not isinstance(wire.get("p"), int) or isinstance(wire.get("p"), bool)
or not 0 <= wire["p"] <= 3
or not isinstance(wire.get("u"), str) or not 0 < len(wire["u"]) <= 64
or not isinstance(wire.get("h"), str) or len(wire["h"]) != 64
or not isinstance(wire.get("j"), str) or len(wire["j"]) != 128
):
return True
address = str(wire.get("a") or "")
session_id = str(wire.get("s") or "")
peer_key = self._peer_key(address, session_id)
local_key = self._peer_key(
self.context.get("address") if self.context else "",
self.context.get("landSessionId") if self.context else "",
)
if not address or not session_id or not self.context or peer_key == local_key:
return True
if str(wire.get("g") or "") != str(self.context.get("groupId") or ""):
return True
timestamp = wire.get("ts")
if (
not isinstance(timestamp, int) or isinstance(timestamp, bool)
or abs(int(time.time() * 1000) - timestamp) > 30_000
):
return True
fields = wire.get("c")
signature = str(wire.get("z") or "")
if not isinstance(fields, dict) or not self._valid_remote_capability(fields, signature, address):
return True
if (
str(fields.get("groupId") or "") != str(wire.get("g") or "")
or str(fields.get("landSessionId") or "") != str(wire.get("s") or "")
):
return True
expected_hash = hashlib.sha256(_canonical(fields) + self.decode_base58(signature)).digest()
if str(wire.get("h") or "") != expected_hash.hex():
return True
announcement_signature = wire.get("j")
signed_announcement = {
key: wire.get(key)
for key in ("t", "v", "e", "a", "g", "s", "u", "b", "ts", "h")
}
try:
ephemeral_key = bytes.fromhex(str(fields.get("ephemeralPublicKey") or ""))
signature_bytes = bytes.fromhex(str(announcement_signature or ""))
if len(signature_bytes) != 64:
return True
RNS.Cryptography.Ed25519PublicKey.from_public_bytes(ephemeral_key).verify(
signature_bytes, _canonical(signed_announcement)
)
except Exception:
return True
destination_hash = str(fields.get("destinationHash") or "").lower()
# The wallet-signed capability is the source of truth for this
# exact Land session. Presence leases are only a fast-path and may
# briefly lag after login, account switching, or bridge restart.
# The subsequent link handshake still verifies that Reticulum
# connected to this precise advertised destination.
resolved_peer = str(
self.resolve_peer(address, destination_hash) or destination_hash
).lower()
if not resolved_peer:
return True
if wire.get("e") is not True:
self._drop_remote(peer_key, "disabled")
return True
self.remote_capabilities[peer_key] = {
"address": address, "sessionId": session_id,
"fields": fields, "signature": signature, "hash": expected_hash,
"peerHash": resolved_peer, "roomId": str(wire.get("u") or ""),
"busy": wire.get("b") is True, "at": time.time(),
}
self._trace("discovery_accepted", address, throttle=10.0)
self._reconcile()
except Exception:
self.stats["invalidFrames"] += 1
return True
def _valid_remote_capability(self, fields: Dict[str, Any], signature: str, address: str) -> bool:
required = {
"type", "protocolVersion", "address", "signerPublicKey", "ephemeralPublicKey",
"groupId", "landSessionId", "destinationHash", "instanceId", "nonce", "createdAt", "expiresAt",
}
if set(fields.keys()) != required or fields.get("type") != "QORTAL_LAND_PROXIMITY_VOICE_SESSION":
return False
now_ms = int(time.time() * 1000)
created, expires = fields.get("createdAt"), fields.get("expiresAt")
if not isinstance(created, int) or not isinstance(expires, int):
return False
if created > now_ms + CAPABILITY_CLOCK_SKEW_MS or expires <= now_ms or expires - created > CAPABILITY_MAX_AGE_MS:
return False
public_key = str(fields.get("signerPublicKey") or "")
try:
bytes.fromhex(str(fields.get("nonce") or ""))
bytes.fromhex(str(fields.get("ephemeralPublicKey") or ""))
except ValueError:
return False
return (
fields.get("protocolVersion") == PROTOCOL_VERSION
and isinstance(fields.get("instanceId"), str)
and self._valid_instance_id(str(fields.get("instanceId")))
and isinstance(fields.get("landSessionId"), str)
and 0 < len(str(fields.get("landSessionId"))) <= 24
and isinstance(fields.get("destinationHash"), str)
and len(str(fields.get("destinationHash"))) == 32
and all(char in "0123456789abcdef" for char in str(fields.get("destinationHash")).lower())
and isinstance(fields.get("nonce"), str)
and len(str(fields.get("nonce"))) == 64
and isinstance(fields.get("ephemeralPublicKey"), str)
and len(str(fields.get("ephemeralPublicKey"))) == 64
and fields.get("address") == address
and self.derive_address(public_key) == address
and self.verify_wallet(fields, public_key, signature)
)
@staticmethod
def _valid_instance_id(value: str) -> bool:
try:
return len(value) == 36 and uuid.UUID(value).version == 4
except (ValueError, AttributeError):
return False
def on_land_state(self, wire: Dict[str, Any], peer_hash: str) -> None:
address = str(wire.get("a") or "")
session_id = str(wire.get("s") or "")
x, y = wire.get("x"), wire.get("y")
timestamp = wire.get("ts")
if (
not address or not session_id or not isinstance(x, (int, float)) or isinstance(x, bool)
or not isinstance(y, (int, float)) or isinstance(y, bool)
or not math.isfinite(float(x)) or not math.isfinite(float(y))
or not isinstance(timestamp, (int, float)) or isinstance(timestamp, bool)
or abs(time.time() * 1000 - float(timestamp)) > LAND_STATE_MAX_AGE * 1000
):
return
peer_key = self._peer_key(address, session_id)
self.remote_positions[peer_key] = {
"address": address, "groupId": str(wire.get("g") or ""), "sessionId": session_id,
"roomId": str(wire.get("u") or ""), "x": float(x),
"y": float(y), "peerHash": str(peer_hash or "").lower(),
"at": time.time(),
}
self._reconcile()
def _eligible(self) -> list[tuple[float, str]]:
if not self.enabled or self.suspended or not self.position or not self.context:
return []
now = time.time()
candidates: list[tuple[float, str]] = []
for peer_key, capability in self.remote_capabilities.items():
address = str(capability.get("address") or "")
if address in self.blocked_addresses:
continue
position = self.remote_positions.get(peer_key)
if not position or capability.get("busy"):
continue
if now - capability["at"] > DISCOVERY_MAX_AGE or now - position["at"] > LAND_STATE_MAX_AGE:
continue
if int(capability["fields"].get("expiresAt") or 0) <= int(now * 1000):
continue
if str(capability["fields"].get("landSessionId") or "") != str(position.get("sessionId") or ""):
continue
if position["roomId"] != self.position["roomId"]:
continue
if str(position["groupId"]) != str(self.context["groupId"]):
continue
distance = math.hypot(position["x"] - self.position["x"], position["y"] - self.position["y"])
if distance <= PRECONNECT_DISTANCE or (peer_key in self.links and distance <= RELEASE_DISTANCE):
candidates.append((distance, peer_key))
candidates.sort(key=lambda item: (item[0], item[1]))
by_peer = {peer_key: distance for distance, peer_key in candidates}
capacity = 5 if now < self.capacity_reduced_until else MAX_PEERS
selected = [peer_key for peer_key in self.links if peer_key in by_peer]
selected.sort(key=lambda peer_key: (by_peer[peer_key], peer_key))
selected = selected[:capacity]
for _distance, peer_key in candidates:
if len(selected) >= capacity:
break
if peer_key not in selected:
selected.append(peer_key)
if len(selected) >= capacity:
outsiders = [(distance, peer_key) for distance, peer_key in candidates if peer_key not in selected]
if outsiders:
newcomer_distance, newcomer = outsiders[0]
worst = max(selected, key=lambda peer_key: (by_peer[peer_key], peer_key))
worst_distance = by_peer[worst]
margin = max(60.0, worst_distance * 0.15)
if newcomer_distance <= worst_distance - margin:
since = self.replacement_since.setdefault(newcomer, now)
if now - since >= 3.0:
selected.remove(worst)
selected.append(newcomer)
self.replacement_since.pop(newcomer, None)
else:
self.replacement_since.pop(newcomer, None)
active_candidates = {peer_key for _distance, peer_key in candidates}
for peer_key in list(self.replacement_since):
if peer_key not in active_candidates:
self.replacement_since.pop(peer_key, None)
return sorted(((by_peer[peer_key], peer_key) for peer_key in selected), key=lambda item: (item[0], item[1]))
def _reconcile(self) -> None:
context = self.context
if not context:
return
selected = {peer_key: distance for distance, peer_key in self._eligible()}
now = time.time()
capacity = 5 if now < self.capacity_reduced_until else MAX_PEERS
for peer_key, state in list(self.links.items()):
distance = selected.get(peer_key)
if distance is None:
remote_position = self.remote_positions.get(peer_key)
if (
self.position and remote_position
and remote_position.get("roomId") != self.position.get("roomId")
):
state.setdefault("roomMismatchAt", now)
if now - state["roomMismatchAt"] >= 5.0:
self._close_peer(peer_key, "room_changed")
continue
state.pop("roomMismatchAt", None)
if len(selected) >= capacity and float(state.get("distance") or 9999) <= RELEASE_DISTANCE:
self._close_peer(peer_key, "capacity_rebalanced")
continue
state.setdefault("outsideAt", now)
if now - state["outsideAt"] >= 30 or float(state.get("distance") or 9999) > RELEASE_DISTANCE:
self._close_peer(peer_key, "out_of_range")
continue
state.pop("outsideAt", None)
state.pop("roomMismatchAt", None)
state["distance"] = distance
self._emit_peer(peer_key, state)
local_key = self._peer_key(str(context["address"]), str(context["landSessionId"]))
for peer_key, distance in selected.items():
if peer_key in self.links:
continue
if local_key < peer_key:
self._open_peer(peer_key, distance)
visible_capacity: set[str] = set()
if self.enabled and self.position:
now = time.time()
for peer_key, capability in self.remote_capabilities.items():
address = str(capability.get("address") or "")
if peer_key in selected or peer_key in self.links or address in self.blocked_addresses:
continue
distance = self._distance_to(peer_key)
if (
distance <= PRECONNECT_DISTANCE
and now - float(capability.get("at") or 0) <= DISCOVERY_MAX_AGE
and capability.get("busy") is not True
):
visible_capacity.add(peer_key)
self._emit_peer(peer_key, {
"address": address, "sessionId": capability.get("sessionId"),
"phase": "capacity", "sourceId": self._source_id(peer_key),
"authenticated": False, "muted": False, "volume": 1.0,
})
for peer_key in self.visible_capacity_peers - visible_capacity:
capability = self.remote_capabilities.get(peer_key) or {}
self.emit("PROXIMITY_PEER_STATE", {
"peerKey": peer_key, "address": capability.get("address", ""),
"sessionId": capability.get("sessionId", ""), "state": "disconnected",
"sourceId": self.source_ids.get(peer_key, 0), "reason": "not_nearby",
})
self.visible_capacity_peers = visible_capacity
def _open_peer(self, peer_key: str, distance: float) -> None:
retry = self.link_retry.get(peer_key) or {}
if time.time() < float(retry.get("nextAt") or 0):
return
capability = self.remote_capabilities.get(peer_key)
if not capability:
return
address = str(capability.get("address") or "")
session_id = str(capability.get("sessionId") or "")
advertised_hash = str(capability.get("fields", {}).get("destinationHash") or "").lower()
peer_hash = str(
self.resolve_peer(address, advertised_hash) or advertised_hash
).lower()
if not peer_hash or peer_hash != advertised_hash:
return
identity = self.resolve_identity(peer_hash) if peer_hash else None
if identity is None:
self._trace("candidate_waiting_identity", address, throttle=5.0)
return
try:
destination = self.build_destination(identity)
destination_hash = bytes(destination.hash)
if not self.path_available(destination_hash):
now = time.time()
if now - self.path_requested_at.get(peer_key, 0) >= 1.0:
self.path_requested_at[peer_key] = now
refreshed = False
if self.refresh_path is not None:
try:
refreshed = self.refresh_path(
peer_hash,
"proximity_link_no_path",
) is True
except Exception as exc:
self.log(
"[qortalland-proximity] path refresh failed "
f"peer={address[:8]} code={str(exc)[:80]}"
)
if not refreshed:
RNS.Transport.request_path(destination_hash)
self._trace("candidate_requesting_path", address, throttle=5.0)
return
state = {
"peerKey": peer_key, "address": address, "sessionId": session_id,
"peerHash": peer_hash, "distance": distance,
"phase": "opening", "createdAt": time.time(), "lastActivity": time.time(),
"outbound": True,
"authenticated": False, "muted": False, "volume": 1.0,
"sourceId": self._source_id(peer_key), "txSequence": 0,
"sendLock": threading.RLock(),
}
link = RNS.Link(destination, established_callback=self._outbound_established, closed_callback=self._link_closed)
state["link"] = link
with self.lock:
self.links[peer_key] = state
self.links_by_object[id(link)] = peer_key
self._trace("link_opening", address)
self._emit_peer(peer_key, state)
except Exception as exc:
self.stats["linkFailures"] += 1
self._schedule_retry(peer_key)
self.log(f"[qortalland-proximity] open failed peer={address[:8]} code={str(exc)[:80]}")
def _schedule_retry(self, peer_key: str) -> None:
previous = self.link_retry.get(peer_key) or {}
attempts = min(6, int(previous.get("attempts") or 0) + 1)
delay = min(5.0, 0.25 * (2 ** (attempts - 1)))
self.link_retry[peer_key] = {"attempts": attempts, "nextAt": time.time() + delay}
def _outbound_established(self, link) -> None:
state = self._state_for_link(link)
if not state or not self.capability or not self.ephemeral_private or not self.position:
return
try:
if self.identify_link is None:
raise RuntimeError("link_identity_unavailable")
self.identify_link(link)
except Exception as exc:
self.log(
f"[qortalland-proximity] link identify failed peer={state['address'][:8]} "
f"code={str(exc)[:80]}"
)
self._close_peer(state["peerKey"], "link_identity_failed")
return
link_id = bytes(self.link_id_bytes(link) or b"")
nonce = secrets.token_bytes(16)
hello = {
"v": PROTOCOL_VERSION, "f": self.context["address"], "t": state["address"],
"g": str(self.context["groupId"]), "s": self.context["landSessionId"],
"o": state["sessionId"],
"r": self.position["roomId"], "c": self.capability_hash, "l": link_id,
"n": nonce, "ts": int(time.time() * 1000),
}
hello["z"] = self.ephemeral_private.sign(umsgpack.packb(hello))
raw = LINK_MAGIC + umsgpack.packb(hello)
if len(raw) > 425:
self._close_peer(state["peerKey"], "classifier_oversized")
return
state["phase"] = "authenticating"
state["linkId"] = link_id
state["nonce"] = nonce
link.set_packet_callback(self._on_packet)
self._send_packet(state, raw)
self._trace("classifier_sent", state["address"])
self._emit_peer(state["peerKey"], state)
def handle_classifier(self, link, raw: bytes) -> bool:
if not isinstance(raw, (bytes, bytearray)) or not raw.startswith(LINK_MAGIC):
return False
try:
hello = umsgpack.unpackb(bytes(raw[len(LINK_MAGIC):]))
if not isinstance(hello, dict) or not self._verify_link_hello(link, hello):
self._trace("classifier_rejected", str(hello.get("f") or "") if isinstance(hello, dict) else "", "validation")
_safe_close(link)
return True
address = str(hello["f"])
session_id = str(hello["s"])
peer_key = self._peer_key(address, session_id)
capacity = 5 if time.time() < self.capacity_reduced_until else MAX_PEERS
if len(self.links) >= capacity and peer_key not in self.links:
self._send_control({"link": link}, {"c": "reject", "r": "capacity"})
_safe_close(link)
return True
existing = self.links.get(peer_key)
if existing:
_safe_close(existing.get("link"))
position = self.remote_positions.get(peer_key) or {}
remote = self.remote_capabilities.get(peer_key) or {}
state = {
"peerKey": peer_key, "address": address, "sessionId": session_id,
"peerHash": str(remote.get("fields", {}).get("destinationHash") or ""),
"distance": self._distance_to(peer_key), "phase": "connected", "link": link,
"outbound": False,
"linkId": bytes(self.link_id_bytes(link) or b""), "authenticated": True,
"createdAt": time.time(), "lastActivity": time.time(), "muted": False,
"volume": 1.0, "sourceId": self._source_id(peer_key), "txSequence": 0,
"remoteCapabilityHash": hello["c"], "nonce": hello["n"],
"sendLock": threading.RLock(),
}
with self.lock:
self.links[peer_key] = state
self.links_by_object[id(link)] = peer_key
self.link_retry.pop(peer_key, None)
link.set_packet_callback(self._on_packet)
link.set_link_closed_callback(self._link_closed)
accept = {
"v": PROTOCOL_VERSION, "a": self.context["address"],
"c": "accept", "f": self.context["address"], "t": address,
"s": self.context["landSessionId"], "o": session_id,
"h": self.capability_hash, "q": hello["c"], "l": state["linkId"],
"n": hello["n"], "r": secrets.token_bytes(16),
"ts": int(time.time() * 1000),
}
accept["z"] = self.ephemeral_private.sign(umsgpack.packb(accept))
state["authAccept"] = accept
state["lastAuthAccept"] = time.time()
state["authAcceptAttempts"] = 1
self._send_control(state, accept)
self._trace("accept_sent", address)
self._emit_peer(peer_key, state)
except Exception:
self.stats["invalidFrames"] += 1
_safe_close(link)
return True
def _verify_link_hello(self, link, hello: Dict[str, Any]) -> bool:
required = {"v", "f", "t", "g", "s", "o", "r", "c", "l", "n", "ts", "z"}
if (
set(hello.keys()) != required or hello.get("v") != PROTOCOL_VERSION
or not self.context or not self.enabled or self.suspended or not self.capability
):
return False
address = str(hello.get("f") or "")
session_id = str(hello.get("s") or "")
peer_key = self._peer_key(address, session_id)
local_key = self._peer_key(str(self.context["address"]), str(self.context["landSessionId"]))
remote = self.remote_capabilities.get(peer_key)
now = time.time()
# Exactly one side is allowed to initiate, so crossed/duplicate links
# converge without timing-dependent winner selection.
if (
not remote or now - float(remote.get("at") or 0) > DISCOVERY_MAX_AGE
or peer_key >= local_key
or hello.get("t") != self.context["address"]
or hello.get("o") != self.context["landSessionId"]
or str(hello.get("g")) != str(self.context["groupId"])
):
return False
destination_hash = str(remote.get("fields", {}).get("destinationHash") or "").lower()
resolved_peer = str(
self.resolve_peer(address, destination_hash) or destination_hash
).lower()
link_peer = str(self.resolve_link_peer_hash(link) or "").lower() if self.resolve_link_peer_hash else ""
if not resolved_peer or resolved_peer != destination_hash or link_peer != destination_hash:
return False
if hello.get("c") != remote["hash"] or hello.get("l") != bytes(self.link_id_bytes(link) or b""):
return False
position = self.remote_positions.get(peer_key) or {}
if (
not self.position
or now - float(self.position.get("at") or 0) > LAND_STATE_MAX_AGE
or now - float(position.get("at") or 0) > LAND_STATE_MAX_AGE
or (
hello.get("s") != remote["fields"].get("landSessionId")
or hello.get("s") != position.get("sessionId")
or hello.get("r") != self.position.get("roomId")
or hello.get("r") != position.get("roomId")
)
):
return False
nonce = hello.get("n")
if not isinstance(nonce, bytes) or len(nonce) != 16 or nonce in self.used_link_nonces:
return False
if abs(int(time.time() * 1000) - int(hello.get("ts") or 0)) > CAPABILITY_CLOCK_SKEW_MS:
return False
signed = dict(hello)
signature = signed.pop("z")
ephemeral_hex = str(remote["fields"].get("ephemeralPublicKey") or "")
try:
RNS.Cryptography.Ed25519PublicKey.from_public_bytes(bytes.fromhex(ephemeral_hex)).verify(signature, umsgpack.packb(signed))
except Exception:
return False
self.used_link_nonces[nonce] = now
return self._distance_to(peer_key) <= PRECONNECT_DISTANCE
def _send_packet(self, state: Dict[str, Any], raw: bytes) -> None:
link = state.get("link")
if link is None:
return
send_lock = state.get("sendLock")
if send_lock is None:
send_lock = threading.RLock()
state["sendLock"] = send_lock
with send_lock:
RNS.Packet(link, raw).send()
state["lastSend"] = time.time()
def _send_control(self, state: Dict[str, Any], payload: Dict[str, Any]) -> None:
link = state.get("link")
if link is None:
return
# Signed handshake controls already contain ``v`` and ``ts``. Preserve
# their insertion order because MessagePack signatures cover the exact
# encoded bytes, not just the map's key/value pairs.
if "v" in payload and "ts" in payload:
body = dict(payload)
else:
body = {"v": PROTOCOL_VERSION, "ts": int(time.time() * 1000), **payload}
raw = CONTROL_MAGIC + umsgpack.packb(body)
if len(raw) > 425:
raise ValueError("proximity_control_oversized")
self._send_packet(state, raw)
def _send_control_all(self, payload: Dict[str, Any]) -> None:
for state in list(self.links.values()):
if state.get("authenticated"):
try:
self._send_control(state, payload)
except Exception:
pass
def _on_packet(self, raw: bytes, packet) -> None:
link = getattr(packet, "link", None)
state = self._state_for_link(link)
if not state:
return
state["lastActivity"] = time.time()
if raw.startswith(CONTROL_MAGIC):
try:
control = umsgpack.unpackb(raw[len(CONTROL_MAGIC):])
except Exception:
return
command = control.get("c") if isinstance(control, dict) else None
if (
not isinstance(control, dict) or control.get("v") != PROTOCOL_VERSION
or abs(int(time.time() * 1000) - int(control.get("ts") or 0)) > CAPABILITY_CLOCK_SKEW_MS
):
self.stats["invalidFrames"] += 1
return
if command == "accept":
remote = self.remote_capabilities.get(state["peerKey"])
required = {"v", "a", "c", "f", "t", "s", "o", "h", "q", "l", "n", "r", "ts", "z"}
rejection = ""
if not remote:
rejection = "missing_remote_capability"
elif set(control.keys()) != required:
rejection = "schema"
elif control.get("h") != remote.get("hash"):
rejection = "remote_capability_hash"
elif control.get("q") != self.capability_hash:
rejection = "local_capability_hash"
elif control.get("f") != state["address"]:
rejection = "sender"
elif control.get("s") != state["sessionId"]:
rejection = "sender_session"
elif not self.context or control.get("t") != self.context["address"]:
rejection = "recipient"
elif not self.context or control.get("o") != self.context["landSessionId"]:
rejection = "recipient_session"
elif control.get("l") != state.get("linkId"):
rejection = "link_id"
elif control.get("n") != state.get("nonce"):
rejection = "nonce"
elif not isinstance(control.get("r"), bytes) or len(control["r"]) != 16:
rejection = "response_nonce"
elif abs(int(time.time() * 1000) - int(control.get("ts") or 0)) > CAPABILITY_CLOCK_SKEW_MS:
rejection = "timestamp"
if rejection:
self._trace("accept_rejected", state["address"], rejection, throttle=2.0)
return
signed = dict(control)
signature = signed.pop("z")
try:
remote_key = bytes.fromhex(str(remote["fields"].get("ephemeralPublicKey") or ""))
RNS.Cryptography.Ed25519PublicKey.from_public_bytes(remote_key).verify(
signature, umsgpack.packb(signed)
)
except Exception:
self._trace("accept_rejected", state["address"], "signature", throttle=2.0)
return
state["authenticated"] = True
state["phase"] = "connected"
self.link_retry.pop(state["peerKey"], None)
self._send_control(state, {
"c": "auth_ack", "l": state["linkId"], "n": state["nonce"],
})
self._trace("link_authenticated", state["address"], "outbound")
self._emit_peer(state["peerKey"], state)
elif command == "auth_ack":
if (
set(control.keys()) != {"v", "ts", "c", "l", "n"}
or control.get("l") != state.get("linkId")
or control.get("n") != state.get("nonce")
or not state.get("authAccept")
):
return
state.pop("authAccept", None)
state.pop("lastAuthAccept", None)
state.pop("authAcceptAttempts", None)
state["authenticated"] = True
state["phase"] = "connected"
self.link_retry.pop(state["peerKey"], None)
self._trace("link_authenticated", state["address"], "inbound")
self._emit_peer(state["peerKey"], state)
elif command == "ping":
if set(control.keys()) != {"v", "ts", "c"}:
return
self._send_control(state, {"c": "pong"})
elif command == "pong":
if set(control.keys()) != {"v", "ts", "c"}:
return
ping_at = float(state.get("pingAt") or 0)
if ping_at > 0:
state["rttMs"] = max(0, int((time.time() - ping_at) * 1000))
state["pingAt"] = 0.0
elif command == "talk":
if set(control.keys()) != {"v", "ts", "c", "a"} or not isinstance(control.get("a"), bool):
return
state["remoteSpeaking"] = control.get("a") is True
state["speakingUntil"] = time.time() + 0.35 if state["remoteSpeaking"] else 0.0
self.emit("PROXIMITY_SPEAKING_STATE", {
"peerKey": state["peerKey"], "address": state["address"],
"sessionId": state["sessionId"], "speaking": state["remoteSpeaking"],
})
elif command in {"pause", "resume"}:
if set(control.keys()) != {"v", "ts", "c"}:
return
state["remotePaused"] = command == "pause"
if command == "pause":
self.emit("PROXIMITY_SPEAKING_STATE", {
"peerKey": state["peerKey"], "address": state["address"],
"sessionId": state["sessionId"], "speaking": False,
})
self._emit_peer(state["peerKey"], state)
elif command == "close":
if not set(control.keys()).issubset({"v", "ts", "c", "r"}):
return
self._close_peer(state["peerKey"], "remote_closed")
elif command == "reject":
if set(control.keys()) != {"v", "ts", "c", "r"}:
return
self._close_peer(state["peerKey"], str(control.get("r") or "rejected")[:40])
return
if not raw.startswith(MEDIA_MAGIC) or not state.get("authenticated"):
return
try:
magic, version, flags, generation, sequence, capture_ms, length = RNS_AUDIO_HEADER.unpack_from(raw)
payload = raw[RNS_AUDIO_HEADER.size:]
if (
magic != MEDIA_MAGIC or version != PROTOCOL_VERSION or flags != 0
or length != len(payload) or not payload or length > MAX_OPUS_BYTES
):
raise ValueError("invalid media")
if state.get("muted") or state.get("remotePaused") or self.suspended or self._distance_to(state["peerKey"]) > AUDIBLE_EXIT_DISTANCE:
return
if state.get("rxGeneration") != generation:
retired = state.setdefault("rxRetiredGenerations", set())
if generation in retired:
self.stats["staleDrops"] += 1
return
previous_generation = state.get("rxGeneration")
if previous_generation is not None:
retired.add(previous_generation)
if len(retired) > 8:
retired.pop()
state["rxGeneration"] = generation
state["rxMaxSequence"] = sequence
state["rxSeenSequences"] = {sequence}
else:
seen = state.setdefault("rxSeenSequences", set())
maximum = int(state.get("rxMaxSequence") or 0)
if sequence in seen:
self.stats["duplicateDrops"] += 1
return
if maximum > sequence and maximum - sequence > 32:
self.stats["staleDrops"] += 1
return
seen.add(sequence)
if sequence > maximum:
if maximum >= 0 and sequence > maximum + 1:
self.stats["sequenceSkips"] += sequence - maximum - 1
maximum = sequence
state["rxMaxSequence"] = sequence
state["rxSeenSequences"] = {item for item in seen if maximum - item <= 32}
frame = LOCAL_AUDIO_HEADER.pack(
LOCAL_AUDIO_MAGIC, PROTOCOL_VERSION, 1, state["sourceId"], generation,
sequence, int(time.time() * 1000), len(payload),
) + payload
if self.send_binary(frame, state["sourceId"]):
self.stats["receivedFrames"] += 1
state["speakingUntil"] = time.time() + 0.35
if not state.get("remoteSpeaking"):
state["remoteSpeaking"] = True
self.emit("PROXIMITY_SPEAKING_STATE", {
"peerKey": state["peerKey"], "address": state["address"],
"sessionId": state["sessionId"], "speaking": True,
})
except Exception:
self.stats["invalidFrames"] += 1
def handle_local_audio(self, raw: bytes) -> bool:
if not isinstance(raw, bytes) or len(raw) > MAX_LOCAL_AUDIO_BYTES:
self.stats["invalidFrames"] += 1
return False
try:
magic, version, kind, source_id, generation, sequence, captured_at, length = LOCAL_AUDIO_HEADER.unpack_from(raw)
payload = raw[LOCAL_AUDIO_HEADER.size:]
if (
magic != LOCAL_AUDIO_MAGIC or version != PROTOCOL_VERSION or kind != 0 or source_id != 0
or generation != self.stream_generation
or length != len(payload) or not payload or length > MAX_OPUS_BYTES
):
raise ValueError("invalid local media")
now_ms = int(time.time() * 1000)
if captured_at > now_ms + 2_000:
self.stats["invalidFrames"] += 1
return False
if now_ms - captured_at > 200:
self.stats["staleDrops"] += 1
return True
if sequence <= self.local_media_max_sequence:
self.stats["staleDrops"] += 1
return True
self.local_media_max_sequence = sequence
if not self.enabled or self.suspended or not self.transmitting:
return True
self.stats["localFrames"] += 1
for state in list(self.links.values()):
if not state.get("authenticated") or state.get("remotePaused") or self._distance_to(state["peerKey"]) > AUDIBLE_EXIT_DISTANCE:
continue
frame = RNS_AUDIO_HEADER.pack(MEDIA_MAGIC, PROTOCOL_VERSION, 0, generation, sequence, captured_at & 0xFFFFFFFF, len(payload)) + payload
try:
self._send_packet(state, frame)
self.stats["sentFrames"] += 1
except Exception:
self.stats["linkFailures"] += 1
return True
except Exception:
self.stats["invalidFrames"] += 1
return False
def queue_local_audio(self, raw: bytes) -> bool:
if not isinstance(raw, bytes) or len(raw) < LOCAL_AUDIO_HEADER.size or len(raw) > MAX_LOCAL_AUDIO_BYTES:
return False
item = (time.time(), raw)
try:
self.local_audio_queue.put_nowait(item)
except queue.Full:
try:
self.local_audio_queue.get_nowait()
self.stats["queueDrops"] += 1
self.stats["localQueueDrops"] += 1
self.capacity_reduced_until = time.time() + 10.0
self.local_audio_queue.put_nowait(item)
except (queue.Empty, queue.Full):
self.stats["queueDrops"] += 1
self.stats["localQueueDrops"] += 1
return True
with self.lock:
if self.media_drain_scheduled:
return True
self.media_drain_scheduled = True
if not self.enqueue_media(self._drain_local_audio, ()):
with self.lock:
self.media_drain_scheduled = False
return False
return True
def _drain_local_audio(self) -> None:
started_at = time.monotonic()
drained = 0
while drained < MEDIA_DRAIN_MAX_FRAMES:
try:
queued_at, raw = self.local_audio_queue.get_nowait()
except queue.Empty:
break
if time.time() - queued_at <= 0.2:
self.handle_local_audio(raw)
else:
self.stats["staleDrops"] += 1
drained += 1
# Drain short bursts in one scheduler turn instead of repeatedly
# yielding after four frames. Retain a strict time budget so media
# cannot monopolise its RNS lane when a send becomes slow.
if drained >= 4 and time.monotonic() - started_at >= MEDIA_DRAIN_TIME_BUDGET:
break
with self.lock:
self.media_drain_scheduled = False
has_more = not self.local_audio_queue.empty()
if has_more:
with self.lock:
if self.media_drain_scheduled:
return
self.media_drain_scheduled = True
if not self.enqueue_media(self._drain_local_audio, ()):
with self.lock:
self.media_drain_scheduled = False
def tick(self) -> None:
if not self.context:
return
now = time.time()
if self.renderer_lost_at is not None and now - self.renderer_lost_at >= 30.0:
self.disable("renderer_recovery_expired")
return
if self.suspended_at is not None and now - self.suspended_at >= 30.0:
for peer_key in list(self.links):
self._close_peer(peer_key, "call_suspension_timeout")
if self.pending_fields and int(self.pending_fields.get("createdAt") or 0) + CAPABILITY_CLOCK_SKEW_MS <= int(now * 1000):
self.disable("authorization_expired")
return
if self.capability and int(self.capability.get("expiresAt") or 0) <= int(now * 1000):
self.disable("capability_expired")
return
if (
self.enabled and self.capability and not self.pending_fields
and int(self.capability.get("expiresAt") or 0) - int(now * 1000) <= 5 * 60 * 1000
):
self._enable({"mode": self.mode})
if self.enabled and now - self.last_discovery_at >= 10:
self._broadcast(True)
if self.enabled and now - self.last_stats_at >= 5:
self.last_stats_at = now
self.emit("PROXIMITY_TRANSPORT_STATS", {
"stats": dict(self.stats),
"bitrate": 16000 if now < self.capacity_reduced_until else 24000,
"capacityReduced": now < self.capacity_reduced_until,
})
if self.enabled and now - self.last_media_log_at >= 15 and (
self.stats["localFrames"] or self.stats["receivedFrames"]
):
self.last_media_log_at = now
self.log(
"[qortalland-proximity] stage=media_stats "
f"peers={sum(1 for state in self.links.values() if state.get('authenticated'))} "
f"local={self.stats['localFrames']} sent={self.stats['sentFrames']} "
f"received={self.stats['receivedFrames']} skips={self.stats['sequenceSkips']} "
f"duplicates={self.stats['duplicateDrops']} stale={self.stats['staleDrops']} "
f"queueDrops={self.stats['queueDrops']} localQueueDrops={self.stats['localQueueDrops']} "
f"rendererQueueDrops={self.stats['rendererQueueDrops']} invalid={self.stats['invalidFrames']} "
f"linkFailures={self.stats['linkFailures']}"
)
for nonce, used_at in list(self.used_link_nonces.items()):
if now - used_at > 300:
self.used_link_nonces.pop(nonce, None)
for peer_key, capability in list(self.remote_capabilities.items()):
if now - float(capability.get("at") or 0) > DISCOVERY_MAX_AGE:
self._drop_remote(peer_key, "discovery_expired")
for peer_key, state in list(self.links.items()):
if state.get("remoteSpeaking") and now >= float(state.get("speakingUntil") or 0):
state["remoteSpeaking"] = False
self.emit("PROXIMITY_SPEAKING_STATE", {
"peerKey": peer_key, "address": state["address"],
"sessionId": state["sessionId"], "speaking": False,
})
age = now - float(state.get("lastActivity") or state.get("createdAt") or now)
if state.get("authenticated") and age > LINK_DEAD_AFTER:
self._schedule_retry(peer_key)
self._refresh_failed_link_path(state, "proximity_link_timeout")
self._close_peer(peer_key, "heartbeat_timeout")
continue
if not state.get("authenticated") and now - float(state.get("createdAt") or now) > LINK_TIMEOUT:
self._schedule_retry(peer_key)
self._refresh_failed_link_path(state, "proximity_link_timeout")
self._close_peer(peer_key, "establishment_timeout")
continue
if (
state.get("authAccept")
and now - float(state.get("lastAuthAccept") or 0) >= 1.0
):
attempts = int(state.get("authAcceptAttempts") or 1)
if attempts >= 10:
state.pop("authAccept", None)
state.pop("lastAuthAccept", None)
state.pop("authAcceptAttempts", None)
else:
try:
state["lastAuthAccept"] = now
state["authAcceptAttempts"] = attempts + 1
self._send_control(state, state["authAccept"])
except Exception:
self.stats["linkFailures"] += 1
if state.get("authenticated") and now - float(state.get("lastSend") or 0) >= HEARTBEAT_INTERVAL:
try:
state["pingAt"] = now
self._send_control(state, {"c": "ping"})
except Exception:
pass
self._reconcile()
def owns_link(self, link) -> bool:
return id(link) in self.links_by_object
def _link_closed(self, link) -> None:
state = self._state_for_link(link)
if state:
peer_key = state["peerKey"]
with self.lock:
self.links_by_object.pop(id(link), None)
if self.links.get(peer_key, {}).get("link") is link:
self.links.pop(peer_key, None)
self._schedule_retry(peer_key)
self._refresh_failed_link_path(state, "proximity_link_attempt_closed")
self._trace("link_closed", state["address"], "callback", throttle=1.0)
self.emit("PROXIMITY_PEER_STATE", {
"peerKey": peer_key, "address": state["address"], "sessionId": state["sessionId"],
"state": "disconnected", "sourceId": state.get("sourceId", 0),
})
def _refresh_failed_link_path(self, state: Dict[str, Any], reason: str) -> bool:
"""Invalidate only failed outbound routes; inbound/intentional closes do not churn paths."""
if state.get("outbound") is not True or self.refresh_path is None:
return False
peer_hash = str(state.get("peerHash") or "").strip().lower()
if len(peer_hash) != 32 or any(char not in "0123456789abcdef" for char in peer_hash):
return False
try:
return self.refresh_path(peer_hash, reason) is True
except Exception as exc:
self.log(
"[qortalland-proximity] failed-link path refresh failed "
f"peer={str(state.get('address') or '')[:8]} code={str(exc)[:80]}"
)
return False
def _close_peer(self, peer_key: str, reason: str) -> None:
state = self.links.pop(peer_key, None)
if not state:
return
link = state.get("link")
self.links_by_object.pop(id(link), None)
try:
if state.get("authenticated"):
self._send_control(state, {"c": "close", "r": reason})
except Exception:
pass
_safe_close(link)
self.emit("PROXIMITY_PEER_STATE", {
"peerKey": peer_key, "address": state["address"], "sessionId": state["sessionId"],
"state": "disconnected", "reason": reason, "sourceId": state.get("sourceId", 0),
})
def _drop_remote(self, peer_key: str, reason: str) -> None:
self.remote_capabilities.pop(peer_key, None)
self.remote_positions.pop(peer_key, None)
self._close_peer(peer_key, reason)
self.source_ids.pop(peer_key, None)
self.path_requested_at.pop(peer_key, None)
self.link_retry.pop(peer_key, None)
self.replacement_since.pop(peer_key, None)
self.visible_capacity_peers.discard(peer_key)
def _state_for_link(self, link) -> Optional[Dict[str, Any]]:
peer_key = self.links_by_object.get(id(link))
return self.links.get(peer_key) if peer_key else None
def _distance_to(self, peer_key: str) -> float:
remote = self.remote_positions.get(peer_key)
if (
not remote or not self.position
or time.time() - float(remote.get("at") or 0) > LAND_STATE_MAX_AGE
or remote.get("roomId") != self.position.get("roomId")
):
return float("inf")
return math.hypot(float(remote["x"]) - self.position["x"], float(remote["y"]) - self.position["y"])
def _source_id(self, peer_key: str) -> int:
existing = self.source_ids.get(peer_key)
if existing:
return existing
source_id = self.next_source_id
self.next_source_id = 1 if source_id >= 65535 else source_id + 1
self.source_ids[peer_key] = source_id
return source_id
def _gain(self, distance: float) -> float:
if distance <= FULL_VOLUME_DISTANCE:
return 1.0
if distance >= AUDIBLE_DISTANCE:
return 0.0
t = (distance - FULL_VOLUME_DISTANCE) / (AUDIBLE_DISTANCE - FULL_VOLUME_DISTANCE)
return max(0.0, min(1.0, 1.0 - (t * t * (3.0 - 2.0 * t))))
def _emit_peer(self, peer_key: str, state: Dict[str, Any]) -> None:
capability = self.remote_capabilities.get(peer_key) or {}
address = str(state.get("address") or capability.get("address") or "")
session_id = str(state.get("sessionId") or capability.get("sessionId") or "")
distance = self._distance_to(peer_key)
remote = self.remote_positions.get(peer_key) or {}
relative_x = 0.0 if not self.position else float(remote.get("x") or self.position["x"]) - self.position["x"]
self.emit("PROXIMITY_PEER_STATE", {
"peerKey": peer_key, "address": address, "sessionId": session_id,
"state": state.get("phase", "nearby"),
"sourceId": state.get("sourceId", self._source_id(peer_key)),
"distance": distance if math.isfinite(distance) else None,
"gain": self._gain(distance), "pan": max(-0.65, min(0.65, relative_x / 300.0)),
"muted": state.get("muted") is True, "volume": state.get("volume", 1.0),
"rttMs": state.get("rttMs"),
"audible": (
distance <= AUDIBLE_EXIT_DISTANCE and state.get("authenticated") is True
and not self.suspended and not state.get("remotePaused")
),
})
def _emit_state(self) -> None:
state = "suspended" if self.suspended else "ready" if self.enabled else "off"
self.emit("PROXIMITY_STATE", {
"state": state, "mode": self.mode, "transmitting": self.transmitting,
"connected": sum(1 for item in self.links.values() if item.get("authenticated")),
"maxPeers": MAX_PEERS, "streamGeneration": self.stream_generation,
})
def _emit_snapshot(self, diagnostics: bool = False) -> None:
peers = []
for peer_key, state in self.links.items():
distance = self._distance_to(peer_key)
remote = self.remote_positions.get(peer_key) or {}
relative_x = 0.0 if not self.position else float(remote.get("x") or self.position["x"]) - self.position["x"]
peers.append({
"peerKey": peer_key, "address": state.get("address", ""),
"sessionId": state.get("sessionId", ""),
"state": state.get("phase"), "sourceId": state.get("sourceId"),
"distance": distance if math.isfinite(distance) else None, "gain": self._gain(distance),
"pan": max(-0.65, min(0.65, relative_x / 300.0)),
"muted": state.get("muted") is True, "volume": state.get("volume", 1.0),
"audible": (
distance <= AUDIBLE_EXIT_DISTANCE and state.get("authenticated") is True
and not self.suspended and not state.get("remotePaused")
),
})
payload: Dict[str, Any] = {
"enabled": self.enabled, "suspended": self.suspended, "mode": self.mode,
"transmitting": self.transmitting, "streamGeneration": self.stream_generation,
"peers": peers,
}
if diagnostics:
payload["stats"] = dict(self.stats)
self.emit("PROXIMITY_SNAPSHOT", payload)
Reference in New Issue View Git Blame Copy Permalink