sedona_editor/dasp.py

"""
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

_MSG_TYPE_NAMES = {
    DISCOVER: "DISCOVER", HELLO: "HELLO", CHALLENGE: "CHALLENGE",
    AUTHENTICATE: "AUTHENTICATE", WELCOME: "WELCOME", KEEPALIVE: "KEEPALIVE",
    DATAGRAM: "DATAGRAM", CLOSE: "CLOSE",
}


def _trace(direction, msg, addr=None):
    import app_state
    if not app_state.TRACE_DASP:
        return
    name = _MSG_TYPE_NAMES.get(msg.msg_type, f"?{msg.msg_type}")
    seq = f" seq={msg.seq_num}" if msg.seq_num >= 0 else ""
    ack = f" ack={msg.ack}" if msg.ack >= 0 else ""
    plen = len(msg.payload) if msg.payload else 0
    dest = f" {addr[0]}:{addr[1]}" if addr else ""
    print(f"[DASP] {direction}{dest} {name}{seq}{ack} payload={plen}")


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):
        _trace(">>>", msg, (session.host, session.port))
        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):
        _trace("<<<", 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()