Live Editor: add DASP/Sox device connection with full CRUD and COV
Port the DASP transport and Sox protocol stack from sedonaOntrol Java to Python, enabling live device editing alongside the existing offline SAX editor. Lazy-loads components on navigation, caches them, and subscribes to COV events for the active wiresheet. New files: dasp.py, sox.py, device_session.py, connect_dialog.py, live_hooks.py. Modified: app_state.py (live-mode state), main.py (Device menu, COV polling, lazy tree expand), properties.py (Sox write-through), sax_file.py (structural mutations route through live_hooks), schema.py (find_manifest helper). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
@@ -0,0 +1,952 @@
|
||||
"""
|
||||
DASP (Datagram Authenticated Session Protocol) transport layer.
|
||||
|
||||
Ported from sedonaOntrol/src/sedona/dasp/ Java sources.
|
||||
Provides reliable, authenticated, full-duplex UDP communication
|
||||
as the transport layer beneath the Sox application protocol.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import queue
|
||||
import random
|
||||
import socket
|
||||
import struct
|
||||
import threading
|
||||
import time
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# Constants (from DaspConst.java)
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
|
||||
VERSION_VAL = 0x0100
|
||||
IDEAL_MAX_VAL = 512
|
||||
ABS_MAX_VAL = 1024
|
||||
RECEIVE_MAX_VAL = 31
|
||||
RECEIVE_TIMEOUT_VAL = 30_000 # ms
|
||||
CONNECT_TIMEOUT_VAL = 10_000 # ms
|
||||
SEND_RETRY_VAL = 1000 # ms
|
||||
MAX_RETRY_BACKOFF = 8000 # ms (Ontrol)
|
||||
ACK_MORE_BYTE_LIMIT = 7 # (Ontrol)
|
||||
DIGEST_ALGORITHM = "SHA-1" # (Ontrol: also "SHA-256")
|
||||
|
||||
IDEAL_MAX_DEF = 512
|
||||
ABS_MAX_DEF = 512
|
||||
RECEIVE_MAX_DEF = 31
|
||||
RECEIVE_TIMEOUT_DEF = 30_000 # ms
|
||||
|
||||
# Message types
|
||||
DISCOVER = 0
|
||||
HELLO = 1
|
||||
CHALLENGE = 2
|
||||
AUTHENTICATE = 3
|
||||
WELCOME = 4
|
||||
KEEPALIVE = 5
|
||||
DATAGRAM = 6
|
||||
CLOSE = 7
|
||||
|
||||
# Header field value encodings (low 2 bits of field id)
|
||||
NIL = 0
|
||||
U2 = 1
|
||||
STR = 2
|
||||
BYTES = 3
|
||||
|
||||
# Header field ids
|
||||
HDR_VERSION = 0x05
|
||||
HDR_REMOTE_ID = 0x09
|
||||
HDR_DIGEST_ALGORITHM = 0x0E
|
||||
HDR_NONCE = 0x13
|
||||
HDR_USERNAME = 0x16
|
||||
HDR_DIGEST = 0x1B
|
||||
HDR_IDEAL_MAX = 0x1D
|
||||
HDR_ABS_MAX = 0x21
|
||||
HDR_ACK = 0x25
|
||||
HDR_ACK_MORE = 0x2B
|
||||
HDR_RECEIVE_MAX = 0x2D
|
||||
HDR_RECEIVE_TIMEOUT = 0x31
|
||||
HDR_ERROR_CODE = 0x35
|
||||
HDR_PLATFORM_ID = 0x3A
|
||||
HDR_MESSAGE_LEN = 0x81 # (Ontrol)
|
||||
|
||||
# Error codes
|
||||
ERR_INCOMPATIBLE_VERSION = 0xE1
|
||||
ERR_BUSY = 0xE2
|
||||
ERR_DIGEST_NOT_SUPPORTED = 0xE3
|
||||
ERR_NOT_AUTHENTICATED = 0xE4
|
||||
ERR_TIMEOUT = 0xE5
|
||||
|
||||
DEFAULT_PORT = 1876
|
||||
|
||||
|
||||
class DaspException(Exception):
|
||||
def __init__(self, msg, error_code=-1):
|
||||
super().__init__(msg)
|
||||
self.error_code = error_code
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# DaspMsg (from DaspMsg.java)
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
|
||||
class DaspMsg:
|
||||
"""Binary encoder/decoder for DASP messages."""
|
||||
|
||||
def __init__(self):
|
||||
self.session_id = -1
|
||||
self.msg_type = -1
|
||||
self.seq_num = -1
|
||||
# headers
|
||||
self.version = -1
|
||||
self.remote_id = -1
|
||||
self.digest_algorithm = None
|
||||
self.nonce = None
|
||||
self.username = None
|
||||
self.digest = None
|
||||
self.ideal_max = -1
|
||||
self.abs_max = -1
|
||||
self.ack = -1
|
||||
self.ack_more = None
|
||||
self.receive_max = -1
|
||||
self.receive_timeout = -1 # ms internally
|
||||
self.error_code = -1
|
||||
self.platform_id = None
|
||||
self.include_message_len = True # (Ontrol)
|
||||
# payload
|
||||
self.payload = b""
|
||||
|
||||
# ── Decode ──────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def decode(buf, length=None):
|
||||
if length is None:
|
||||
length = len(buf)
|
||||
msg = DaspMsg()
|
||||
msg._decode(buf, length)
|
||||
return msg
|
||||
|
||||
def _decode(self, buf, length):
|
||||
self.session_id = (buf[0] << 8) | buf[1]
|
||||
self.seq_num = (buf[2] << 8) | buf[3]
|
||||
self.msg_type = (buf[4] & 0xFF) >> 4
|
||||
num_fields = buf[4] & 0x0F
|
||||
|
||||
pos = 5
|
||||
for _ in range(num_fields):
|
||||
fid = buf[pos] & 0xFF
|
||||
pos += 1
|
||||
ftype = fid & 0x03
|
||||
|
||||
u2_val = 0
|
||||
str_val = None
|
||||
bytes_val = None
|
||||
|
||||
if ftype == NIL:
|
||||
pass
|
||||
elif ftype == U2:
|
||||
u2_val = (buf[pos] << 8) | buf[pos + 1]
|
||||
pos += 2
|
||||
elif ftype == STR:
|
||||
start = pos
|
||||
while buf[pos] != 0:
|
||||
pos += 1
|
||||
str_val = buf[start:pos].decode("utf-8")
|
||||
pos += 1 # skip null terminator
|
||||
elif ftype == BYTES:
|
||||
blen = buf[pos] & 0xFF
|
||||
pos += 1
|
||||
bytes_val = bytes(buf[pos:pos + blen])
|
||||
pos += blen
|
||||
|
||||
if fid == HDR_VERSION: self.version = u2_val
|
||||
elif fid == HDR_REMOTE_ID: self.remote_id = u2_val
|
||||
elif fid == HDR_DIGEST_ALGORITHM: self.digest_algorithm = str_val
|
||||
elif fid == HDR_NONCE: self.nonce = bytes_val
|
||||
elif fid == HDR_USERNAME: self.username = str_val
|
||||
elif fid == HDR_DIGEST: self.digest = bytes_val
|
||||
elif fid == HDR_IDEAL_MAX: self.ideal_max = u2_val
|
||||
elif fid == HDR_ABS_MAX: self.abs_max = u2_val
|
||||
elif fid == HDR_ACK: self.ack = u2_val
|
||||
elif fid == HDR_ACK_MORE: self.ack_more = bytes_val
|
||||
elif fid == HDR_RECEIVE_MAX: self.receive_max = u2_val
|
||||
elif fid == HDR_RECEIVE_TIMEOUT: self.receive_timeout = u2_val * 1000
|
||||
elif fid == HDR_ERROR_CODE: self.error_code = u2_val
|
||||
elif fid == HDR_PLATFORM_ID: self.platform_id = str_val
|
||||
elif fid == HDR_MESSAGE_LEN: pass # not used after decode
|
||||
|
||||
self.payload = bytes(buf[pos:length])
|
||||
|
||||
# ── Encode ──────────────────────────────────────────────────
|
||||
|
||||
def encode(self):
|
||||
buf = bytearray(ABS_MAX_VAL)
|
||||
length = self._encode_into(buf)
|
||||
return bytes(buf[:length])
|
||||
|
||||
def _encode_into(self, buf):
|
||||
struct.pack_into(">HH", buf, 0, self.session_id & 0xFFFF, self.seq_num & 0xFFFF)
|
||||
pos = 5 # skip byte 4, will backpatch
|
||||
|
||||
has_ideal_max = self.ideal_max > -1 and self.ideal_max != IDEAL_MAX_DEF
|
||||
has_abs_max = self.abs_max > -1 and self.abs_max != ABS_MAX_DEF
|
||||
has_receive_max = self.receive_max > -1 and self.receive_max != RECEIVE_MAX_DEF
|
||||
has_receive_timeout = self.receive_timeout > -1 and self.receive_timeout != RECEIVE_TIMEOUT_DEF
|
||||
|
||||
num = 0
|
||||
|
||||
# Ontrol: MESSAGE_LEN first (backpatched)
|
||||
msg_len_pos = -1
|
||||
if self.include_message_len:
|
||||
num += 1
|
||||
buf[pos] = HDR_MESSAGE_LEN
|
||||
pos += 1
|
||||
msg_len_pos = pos
|
||||
pos += 2 # placeholder for length
|
||||
|
||||
if self.version > -1:
|
||||
num += 1; buf[pos] = HDR_VERSION; pos += 1
|
||||
pos = _put_u2(buf, pos, self.version)
|
||||
if self.remote_id > -1:
|
||||
num += 1; buf[pos] = HDR_REMOTE_ID; pos += 1
|
||||
pos = _put_u2(buf, pos, self.remote_id)
|
||||
if self.digest_algorithm is not None:
|
||||
num += 1; buf[pos] = HDR_DIGEST_ALGORITHM; pos += 1
|
||||
pos = _put_str(buf, pos, self.digest_algorithm)
|
||||
if self.nonce is not None:
|
||||
num += 1; buf[pos] = HDR_NONCE; pos += 1
|
||||
pos = _put_bytes(buf, pos, self.nonce)
|
||||
if self.username is not None:
|
||||
num += 1; buf[pos] = HDR_USERNAME; pos += 1
|
||||
pos = _put_str(buf, pos, self.username)
|
||||
if self.digest is not None:
|
||||
num += 1; buf[pos] = HDR_DIGEST; pos += 1
|
||||
pos = _put_bytes(buf, pos, self.digest)
|
||||
if has_ideal_max:
|
||||
num += 1; buf[pos] = HDR_IDEAL_MAX; pos += 1
|
||||
pos = _put_u2(buf, pos, self.ideal_max)
|
||||
if has_abs_max:
|
||||
num += 1; buf[pos] = HDR_ABS_MAX; pos += 1
|
||||
pos = _put_u2(buf, pos, self.abs_max)
|
||||
if self.ack > -1:
|
||||
num += 1; buf[pos] = HDR_ACK; pos += 1
|
||||
pos = _put_u2(buf, pos, self.ack)
|
||||
if self.ack_more is not None:
|
||||
num += 1; buf[pos] = HDR_ACK_MORE; pos += 1
|
||||
pos = _put_bytes(buf, pos, self.ack_more)
|
||||
if has_receive_max:
|
||||
num += 1; buf[pos] = HDR_RECEIVE_MAX; pos += 1
|
||||
pos = _put_u2(buf, pos, self.receive_max)
|
||||
if has_receive_timeout:
|
||||
num += 1; buf[pos] = HDR_RECEIVE_TIMEOUT; pos += 1
|
||||
pos = _put_u2(buf, pos, int(self.receive_timeout / 1000))
|
||||
if self.error_code > -1:
|
||||
num += 1; buf[pos] = HDR_ERROR_CODE; pos += 1
|
||||
pos = _put_u2(buf, pos, self.error_code)
|
||||
if self.platform_id is not None:
|
||||
num += 1; buf[pos] = HDR_PLATFORM_ID; pos += 1
|
||||
pos = _put_str(buf, pos, self.platform_id)
|
||||
|
||||
# backpatch msg_type | num_fields
|
||||
buf[4] = ((self.msg_type & 0x0F) << 4) | (num & 0x0F)
|
||||
|
||||
# payload
|
||||
if self.payload:
|
||||
buf[pos:pos + len(self.payload)] = self.payload
|
||||
pos += len(self.payload)
|
||||
|
||||
# backpatch MESSAGE_LEN
|
||||
if msg_len_pos >= 0:
|
||||
_put_u2(buf, msg_len_pos, pos)
|
||||
|
||||
return pos
|
||||
|
||||
|
||||
def _put_u2(buf, pos, val):
|
||||
buf[pos] = (val >> 8) & 0xFF
|
||||
buf[pos + 1] = val & 0xFF
|
||||
return pos + 2
|
||||
|
||||
def _put_str(buf, pos, val):
|
||||
encoded = val.encode("ascii")
|
||||
buf[pos:pos + len(encoded)] = encoded
|
||||
pos += len(encoded)
|
||||
buf[pos] = 0
|
||||
return pos + 1
|
||||
|
||||
def _put_bytes(buf, pos, val):
|
||||
buf[pos] = len(val) & 0xFF
|
||||
pos += 1
|
||||
buf[pos:pos + len(val)] = val
|
||||
return pos + len(val)
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# ReceiveWindow (from ReceiveWindow.java)
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
|
||||
class ReceiveWindow:
|
||||
"""Manages incoming message sequence numbers and acknowledgements."""
|
||||
|
||||
MAX = RECEIVE_MAX_VAL # 31
|
||||
|
||||
def __init__(self, session):
|
||||
self._session = session
|
||||
self._lock = threading.Lock()
|
||||
self._seq_num = -1
|
||||
self._seq_more = -1
|
||||
self._acked = False
|
||||
|
||||
def unacked(self):
|
||||
with self._lock:
|
||||
return not self._acked
|
||||
|
||||
def init(self, num):
|
||||
with self._lock:
|
||||
if self._seq_num >= 0:
|
||||
raise RuntimeError("ReceiveWindow already initialized")
|
||||
self._seq_num = (num - 1) & 0xFFFF
|
||||
self._seq_more = 1
|
||||
self._acked = False
|
||||
|
||||
def set_ack_headers(self, msg):
|
||||
with self._lock:
|
||||
payload_len = len(msg.payload)
|
||||
# room for 5-byte fixed header + 3-byte ack header = 8
|
||||
if payload_len + 8 <= self._session.ideal_max:
|
||||
msg.ack = self._seq_num
|
||||
self._acked = True
|
||||
# room for ack_more: +3 bytes = 11 total overhead
|
||||
if self._seq_more > 0x01 and payload_len + 11 <= self._session.ideal_max:
|
||||
max_bytes = self._session.ideal_max - payload_len - 10
|
||||
msg.ack_more = _to_ack_more(self._seq_more, max_bytes)
|
||||
|
||||
def receive(self, num):
|
||||
with self._lock:
|
||||
if num < 0 or num > 0xFFFF:
|
||||
raise RuntimeError(f"Invalid seq num: {num}")
|
||||
if self._seq_num < 0:
|
||||
raise RuntimeError("ReceiveWindow not initialized")
|
||||
|
||||
start = self._seq_num
|
||||
end = start + self.MAX
|
||||
|
||||
if end <= 0xFFFF:
|
||||
if num <= start or num > end:
|
||||
self._acked = False
|
||||
return False
|
||||
else:
|
||||
end -= 0xFFFF
|
||||
if num <= start and num > end:
|
||||
self._acked = False
|
||||
return False
|
||||
|
||||
# next in sequence — slide window
|
||||
if num == (start + 1) & 0xFFFF:
|
||||
while True:
|
||||
self._seq_num = (self._seq_num + 1) & 0xFFFF
|
||||
self._seq_more = (self._seq_more >> 1) & 0x7FFFFFFF
|
||||
if (self._seq_more & 0x02) == 0:
|
||||
break
|
||||
self._seq_more |= 0x01
|
||||
else:
|
||||
# out of order
|
||||
bit = 1 << ((num - start) & 0xFFFF)
|
||||
if self._seq_more & bit:
|
||||
return False
|
||||
self._seq_more |= bit
|
||||
|
||||
self._acked = False
|
||||
return True
|
||||
|
||||
|
||||
def _to_ack_more(seq_more, max_bytes):
|
||||
if seq_more <= 0xFF or max_bytes <= 1:
|
||||
return bytes([seq_more & 0xFF])
|
||||
if seq_more <= 0xFFFF or max_bytes <= 2:
|
||||
return bytes([(seq_more >> 8) & 0xFF, seq_more & 0xFF])
|
||||
if seq_more <= 0xFFFFFF or max_bytes <= 3:
|
||||
return bytes([(seq_more >> 16) & 0xFF, (seq_more >> 8) & 0xFF, seq_more & 0xFF])
|
||||
return bytes([(seq_more >> 24) & 0xFF, (seq_more >> 16) & 0xFF,
|
||||
(seq_more >> 8) & 0xFF, seq_more & 0xFF])
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# SendWindow (from SendWindow.java)
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
|
||||
class _Packet:
|
||||
__slots__ = ("seq_num", "payload", "acked", "enqueued_time",
|
||||
"sent_time", "send_attempts", "next")
|
||||
|
||||
def __init__(self):
|
||||
self.seq_num = 0
|
||||
self.payload = b""
|
||||
self.acked = False
|
||||
self.enqueued_time = 0.0
|
||||
self.sent_time = 0.0
|
||||
self.send_attempts = 1
|
||||
self.next = None
|
||||
|
||||
|
||||
class SendWindow:
|
||||
"""Manages the outgoing datagram queue with reliable delivery."""
|
||||
|
||||
FAST_RETRANSMIT_DUP_ACKS = 2
|
||||
|
||||
def __init__(self, session):
|
||||
self._session = session
|
||||
self._cond = threading.Condition()
|
||||
self._head = None
|
||||
self._tail = None
|
||||
self._is_alive = True
|
||||
self._size = 0
|
||||
self._seq_num = random.randint(0, 0xFFFF)
|
||||
self.send_size = 8
|
||||
self.send_retry = SEND_RETRY_VAL / 1000.0 # seconds
|
||||
self.max_retry_backoff = MAX_RETRY_BACKOFF / 1000.0 # seconds
|
||||
self._last_checked_ack = -1
|
||||
self._dup_ack_count = 0
|
||||
|
||||
def size(self):
|
||||
with self._cond:
|
||||
return self._size
|
||||
|
||||
def cur_seq_num(self):
|
||||
return self._seq_num & 0xFFFF
|
||||
|
||||
def kill(self):
|
||||
with self._cond:
|
||||
self._is_alive = False
|
||||
self._cond.notify_all()
|
||||
|
||||
def clear(self):
|
||||
with self._cond:
|
||||
self._size = 0
|
||||
self._head = None
|
||||
self._tail = None
|
||||
self._cond.notify_all()
|
||||
|
||||
def send(self, payload):
|
||||
p = _Packet()
|
||||
p.enqueued_time = _ticks()
|
||||
with self._cond:
|
||||
p.seq_num = self._seq_num & 0xFFFF
|
||||
self._seq_num = (self._seq_num + 1) & 0xFFFF
|
||||
|
||||
self._enqueue(p)
|
||||
p.payload = payload
|
||||
p.sent_time = _ticks()
|
||||
self._session._send_raw(self._to_msg(p))
|
||||
|
||||
def _enqueue(self, p):
|
||||
with self._cond:
|
||||
while self._is_alive and (self._full() or self._exceeds_ack_more_limit(p)):
|
||||
self._cond.wait()
|
||||
if not self._is_alive:
|
||||
return
|
||||
if self._tail is None:
|
||||
self._head = self._tail = p
|
||||
else:
|
||||
self._tail.next = p
|
||||
self._tail = p
|
||||
self._size += 1
|
||||
self._cond.notify_all()
|
||||
|
||||
def _full(self):
|
||||
return self._size >= self.send_size
|
||||
|
||||
def _exceeds_ack_more_limit(self, p):
|
||||
last_ack = self._session.last_ack
|
||||
if last_ack < 0:
|
||||
return False
|
||||
span = (p.seq_num - last_ack) & 0xFFFF
|
||||
return span > self._session.ack_more_byte_limit
|
||||
|
||||
def _to_msg(self, packet):
|
||||
msg = DaspMsg()
|
||||
msg.msg_type = DATAGRAM
|
||||
msg.session_id = self._session.remote_id
|
||||
msg.seq_num = packet.seq_num
|
||||
msg.payload = packet.payload
|
||||
self._session.receive_window.set_ack_headers(msg)
|
||||
return msg
|
||||
|
||||
def check_ack_headers(self, msg):
|
||||
with self._cond:
|
||||
ack_num = msg.ack
|
||||
if ack_num < 0:
|
||||
return
|
||||
|
||||
if self._head is None:
|
||||
self._track_dup_acks(ack_num)
|
||||
return
|
||||
|
||||
unack_start = (ack_num + 1) & 0xFFFF
|
||||
unack_end = (unack_start + 10000) & 0xFFFF
|
||||
|
||||
p = self._head
|
||||
while p is not None:
|
||||
if not p.acked and self._is_acked(unack_start, unack_end, msg, p.seq_num):
|
||||
p.acked = True
|
||||
if p is self._head and p.acked:
|
||||
self._head = p.next
|
||||
self._size -= 1
|
||||
p = p.next
|
||||
|
||||
if self._head is None:
|
||||
self._tail = None
|
||||
|
||||
self._track_dup_acks(ack_num)
|
||||
self._cond.notify_all()
|
||||
|
||||
def _track_dup_acks(self, ack_num):
|
||||
if ack_num != self._last_checked_ack:
|
||||
self._last_checked_ack = ack_num
|
||||
self._dup_ack_count = 0
|
||||
return
|
||||
self._dup_ack_count += 1
|
||||
if self._dup_ack_count < self.FAST_RETRANSMIT_DUP_ACKS:
|
||||
return
|
||||
self._dup_ack_count = 0
|
||||
if self._head is not None and not self._head.acked:
|
||||
self._resend(self._head)
|
||||
|
||||
@staticmethod
|
||||
def _is_acked(unack_start, unack_end, msg, seq_num):
|
||||
if unack_start <= unack_end:
|
||||
if seq_num < unack_start or seq_num > unack_end:
|
||||
return True
|
||||
else:
|
||||
if seq_num < unack_start and seq_num > unack_end:
|
||||
return True
|
||||
|
||||
ack_more = msg.ack_more
|
||||
if ack_more is not None:
|
||||
diff = (seq_num - msg.ack) & 0xFFFF
|
||||
index = len(ack_more) - (diff >> 3) - 1
|
||||
if 0 <= index < len(ack_more):
|
||||
ack_more_byte = ack_more[index] & 0xFF
|
||||
mask = 1 << (diff & 0x07)
|
||||
if ack_more_byte & mask:
|
||||
return True
|
||||
return False
|
||||
|
||||
def send_retries(self):
|
||||
with self._cond:
|
||||
now = _ticks()
|
||||
oldest = 0.0
|
||||
p = self._head
|
||||
while p is not None:
|
||||
if not p.acked:
|
||||
backoff_shift = min(p.send_attempts - 1, 32)
|
||||
interval = min(self.send_retry * (1 << backoff_shift),
|
||||
self.max_retry_backoff)
|
||||
if now - p.sent_time >= interval:
|
||||
self._resend(p)
|
||||
oldest = max(oldest, now - p.enqueued_time)
|
||||
p = p.next
|
||||
return oldest
|
||||
|
||||
def _resend(self, p):
|
||||
p.sent_time = _ticks()
|
||||
p.send_attempts += 1
|
||||
self._session.num_retries += 1
|
||||
self._session._send_raw(self._to_msg(p))
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# DaspSession (from DaspSession.java)
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
|
||||
class DaspSession:
|
||||
"""
|
||||
Client endpoint of a DASP session. Handles the handshake,
|
||||
keepalive, reliable send/receive, and session close.
|
||||
"""
|
||||
|
||||
def __init__(self, dasp_socket, session_id, host, port, **options):
|
||||
self.socket = dasp_socket
|
||||
self.id = session_id
|
||||
self.host = host
|
||||
self.port = port
|
||||
self.remote_id = -1
|
||||
self.is_closed = False
|
||||
self.close_cause = "???"
|
||||
self.user = ""
|
||||
self.password = ""
|
||||
|
||||
self.ideal_max = options.get("ideal_max", IDEAL_MAX_DEF)
|
||||
self.abs_max = options.get("abs_max", ABS_MAX_DEF)
|
||||
self.receive_timeout = options.get("receive_timeout", RECEIVE_TIMEOUT_VAL) / 1000.0 # seconds
|
||||
self.connect_timeout = options.get("connect_timeout", CONNECT_TIMEOUT_VAL) / 1000.0 # seconds
|
||||
self.ack_more_byte_limit = options.get("ack_more_byte_limit", ACK_MORE_BYTE_LIMIT)
|
||||
self.digest_algorithm = options.get("digest_algorithm", DIGEST_ALGORITHM)
|
||||
self.include_message_len = options.get("include_message_len", True)
|
||||
|
||||
self.receive_window = ReceiveWindow(self)
|
||||
self.send_window = SendWindow(self)
|
||||
self.send_window.send_retry = options.get("send_retry", SEND_RETRY_VAL) / 1000.0
|
||||
self.send_window.max_retry_backoff = options.get("max_retry_backoff", MAX_RETRY_BACKOFF) / 1000.0
|
||||
|
||||
self._receive_queue = queue.Queue(maxsize=2000)
|
||||
self.remote_receive_max = RECEIVE_MAX_DEF
|
||||
self.last_ack = -1
|
||||
self.last_send = _ticks()
|
||||
self.last_receive = _ticks()
|
||||
self.num_sent = 0
|
||||
self.num_received = 0
|
||||
self.num_retries = 0
|
||||
self._nonce = None
|
||||
self._close_lock = threading.Lock()
|
||||
self.on_close = None # callback
|
||||
|
||||
# ── Client handshake ────────────────────────────────────────
|
||||
|
||||
def connect(self):
|
||||
res = self._hello()
|
||||
|
||||
if res.msg_type == CLOSE:
|
||||
raise DaspException("Connection denied", res.error_code)
|
||||
elif res.msg_type == CHALLENGE:
|
||||
self.remote_id = res.remote_id
|
||||
welcome = self._authenticate(res)
|
||||
elif res.msg_type == WELCOME:
|
||||
self.remote_id = res.remote_id
|
||||
welcome = res
|
||||
else:
|
||||
raise DaspException(f"Unexpected response to hello: {res.msg_type}")
|
||||
|
||||
if welcome.msg_type == CLOSE:
|
||||
raise DaspException("Connection denied", welcome.error_code)
|
||||
elif welcome.msg_type == WELCOME:
|
||||
self._tune(welcome)
|
||||
else:
|
||||
raise DaspException(f"Expected welcome, not {welcome.msg_type}")
|
||||
|
||||
def _hello(self):
|
||||
msg = DaspMsg()
|
||||
msg.msg_type = HELLO
|
||||
msg.session_id = 0xFFFF
|
||||
msg.seq_num = self.send_window.cur_seq_num()
|
||||
msg.remote_id = self.id
|
||||
msg.version = VERSION_VAL
|
||||
msg.ideal_max = self.ideal_max
|
||||
msg.abs_max = self.abs_max
|
||||
msg.receive_max = RECEIVE_MAX_VAL
|
||||
msg.receive_timeout = int(self.receive_timeout * 1000)
|
||||
|
||||
timeout = max(self.connect_timeout / 3, 1.0)
|
||||
for _ in range(3):
|
||||
self._send_raw(msg)
|
||||
try:
|
||||
res = self._receive_queue.get(timeout=timeout)
|
||||
return res
|
||||
except queue.Empty:
|
||||
continue
|
||||
raise DaspException("No response from hello")
|
||||
|
||||
def _authenticate(self, challenge):
|
||||
algorithm = challenge.digest_algorithm or "SHA-1"
|
||||
if algorithm not in ("SHA-1", "SHA-256"):
|
||||
self.close(ERR_DIGEST_NOT_SUPPORTED, "digest not supported")
|
||||
raise DaspException("Unsupported algorithm", ERR_DIGEST_NOT_SUPPORTED)
|
||||
|
||||
# credentials: always SHA-1 of "user:pass"
|
||||
cred = hashlib.sha1(f"{self.user}:{self.password}".encode("utf-8")).digest()
|
||||
|
||||
# outer digest uses negotiated algorithm
|
||||
algo_name = "sha1" if algorithm == "SHA-1" else "sha256"
|
||||
md = hashlib.new(algo_name)
|
||||
md.update(cred)
|
||||
md.update(challenge.nonce)
|
||||
digest_bytes = md.digest()
|
||||
|
||||
msg = DaspMsg()
|
||||
msg.msg_type = AUTHENTICATE
|
||||
msg.session_id = self.remote_id
|
||||
msg.seq_num = self.send_window.cur_seq_num()
|
||||
msg.username = self.user
|
||||
msg.digest = digest_bytes
|
||||
|
||||
timeout = max(self.connect_timeout / 3, 1.0)
|
||||
for _ in range(3):
|
||||
self._send_raw(msg)
|
||||
try:
|
||||
res = self._receive_queue.get(timeout=timeout)
|
||||
if res.msg_type != CHALLENGE:
|
||||
return res
|
||||
# drop duplicate challenge from multiple hellos
|
||||
except queue.Empty:
|
||||
continue
|
||||
raise DaspException("No response from authenticate")
|
||||
|
||||
def _tune(self, msg):
|
||||
remote_ideal = msg.ideal_max if msg.ideal_max > -1 else IDEAL_MAX_DEF
|
||||
remote_abs = msg.abs_max if msg.abs_max > -1 else ABS_MAX_DEF
|
||||
remote_recv_max = msg.receive_max if msg.receive_max > -1 else RECEIVE_MAX_DEF
|
||||
remote_timeout = (msg.receive_timeout if msg.receive_timeout > -1
|
||||
else RECEIVE_TIMEOUT_DEF) / 1000.0
|
||||
|
||||
self.ideal_max = min(remote_ideal, self.ideal_max)
|
||||
self.abs_max = min(remote_abs, self.abs_max)
|
||||
self.remote_receive_max = remote_recv_max
|
||||
self.receive_timeout = max(remote_timeout, self.receive_timeout)
|
||||
self.send_window.send_size = min(self.remote_receive_max, self.ack_more_byte_limit)
|
||||
self.receive_window.init(msg.seq_num)
|
||||
|
||||
# ── Send / Receive ──────────────────────────────────────────
|
||||
|
||||
def send(self, payload):
|
||||
if self.is_closed:
|
||||
raise DaspException(f"Session closed: {self.close_cause}")
|
||||
self.send_window.send(payload)
|
||||
|
||||
def receive(self, timeout=1.0):
|
||||
if self.is_closed:
|
||||
raise DaspException(f"Session closed: {self.close_cause}")
|
||||
try:
|
||||
msg = self._receive_queue.get(timeout=timeout)
|
||||
except queue.Empty:
|
||||
return None
|
||||
if self.is_closed:
|
||||
raise DaspException(f"Session closed: {self.close_cause}")
|
||||
if not isinstance(msg, DaspMsg):
|
||||
return msg
|
||||
if msg.msg_type != DATAGRAM:
|
||||
raise DaspException(f"Invalid message type received: {msg.msg_type}")
|
||||
return msg
|
||||
|
||||
def _send_raw(self, msg):
|
||||
msg.include_message_len = self.include_message_len
|
||||
self.last_send = _ticks()
|
||||
self.num_sent += 1
|
||||
self.socket._send_msg(self, msg)
|
||||
|
||||
# ── Dispatch (called on receiver thread) ────────────────────
|
||||
|
||||
def dispatch(self, msg):
|
||||
self.num_received += 1
|
||||
self.last_receive = _ticks()
|
||||
|
||||
if msg.ack >= 0:
|
||||
self.last_ack = msg.ack
|
||||
|
||||
self.send_window.check_ack_headers(msg)
|
||||
|
||||
if msg.msg_type == DATAGRAM:
|
||||
if not self.receive_window.receive(msg.seq_num):
|
||||
return
|
||||
self._enqueue(msg)
|
||||
elif msg.msg_type in (CHALLENGE, WELCOME):
|
||||
self._enqueue(msg)
|
||||
elif msg.msg_type == CLOSE:
|
||||
self._enqueue(msg)
|
||||
self._shutdown(f"remote close (error={msg.error_code})")
|
||||
elif msg.msg_type == KEEPALIVE:
|
||||
pass # lastReceive already updated
|
||||
|
||||
def _enqueue(self, msg):
|
||||
try:
|
||||
self._receive_queue.put_nowait(msg)
|
||||
except queue.Full:
|
||||
self.close(ERR_TIMEOUT, "receive queue full")
|
||||
|
||||
# ── Housekeeping (called every 100ms) ───────────────────────
|
||||
|
||||
def housekeeping(self):
|
||||
if self.is_closed:
|
||||
return
|
||||
|
||||
if _ticks() - self.last_receive > self.receive_timeout:
|
||||
self.close(ERR_TIMEOUT, "receive timeout")
|
||||
return
|
||||
|
||||
oldest = self.send_window.send_retries()
|
||||
if oldest > self.receive_timeout:
|
||||
self.close(ERR_TIMEOUT, "unacked send timeout")
|
||||
return
|
||||
|
||||
if (self.receive_window.unacked() or
|
||||
_ticks() - self.last_send > self.receive_timeout / 3):
|
||||
self._keepalive()
|
||||
|
||||
def _keepalive(self):
|
||||
msg = DaspMsg()
|
||||
msg.msg_type = KEEPALIVE
|
||||
msg.session_id = self.remote_id
|
||||
msg.seq_num = 0xFFFF
|
||||
self.receive_window.set_ack_headers(msg)
|
||||
self._send_raw(msg)
|
||||
|
||||
# ── Close ───────────────────────────────────────────────────
|
||||
|
||||
def close(self, error_code=-1, cause="close()"):
|
||||
with self._close_lock:
|
||||
if self.is_closed:
|
||||
return
|
||||
close_msg = DaspMsg()
|
||||
close_msg.msg_type = CLOSE
|
||||
close_msg.session_id = self.remote_id if self.remote_id >= 0 else 0xFFFF
|
||||
close_msg.seq_num = 0xFFFF
|
||||
if error_code >= 0:
|
||||
close_msg.error_code = error_code
|
||||
self._send_raw(close_msg)
|
||||
if error_code < 0:
|
||||
time.sleep(0.05)
|
||||
self._send_raw(close_msg)
|
||||
self._shutdown(cause)
|
||||
|
||||
def _shutdown(self, cause):
|
||||
with self._close_lock:
|
||||
if self.is_closed:
|
||||
return
|
||||
self.is_closed = True
|
||||
self.close_cause = cause
|
||||
self.send_window.kill()
|
||||
# wake anyone blocked on receive
|
||||
try:
|
||||
self._receive_queue.put_nowait(None)
|
||||
except queue.Full:
|
||||
pass
|
||||
if self.on_close:
|
||||
try:
|
||||
self.on_close(self)
|
||||
except Exception:
|
||||
pass
|
||||
self.socket._free_session(self)
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# DaspSocket (from DaspSocket.java + DaspSocketInterface.java)
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
|
||||
class DaspSocket:
|
||||
"""
|
||||
Manages a UDP socket shared by DASP sessions.
|
||||
Provides receiver and housekeeping background threads.
|
||||
"""
|
||||
|
||||
def __init__(self, port=-1):
|
||||
self._sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
self._sock.settimeout(1.0)
|
||||
if port > 0:
|
||||
self._sock.bind(("", port))
|
||||
else:
|
||||
self._sock.bind(("", 0))
|
||||
|
||||
self._sessions = {} # id -> DaspSession
|
||||
self._sessions_lock = threading.Lock()
|
||||
self._is_alive = True
|
||||
self._rand = random.Random()
|
||||
self._send_lock = threading.Lock()
|
||||
|
||||
self._receiver_thread = threading.Thread(
|
||||
target=self._receiver_loop, name="DASP-Receiver", daemon=True)
|
||||
self._housekeeping_thread = threading.Thread(
|
||||
target=self._housekeeping_loop, name="DASP-Housekeeping", daemon=True)
|
||||
self._receiver_thread.start()
|
||||
self._housekeeping_thread.start()
|
||||
|
||||
@property
|
||||
def port(self):
|
||||
return self._sock.getsockname()[1]
|
||||
|
||||
@property
|
||||
def is_closed(self):
|
||||
return not self._is_alive
|
||||
|
||||
def connect(self, host, port, user, password, **options):
|
||||
if not self._is_alive:
|
||||
raise DaspException("Socket is closed")
|
||||
session = self._alloc_session(host, port, **options)
|
||||
session.user = user
|
||||
session.password = password
|
||||
session.connect()
|
||||
return session
|
||||
|
||||
def close(self):
|
||||
if not self._is_alive:
|
||||
return
|
||||
self._is_alive = False
|
||||
with self._sessions_lock:
|
||||
for s in list(self._sessions.values()):
|
||||
try:
|
||||
s.close(-1, "socket closed")
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
self._sock.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── Session management ──────────────────────────────────────
|
||||
|
||||
def _alloc_session(self, host, port, **options):
|
||||
with self._sessions_lock:
|
||||
while True:
|
||||
sid = self._rand.randint(0, 0xFFFE)
|
||||
if sid not in self._sessions:
|
||||
break
|
||||
session = DaspSession(self, sid, host, port, **options)
|
||||
self._sessions[sid] = session
|
||||
return session
|
||||
|
||||
def _free_session(self, session):
|
||||
with self._sessions_lock:
|
||||
self._sessions.pop(session.id, None)
|
||||
|
||||
# ── Network I/O ─────────────────────────────────────────────
|
||||
|
||||
def _send_msg(self, session, msg):
|
||||
data = msg.encode()
|
||||
with self._send_lock:
|
||||
try:
|
||||
self._sock.sendto(data, (session.host, session.port))
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def _receiver_loop(self):
|
||||
buf = bytearray(ABS_MAX_VAL + 64)
|
||||
while self._is_alive:
|
||||
try:
|
||||
nbytes, addr = self._sock.recvfrom_into(buf)
|
||||
if nbytes < 5:
|
||||
continue
|
||||
msg = DaspMsg.decode(buf, nbytes)
|
||||
self._dispatch(msg, addr)
|
||||
except socket.timeout:
|
||||
continue
|
||||
except OSError:
|
||||
if self._is_alive:
|
||||
continue
|
||||
break
|
||||
|
||||
def _dispatch(self, msg, addr):
|
||||
session_id = msg.session_id
|
||||
with self._sessions_lock:
|
||||
session = self._sessions.get(session_id)
|
||||
if session is None:
|
||||
return
|
||||
if session.host != addr[0] or session.port != addr[1]:
|
||||
return
|
||||
session.dispatch(msg)
|
||||
|
||||
def _housekeeping_loop(self):
|
||||
while self._is_alive:
|
||||
time.sleep(0.1)
|
||||
with self._sessions_lock:
|
||||
sessions = list(self._sessions.values())
|
||||
for s in sessions:
|
||||
try:
|
||||
s.housekeeping()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
# Utility
|
||||
# ────────────────────────────────────────────────────────────────
|
||||
|
||||
def _ticks():
|
||||
"""Monotonic time in seconds (float)."""
|
||||
return time.monotonic()
|
||||
Reference in New Issue
Block a user