""" DVeProto — эталон (standalone site). Package 1.5 LTS (binary wire = 1.4 / ver=0x14). v1.0: JSON text frames v1.1: binary frames + FILE_* v1.2: stream_id mux + PING/PONG/CLOSE; NET/VPN reserved for 2.0 v1.3: AAD + counter nonces + WINDOW_UPDATE v1.4: pkt_seq + Session Resume + Priority/Cancel + DATA_ACK + RTT + caps v1.5 LTS: same binary as 1.4 — Wire Freeze, Ticket Store, hardening, Production API. Next major: 2.0 (DVeNet / DVeVPN / UDP+TCP). DVeProto 1.5 LTS is the final release of the first-generation transport protocol. Future network-layer features are developed in DVeProto 2.x. """ from __future__ import annotations import base64 import hashlib import json import os import struct import time from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple, Union, Iterable from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric.x25519 import ( X25519PrivateKey, X25519PublicKey, ) from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF PROTO_NAME = "DVeProto" PACKAGE_VERSION = "1.5" # Версия рукопожатия / HKDF (не менять: совместимость с клиентами v1). PROTO_HANDSHAKE_VERSION = 1 WIRE_V10 = "1.0" WIRE_V11 = "1.1" WIRE_V12 = "1.2" WIRE_V13 = "1.3" WIRE_V14 = "1.4" WIRE_V15 = "1.5" # LTS alias → same binary as 1.4 SUPPORTED_WIRES = (WIRE_V10, WIRE_V11, WIRE_V12, WIRE_V13, WIRE_V14, WIRE_V15) DEFAULT_OFFER: Tuple[str, ...] = ( WIRE_V10, WIRE_V11, WIRE_V12, WIRE_V13, WIRE_V14, WIRE_V15, ) WIRE_RANK = { WIRE_V10: 10, WIRE_V11: 11, WIRE_V12: 12, WIRE_V13: 13, WIRE_V14: 14, WIRE_V15: 14, # same binary rank as 1.4 } INFO_C2S = b"DVeProto-v1/c2s" INFO_S2C = b"DVeProto-v1/s2c" INFO_RESUME_C2S = b"DVeProto-v1/resume-c2s" INFO_RESUME_S2C = b"DVeProto-v1/resume-s2c" # --- Binary wire (FROZEN at 1.5 LTS — changes only in 2.x) ------------------- BIN_VERSION_11 = 0x11 BIN_VERSION_12 = 0x12 BIN_VERSION_13 = 0x13 BIN_VERSION_14 = 0x14 # also used for select 1.5 BIN_VERSION = BIN_VERSION_11 HEADER_LEN_11 = 14 HEADER_LEN_12 = 16 HEADER_LEN_13 = 16 HEADER_LEN_14 = 20 HEADER_LEN = HEADER_LEN_11 GCM_TAG_LEN = 16 NONCE_LEN = 12 STREAM_DEFAULT = 0 DEFAULT_STREAM_WINDOW = 256 * 1024 # --- 1.5 LTS security limits ------------------------------------------------- MAX_PLAINTEXT_BYTES = 1024 * 1024 # 1 MiB MAX_FRAME_BYTES = HEADER_LEN_14 + MAX_PLAINTEXT_BYTES + GCM_TAG_LEN SESSION_IDLE_TIMEOUT_S = 300 # 5 min without activity SESSION_MAX_LIFETIME_S = 86400 # hard cap; re-handshake / resume REPLAY_WINDOW = 64 # pkt_seq anti-replay window per stream DEFAULT_TICKET_TTL_S = 86400 MIN_TICKET_TTL_S = 60 MAX_TICKET_TTL_S = 7 * 86400 REKEY_AFTER_BYTES = 512 * 1024 * 1024 # suggest rekey after 512 MiB PTYPE_APP_JSON = 0x01 PTYPE_APP_BIN = 0x02 PTYPE_TEXT = 0x03 PTYPE_FILE_BEGIN = 0x10 PTYPE_FILE_CHUNK = 0x11 PTYPE_FILE_END = 0x12 PTYPE_FILE_ABORT = 0x13 PTYPE_FILE_ACK = 0x14 PTYPE_FILE_RESUME = 0x15 # Reserved for DVeNet / DVeVPN (protocol 2.0) — wire 1.2 may carry them. PTYPE_NET_CTRL = 0x20 PTYPE_NET_PKT = 0x21 PTYPE_VPN_PKT = 0x22 # Session control (wire 1.2+) PTYPE_PING = 0x30 PTYPE_PONG = 0x31 PTYPE_CLOSE = 0x32 PTYPE_WINDOW_UPDATE = 0x33 # wire 1.3+ flow control # Wire 1.4+ PTYPE_STREAM_PRIORITY = 0x34 PTYPE_STREAM_CANCEL = 0x35 PTYPE_DATA_ACK = 0x36 PTYPE_SESSION_TICKET = 0x37 PTYPE_CAPS = 0x38 # Capability bitmask (hello.caps / CAPS ptype) CAP_RESUME = 1 << 0 CAP_PRIORITY = 1 << 1 CAP_DATA_ACK = 1 << 2 CAP_CANCEL = 1 << 3 CAP_RTT = 1 << 4 CAP_PKT_SEQ = 1 << 5 CAP_COMPRESS = 1 << 6 CAP_ZSTD = 1 << 7 CAP_UDP = 1 << 8 # 2.0 carrier advertise CAP_TCP = 1 << 9 # 2.0 carrier advertise CAP_MULTIPATH = 1 << 10 CAP_WEBSOCKET = 1 << 11 DEFAULT_CAPS_14 = ( CAP_RESUME | CAP_PRIORITY | CAP_DATA_ACK | CAP_CANCEL | CAP_RTT | CAP_PKT_SEQ | CAP_WEBSOCKET ) PRIORITY_CONTROL = 0 PRIORITY_CHAT = 1 PRIORITY_VOICE = 2 PRIORITY_FILES = 3 FILE_FLAG_SHA256 = 0x01 FILE_FLAG_PHOTO = 0x02 # chat photo → CHAT_PHOTO_DIR / chat_photo.php DEFAULT_CHUNK_SIZE = 256 * 1024 # 256 KiB # FILE_ACK status ACK_OK = 0 ACK_NEED_RESUME = 1 ACK_COMPLETE = 2 ACK_ERROR = 3 def _b64e(raw: bytes) -> str: return base64.b64encode(raw).decode("ascii") def _b64d(s: str) -> bytes: return base64.b64decode(s.encode("ascii"), validate=True) def derive_aes_keys(shared_secret: bytes) -> Tuple[bytes, bytes]: """Ключ для client→server и server→client (симметрично на обеих сторонах).""" hkdf_c2s = HKDF( algorithm=hashes.SHA256(), length=32, salt=b"", info=INFO_C2S, ) hkdf_s2c = HKDF( algorithm=hashes.SHA256(), length=32, salt=b"", info=INFO_S2C, ) return hkdf_c2s.derive(shared_secret), hkdf_s2c.derive(shared_secret) def derive_resume_keys(token: bytes) -> Tuple[bytes, bytes]: """Ключи сессии из resume token (wire 1.4 Session Resume).""" if len(token) != 32: raise ValueError("resume token must be 32 bytes") hkdf_c2s = HKDF( algorithm=hashes.SHA256(), length=32, salt=b"", info=INFO_RESUME_C2S ) hkdf_s2c = HKDF( algorithm=hashes.SHA256(), length=32, salt=b"", info=INFO_RESUME_S2C ) return hkdf_c2s.derive(token), hkdf_s2c.derive(token) def normalize_wire(value: Any) -> Optional[str]: if value is None: return None s = str(value).strip() if s in ("1", "1.0", "v1", "v1.0", "json"): return WIRE_V10 if s in ("1.1", "v1.1", "bin", "bin11", "binary"): return WIRE_V11 if s in ("1.2", "v1.2", "bin12", "mux"): return WIRE_V12 if s in ("1.3", "v1.3", "bin13", "aad"): return WIRE_V13 if s in ("1.4", "v1.4", "bin14", "resume", "final"): return WIRE_V14 if s in ("1.5", "v1.5", "lts"): # 1.5 LTS is the same binary wire as 1.4 (ver=0x14). return WIRE_V14 return None def _best_wire(offered: set) -> str: if not offered: return WIRE_V10 return max(offered, key=lambda w: WIRE_RANK.get(w, 0)) def pick_wire(offer: Iterable[str], select: Optional[str]) -> str: """ Выбор общего режима. - select ∈ offer → select - иначе высший offered с rank ≤ preferred (или высший offered) - select is None / пустой → вызывающий код для legacy обычно форсит 1.0 """ offered: set = set() for item in offer: w = normalize_wire(item) if w: offered.add(w) if not offered: offered.add(WIRE_V10) chosen = normalize_wire(select) if select is not None else None if chosen and chosen in offered: return chosen if chosen: pref = WIRE_RANK.get(chosen, 0) candidates = [w for w in offered if WIRE_RANK.get(w, 0) <= pref] if candidates: return _best_wire(set(candidates)) return _best_wire(offered) @dataclass(frozen=True) class ClientAck: client_pk: Optional[bytes] select: str # "1.0" … "1.4" resume_session_id: Optional[bytes] = None resume_token: Optional[bytes] = None caps: int = 0 @property def is_resume(self) -> bool: return self.resume_session_id is not None and self.resume_token is not None @dataclass(frozen=True) class DecodedFrame: """Расшифрованный бинарный кадр 1.1+.""" ptype: int plaintext: bytes stream_id: int = STREAM_DEFAULT bin_ver: int = BIN_VERSION_11 pkt_seq: int = 0 @property def is_app_json(self) -> bool: return self.ptype == PTYPE_APP_JSON @property def is_control(self) -> bool: return self.ptype in ( PTYPE_PING, PTYPE_PONG, PTYPE_CLOSE, PTYPE_WINDOW_UPDATE, PTYPE_STREAM_PRIORITY, PTYPE_STREAM_CANCEL, PTYPE_DATA_ACK, PTYPE_SESSION_TICKET, PTYPE_CAPS, ) def as_json(self) -> Dict[str, Any]: if self.ptype != PTYPE_APP_JSON: raise ValueError("frame is not APP_JSON") return json.loads(self.plaintext.decode("utf-8")) def as_text(self) -> str: if self.ptype != PTYPE_TEXT: raise ValueError("frame is not TEXT") return self.plaintext.decode("utf-8") # --- File transfer plaintext codecs (inside AES-GCM) ------------------------- def pack_file_begin( transfer_id: bytes, *, file_size: int, chunk_size: int, total_chunks: int, name: str, mime: str = "application/octet-stream", sha256: Optional[bytes] = None, flags: int = 0, ) -> bytes: if len(transfer_id) != 16: raise ValueError("transfer_id must be 16 bytes") name_b = name.encode("utf-8") mime_b = mime.encode("utf-8") if len(name_b) > 0xFFFF or len(mime_b) > 0xFFFF: raise ValueError("name/mime too long") fl = int(flags) & 0xFF if sha256 is not None: fl |= FILE_FLAG_SHA256 parts = [ transfer_id, struct.pack(">QII", int(file_size), int(chunk_size), int(total_chunks)), bytes([fl]), struct.pack(">H", len(name_b)), name_b, struct.pack(">H", len(mime_b)), mime_b, ] if sha256 is not None: if len(sha256) != 32: raise ValueError("sha256 must be 32 bytes") parts.append(sha256) return b"".join(parts) def unpack_file_begin(pt: bytes) -> Dict[str, Any]: if len(pt) < 16 + 8 + 4 + 4 + 1 + 2: raise ValueError("FILE_BEGIN too short") o = 0 transfer_id = pt[o : o + 16] o += 16 file_size, chunk_size, total_chunks = struct.unpack_from(">QII", pt, o) o += 16 flags = pt[o] o += 1 (name_len,) = struct.unpack_from(">H", pt, o) o += 2 name = pt[o : o + name_len].decode("utf-8") o += name_len (mime_len,) = struct.unpack_from(">H", pt, o) o += 2 mime = pt[o : o + mime_len].decode("utf-8") o += mime_len sha256 = None if flags & FILE_FLAG_SHA256: if len(pt) < o + 32: raise ValueError("FILE_BEGIN missing sha256") sha256 = pt[o : o + 32] o += 32 return { "transfer_id": transfer_id, "file_size": file_size, "chunk_size": chunk_size, "total_chunks": total_chunks, "flags": flags, "name": name, "mime": mime, "sha256": sha256, } def pack_file_chunk(transfer_id: bytes, chunk_index: int, data: bytes) -> bytes: if len(transfer_id) != 16: raise ValueError("transfer_id must be 16 bytes") return transfer_id + struct.pack(">I", int(chunk_index)) + data def unpack_file_chunk(pt: bytes) -> Dict[str, Any]: if len(pt) < 20: raise ValueError("FILE_CHUNK too short") return { "transfer_id": pt[:16], "chunk_index": struct.unpack_from(">I", pt, 16)[0], "data": pt[20:], } def pack_file_end(transfer_id: bytes, sha256: bytes) -> bytes: if len(transfer_id) != 16 or len(sha256) != 32: raise ValueError("FILE_END fields invalid") return transfer_id + sha256 def unpack_file_end(pt: bytes) -> Dict[str, Any]: if len(pt) != 48: raise ValueError("FILE_END must be 48 bytes") return {"transfer_id": pt[:16], "sha256": pt[16:48]} def pack_file_abort(transfer_id: bytes, reason_code: int = 0) -> bytes: if len(transfer_id) != 16: raise ValueError("transfer_id must be 16 bytes") return transfer_id + bytes([reason_code & 0xFF]) def unpack_file_abort(pt: bytes) -> Dict[str, Any]: if len(pt) < 17: raise ValueError("FILE_ABORT too short") return {"transfer_id": pt[:16], "reason_code": pt[16]} def pack_file_ack(transfer_id: bytes, status: int, last_ok: int) -> bytes: if len(transfer_id) != 16: raise ValueError("transfer_id must be 16 bytes") return transfer_id + bytes([status & 0xFF]) + struct.pack(">I", int(last_ok) & 0xFFFFFFFF) def unpack_file_ack(pt: bytes) -> Dict[str, Any]: if len(pt) < 21: raise ValueError("FILE_ACK too short") return { "transfer_id": pt[:16], "status": pt[16], "last_ok": struct.unpack_from(">I", pt, 17)[0], } def pack_file_resume(transfer_id: bytes, from_chunk: int) -> bytes: if len(transfer_id) != 16: raise ValueError("transfer_id must be 16 bytes") return transfer_id + struct.pack(">I", int(from_chunk) & 0xFFFFFFFF) def unpack_file_resume(pt: bytes) -> Dict[str, Any]: if len(pt) < 20: raise ValueError("FILE_RESUME too short") return { "transfer_id": pt[:16], "from_chunk": struct.unpack_from(">I", pt, 16)[0], } def pack_close_payload(code: int = 0, reason: bytes = b"") -> bytes: return struct.pack(">H", int(code) & 0xFFFF) + (reason or b"") def unpack_close_payload(pt: bytes) -> Dict[str, Any]: if len(pt) < 2: return {"code": 0, "reason": b""} return {"code": struct.unpack_from(">H", pt, 0)[0], "reason": pt[2:]} def make_aad(bin_ver: int, ptype: int, stream_id: int = STREAM_DEFAULT) -> bytes: """AES-GCM AAD for wire 1.3: binds header fields into the tag.""" return bytes([bin_ver & 0xFF, ptype & 0xFF]) + struct.pack(">H", int(stream_id) & 0xFFFF) def make_aad_14( ptype: int, stream_id: int = STREAM_DEFAULT, pkt_seq: int = 0 ) -> bytes: """AES-GCM AAD for wire 1.4: ver|ptype|stream_id|pkt_seq.""" return ( bytes([BIN_VERSION_14, ptype & 0xFF]) + struct.pack(">H", int(stream_id) & 0xFFFF) + struct.pack(">I", int(pkt_seq) & 0xFFFFFFFF) ) def pack_window_update(credit: int) -> bytes: return struct.pack(">I", int(credit) & 0xFFFFFFFF) def unpack_window_update(pt: bytes) -> Dict[str, Any]: if len(pt) < 4: raise ValueError("WINDOW_UPDATE too short") return {"credit": struct.unpack_from(">I", pt, 0)[0]} def pack_stream_priority(priority: int) -> bytes: return bytes([int(priority) & 0xFF]) def unpack_stream_priority(pt: bytes) -> Dict[str, Any]: if len(pt) < 1: raise ValueError("STREAM_PRIORITY too short") return {"priority": pt[0]} def pack_stream_cancel(reason: int = 0, note: bytes = b"") -> bytes: note = note or b"" if len(note) > 0xFFFF: raise ValueError("cancel note too long") return struct.pack(">HH", int(reason) & 0xFFFF, len(note)) + note def unpack_stream_cancel(pt: bytes) -> Dict[str, Any]: if len(pt) < 4: raise ValueError("STREAM_CANCEL too short") reason, note_len = struct.unpack_from(">HH", pt, 0) return {"reason": reason, "note": pt[4 : 4 + note_len]} def pack_data_ack(last_seq: int) -> bytes: return struct.pack(">I", int(last_seq) & 0xFFFFFFFF) def unpack_data_ack(pt: bytes) -> Dict[str, Any]: if len(pt) < 4: raise ValueError("DATA_ACK too short") return {"last_seq": struct.unpack_from(">I", pt, 0)[0]} def pack_session_ticket( session_id: bytes, token: bytes, expire_unix: int ) -> bytes: if len(session_id) != 16 or len(token) != 32: raise ValueError("SESSION_TICKET fields invalid") return session_id + struct.pack(">I", int(expire_unix) & 0xFFFFFFFF) + token def unpack_session_ticket(pt: bytes) -> Dict[str, Any]: if len(pt) != 52: raise ValueError("SESSION_TICKET must be 52 bytes") return { "session_id": pt[:16], "expire_unix": struct.unpack_from(">I", pt, 16)[0], "token": pt[20:52], } def pack_caps(caps: int) -> bytes: return struct.pack(">I", int(caps) & 0xFFFFFFFF) def unpack_caps(pt: bytes) -> Dict[str, Any]: if len(pt) < 4: raise ValueError("CAPS too short") return {"caps": struct.unpack_from(">I", pt, 0)[0]} def pack_ping_timestamp(timestamp_ms: Optional[int] = None, opaque: bytes = b"") -> bytes: ts = int(time.time() * 1000) if timestamp_ms is None else int(timestamp_ms) return struct.pack(">Q", ts & 0xFFFFFFFFFFFFFFFF) + (opaque or b"") def unpack_ping_timestamp(pt: bytes) -> Dict[str, Any]: if len(pt) < 8: return {"timestamp_ms": None, "opaque": pt} return { "timestamp_ms": struct.unpack_from(">Q", pt, 0)[0], "opaque": pt[8:], } def decode_file_payload(ptype: int, plaintext: bytes) -> Dict[str, Any]: if ptype == PTYPE_FILE_BEGIN: return unpack_file_begin(plaintext) if ptype == PTYPE_FILE_CHUNK: return unpack_file_chunk(plaintext) if ptype == PTYPE_FILE_END: return unpack_file_end(plaintext) if ptype == PTYPE_FILE_ABORT: return unpack_file_abort(plaintext) if ptype == PTYPE_FILE_ACK: return unpack_file_ack(plaintext) if ptype == PTYPE_FILE_RESUME: return unpack_file_resume(plaintext) raise ValueError(f"not a file ptype: 0x{ptype:02x}") class DVeSession: """ Сессия после рукопожатия. Server: decrypt c2s (от клиента), encrypt s2c (к клиенту). Production API (1.5 LTS): send / receive / stats / close / rekey. """ __slots__ = ( "_c2s", "_s2c", "wire_version", "_send_prefix", "_send_counter", "_send_credit", "_recv_credit_given", "_pkt_seq_send", "_pkt_seq_recv", "_stream_priority", "_stream_cancelled", "_last_ack_sent", "_last_ack_recv", "_caps", "_session_id", "_resume_token", "_started_at", "_last_activity", "_bytes_sent", "_bytes_recv", "_rtt_ms", "_rtt_samples", "_pending_ping_ms", "_recv_seq_hi", "_recv_seq_seen", "_closed", "_c2s_raw", "_s2c_raw", ) def __init__( self, c2s_key: bytes, s2c_key: bytes, wire_version: str = WIRE_V10, caps: int = 0 ) -> None: self._c2s_raw = bytes(c2s_key) self._s2c_raw = bytes(s2c_key) self._c2s = AESGCM(c2s_key) self._s2c = AESGCM(s2c_key) w = normalize_wire(wire_version) or WIRE_V10 self.wire_version = w self._send_prefix = os.urandom(4) self._send_counter = 0 self._send_credit: Dict[int, int] = {STREAM_DEFAULT: DEFAULT_STREAM_WINDOW} self._recv_credit_given: Dict[int, int] = {STREAM_DEFAULT: DEFAULT_STREAM_WINDOW} self._pkt_seq_send = 0 self._pkt_seq_recv = 0 self._stream_priority: Dict[int, int] = {STREAM_DEFAULT: PRIORITY_CHAT} self._stream_cancelled: Dict[int, bool] = {} self._last_ack_sent: Dict[int, int] = {} self._last_ack_recv: Dict[int, int] = {} self._caps = int(caps) if caps else (DEFAULT_CAPS_14 if w == WIRE_V14 else 0) self._session_id: Optional[bytes] = None self._resume_token: Optional[bytes] = None now = time.time() self._started_at = now self._last_activity = now self._bytes_sent = 0 self._bytes_recv = 0 self._rtt_ms: Optional[float] = None self._rtt_samples: List[float] = [] self._pending_ping_ms: Optional[int] = None self._recv_seq_hi: Dict[int, int] = {} self._recv_seq_seen: Dict[int, set] = {} self._closed = False def _touch(self) -> None: self._last_activity = time.time() def _ensure_alive(self) -> None: if self._closed: raise RuntimeError("DVeProto session closed") if time.time() - self._started_at > SESSION_MAX_LIFETIME_S: raise RuntimeError("DVeProto session max lifetime exceeded") # Idle timeout is soft (see is_alive / stats); PING keeps sessions warm. def is_alive(self) -> bool: if self._closed: return False now = time.time() return (now - self._started_at) <= SESSION_MAX_LIFETIME_S and ( now - self._last_activity ) <= SESSION_IDLE_TIMEOUT_S def needs_rekey(self) -> bool: return (self._bytes_sent + self._bytes_recv) >= REKEY_AFTER_BYTES def rekey_from_token(self, token: bytes) -> None: """Rotate AEAD keys from a fresh 32-byte secret (ticket-based rekey).""" c2s, s2c = derive_resume_keys(token) self._c2s_raw, self._s2c_raw = c2s, s2c self._c2s, self._s2c = AESGCM(c2s), AESGCM(s2c) self._send_prefix = os.urandom(4) self._send_counter = 0 self._pkt_seq_send = 0 self._pkt_seq_recv = 0 self._recv_seq_hi.clear() self._recv_seq_seen.clear() self._resume_token = token self._bytes_sent = 0 self._bytes_recv = 0 self._touch() def _check_plaintext_size(self, plaintext: bytes) -> None: if len(plaintext) > MAX_PLAINTEXT_BYTES: raise ValueError( f"DVeProto frame plaintext too large: {len(plaintext)} > {MAX_PLAINTEXT_BYTES}" ) def _check_frame_size(self, raw: bytes) -> None: if len(raw) > MAX_FRAME_BYTES: raise ValueError( f"DVeProto wire frame too large: {len(raw)} > {MAX_FRAME_BYTES}" ) def _check_replay(self, stream_id: int, pkt_seq: int) -> None: sid = int(stream_id) & 0xFFFF seq = int(pkt_seq) & 0xFFFFFFFF if seq == 0: raise ValueError("replay: pkt_seq must start at 1") hi = self._recv_seq_hi.get(sid, 0) if hi and seq + REPLAY_WINDOW <= hi: raise ValueError("replay: pkt_seq outside window") seen = self._recv_seq_seen.setdefault(sid, set()) if seq in seen: raise ValueError("replay: duplicate pkt_seq") seen.add(seq) if seq > hi: self._recv_seq_hi[sid] = seq if len(seen) > REPLAY_WINDOW * 2: cutoff = self._recv_seq_hi[sid] - REPLAY_WINDOW self._recv_seq_seen[sid] = {s for s in seen if s > cutoff} @property def is_binary(self) -> bool: return self.wire_version in (WIRE_V11, WIRE_V12, WIRE_V13, WIRE_V14) @property def is_mux(self) -> bool: return self.wire_version in (WIRE_V12, WIRE_V13, WIRE_V14) @property def is_v13(self) -> bool: return self.wire_version == WIRE_V13 @property def is_v14(self) -> bool: return self.wire_version == WIRE_V14 @property def has_aad_fc(self) -> bool: return self.wire_version in (WIRE_V13, WIRE_V14) @classmethod def from_server_keys( cls, server_private: X25519PrivateKey, client_public_raw: bytes, wire_version: str = WIRE_V10, caps: int = 0, ) -> "DVeSession": client_pub = X25519PublicKey.from_public_bytes(client_public_raw) shared = server_private.exchange(client_pub) c2s, s2c = derive_aes_keys(shared) return cls(c2s, s2c, wire_version=wire_version, caps=caps) @classmethod def from_resume_token( cls, token: bytes, wire_version: str = WIRE_V14, caps: int = 0 ) -> "DVeSession": c2s, s2c = derive_resume_keys(token) sess = cls(c2s, s2c, wire_version=wire_version, caps=caps or DEFAULT_CAPS_14) sess._resume_token = token return sess def stats(self) -> Dict[str, Any]: jitter = None if len(self._rtt_samples) >= 2: diffs = [ abs(self._rtt_samples[i] - self._rtt_samples[i - 1]) for i in range(1, len(self._rtt_samples)) ] jitter = sum(diffs) / len(diffs) return { "package": PACKAGE_VERSION, "wire": self.wire_version, "rtt_ms": self._rtt_ms, "rtt_jitter_ms": jitter, "send_window": dict(self._send_credit), "streams": sorted(set(self._send_credit) | set(self._stream_priority)), "stream_priority": dict(self._stream_priority), "uptime_s": round(time.time() - self._started_at, 3), "idle_s": round(time.time() - self._last_activity, 3), "alive": self.is_alive(), "needs_rekey": self.needs_rekey(), "pkt_seq_send": self._pkt_seq_send, "pkt_seq_recv": self._pkt_seq_recv, "bytes_sent": self._bytes_sent, "bytes_recv": self._bytes_recv, "caps": self._caps, "session_id": _b64e(self._session_id) if self._session_id else None, "closed": self._closed, } # --- Production API (1.5 LTS) -------------------------------------------- def send( self, obj: Dict[str, Any], stream_id: int = STREAM_DEFAULT ) -> Union[str, bytes]: """Server→client APP_JSON (or JSON 1.0).""" self._ensure_alive() if self.is_binary: return self.encrypt_app_json(obj, stream_id=stream_id) return self.encrypt_json(obj) def receive( self, message: Union[str, bytes, bytearray] ) -> Union[Dict[str, Any], DecodedFrame]: """Decrypt inbound; returns DecodedFrame (binary) or dict (1.0).""" self._ensure_alive() return self.decrypt_raw(message) def close(self, code: int = 0, reason: bytes = b"") -> Union[str, bytes]: """Emit CLOSE and mark session closed (caller still sends the frame).""" if self.is_binary: frame = self.encrypt_close(code, reason) else: frame = self.encrypt_json( {"type": "close", "code": int(code), "reason": (reason or b"").decode("utf-8", "replace")} ) self._closed = True return frame # --- v1.0 JSON ----------------------------------------------------------- def encrypt_json(self, obj: Dict[str, Any]) -> str: nonce = os.urandom(NONCE_LEN) pt = json.dumps(obj, separators=(",", ":"), ensure_ascii=False).encode("utf-8") ct = self._s2c.encrypt(nonce, pt, None) return json.dumps( { "type": "dve", "proto": PROTO_NAME, "v": PROTO_HANDSHAKE_VERSION, "n": _b64e(nonce), "c": _b64e(ct), }, separators=(",", ":"), ensure_ascii=False, ) def decrypt_json(self, text: str) -> Dict[str, Any]: data = json.loads(text) if data.get("type") != "dve" or data.get("proto") != PROTO_NAME: raise ValueError("invalid DVe frame") if int(data.get("v", 0)) != PROTO_HANDSHAKE_VERSION: raise ValueError("unsupported DVe version") nonce = _b64d(data["n"]) ct = _b64d(data["c"]) pt = self._c2s.decrypt(nonce, ct, None) return json.loads(pt.decode("utf-8")) # --- v1.1 / v1.2 binary -------------------------------------------------- def _bin_ver(self) -> int: if self.is_v14: return BIN_VERSION_14 if self.is_v13: return BIN_VERSION_13 if self.is_mux: return BIN_VERSION_12 return BIN_VERSION_11 def _header_len(self) -> int: if self.is_v14: return HEADER_LEN_14 return HEADER_LEN_12 if self.is_mux else HEADER_LEN_11 def _next_nonce(self) -> bytes: self._send_counter += 1 if self._send_counter >= (1 << 64): raise RuntimeError("DVeProto nonce counter exhausted") return self._send_prefix + struct.pack(">Q", self._send_counter) def _next_pkt_seq(self) -> int: self._pkt_seq_send += 1 if self._pkt_seq_send >= (1 << 32): raise RuntimeError("DVeProto pkt_seq exhausted") return self._pkt_seq_send def _ensure_send_credit(self, stream_id: int, nbytes: int) -> None: sid = int(stream_id) & 0xFFFF credit = self._send_credit.get(sid) if credit is None: credit = DEFAULT_STREAM_WINDOW self._send_credit[sid] = credit if nbytes > credit: raise ValueError( f"DVeProto flow control: need {nbytes} bytes, credit={credit} on stream {sid}" ) self._send_credit[sid] = credit - nbytes @staticmethod def ptype_exempt_from_window(ptype: int) -> bool: return ptype in ( PTYPE_PING, PTYPE_PONG, PTYPE_CLOSE, PTYPE_WINDOW_UPDATE, PTYPE_STREAM_PRIORITY, PTYPE_STREAM_CANCEL, PTYPE_DATA_ACK, PTYPE_SESSION_TICKET, PTYPE_CAPS, ) def apply_window_update(self, stream_id: int, credit: int) -> None: sid = int(stream_id) & 0xFFFF add = int(credit) & 0xFFFFFFFF self._send_credit[sid] = self._send_credit.get(sid, 0) + add def set_stream_priority(self, stream_id: int, priority: int) -> None: self._stream_priority[int(stream_id) & 0xFFFF] = int(priority) & 0xFF def note_pong_rtt(self, pong_plaintext: bytes) -> Optional[float]: meta = unpack_ping_timestamp(pong_plaintext) if meta["timestamp_ms"] is None: return None rtt = max(0.0, (time.time() * 1000) - float(meta["timestamp_ms"])) self._rtt_ms = rtt self._rtt_samples.append(rtt) if len(self._rtt_samples) > 32: self._rtt_samples = self._rtt_samples[-32:] return rtt def _seal( self, aead: AESGCM, ptype: int, plaintext: bytes, stream_id: int = STREAM_DEFAULT, ) -> bytes: self._ensure_alive() self._check_plaintext_size(plaintext) self._touch() if not (0 <= ptype <= 255): raise ValueError("ptype out of range") sid = int(stream_id) & 0xFFFF if self._stream_cancelled.get(sid): raise ValueError(f"stream {sid} cancelled") if self.has_aad_fc and not self.ptype_exempt_from_window(ptype): self._ensure_send_credit(sid, len(plaintext)) if self.is_v14: pkt_seq = self._next_pkt_seq() nonce = self._next_nonce() aad = make_aad_14(ptype, sid, pkt_seq) ct = aead.encrypt(nonce, plaintext, aad) out = bytearray(HEADER_LEN_14 + len(ct)) out[0] = BIN_VERSION_14 out[1] = ptype out[2] = (sid >> 8) & 0xFF out[3] = sid & 0xFF struct.pack_into(">I", out, 4, pkt_seq) out[8:20] = nonce out[20:] = ct frame = bytes(out) self._bytes_sent += len(frame) return frame if self.is_v13: nonce = self._next_nonce() aad = make_aad(BIN_VERSION_13, ptype, sid) ct = aead.encrypt(nonce, plaintext, aad) out = bytearray(HEADER_LEN_13 + len(ct)) out[0] = BIN_VERSION_13 out[1] = ptype out[2] = (sid >> 8) & 0xFF out[3] = sid & 0xFF out[4:16] = nonce out[16:] = ct frame = bytes(out) self._bytes_sent += len(frame) return frame nonce = os.urandom(NONCE_LEN) ct = aead.encrypt(nonce, plaintext, None) # ct || tag if self.is_mux: out = bytearray(HEADER_LEN_12 + len(ct)) out[0] = BIN_VERSION_12 out[1] = ptype out[2] = (sid >> 8) & 0xFF out[3] = sid & 0xFF out[4:16] = nonce out[16:] = ct else: out = bytearray(HEADER_LEN_11 + len(ct)) out[0] = BIN_VERSION_11 out[1] = ptype out[2:14] = nonce out[14:] = ct frame = bytes(out) self._bytes_sent += len(frame) return frame def _open(self, aead: AESGCM, frame: bytes) -> DecodedFrame: if not isinstance(frame, (bytes, bytearray, memoryview)): raise TypeError("binary frame required") self._ensure_alive() raw = bytes(frame) self._check_frame_size(raw) self._bytes_recv += len(raw) self._touch() if len(raw) < 2 + NONCE_LEN + GCM_TAG_LEN: raise ValueError("binary frame too short") ver = raw[0] ptype = raw[1] if ver == BIN_VERSION_14: if len(raw) < HEADER_LEN_14 + GCM_TAG_LEN: raise ValueError("binary frame too short") if not self.is_v14: raise ValueError("got 1.4 frame on non-1.4 session") stream_id = (raw[2] << 8) | raw[3] pkt_seq = struct.unpack_from(">I", raw, 4)[0] self._check_replay(stream_id, pkt_seq) nonce = raw[8:20] ct = raw[20:] aad = make_aad_14(ptype, stream_id, pkt_seq) pt = aead.decrypt(nonce, ct, aad) self._pkt_seq_recv = max(self._pkt_seq_recv, pkt_seq) self._apply_control_side_effects(ptype, pt, stream_id) return DecodedFrame( ptype=ptype, plaintext=pt, stream_id=stream_id, bin_ver=ver, pkt_seq=pkt_seq, ) if ver == BIN_VERSION_13: if len(raw) < HEADER_LEN_13 + GCM_TAG_LEN: raise ValueError("binary frame too short") if not self.is_v13: raise ValueError("got 1.3 frame on non-1.3 session") stream_id = (raw[2] << 8) | raw[3] nonce = raw[4:16] ct = raw[16:] aad = make_aad(BIN_VERSION_13, ptype, stream_id) pt = aead.decrypt(nonce, ct, aad) if ptype == PTYPE_WINDOW_UPDATE: meta = unpack_window_update(pt) self.apply_window_update(stream_id, meta["credit"]) return DecodedFrame( ptype=ptype, plaintext=pt, stream_id=stream_id, bin_ver=ver ) if ver == BIN_VERSION_12: if len(raw) < HEADER_LEN_12 + GCM_TAG_LEN: raise ValueError("binary frame too short") if self.wire_version != WIRE_V12: raise ValueError("got 1.2 frame on mismatched session") stream_id = (raw[2] << 8) | raw[3] nonce = raw[4:16] ct = raw[16:] pt = aead.decrypt(nonce, ct, None) return DecodedFrame( ptype=ptype, plaintext=pt, stream_id=stream_id, bin_ver=ver ) if ver == BIN_VERSION_11: if len(raw) < HEADER_LEN_11 + GCM_TAG_LEN: raise ValueError("binary frame too short") if self.is_mux: raise ValueError("got 1.1 frame on mux session") nonce = raw[2:14] ct = raw[14:] pt = aead.decrypt(nonce, ct, None) return DecodedFrame( ptype=ptype, plaintext=pt, stream_id=STREAM_DEFAULT, bin_ver=ver ) raise ValueError("unsupported binary DVe version") def _apply_control_side_effects( self, ptype: int, pt: bytes, stream_id: int ) -> None: if ptype == PTYPE_WINDOW_UPDATE: meta = unpack_window_update(pt) self.apply_window_update(stream_id, meta["credit"]) elif ptype == PTYPE_STREAM_PRIORITY: meta = unpack_stream_priority(pt) self.set_stream_priority(stream_id, meta["priority"]) elif ptype == PTYPE_STREAM_CANCEL: self._stream_cancelled[stream_id] = True elif ptype == PTYPE_DATA_ACK: meta = unpack_data_ack(pt) self._last_ack_recv[stream_id] = meta["last_seq"] elif ptype == PTYPE_SESSION_TICKET: meta = unpack_session_ticket(pt) self._session_id = meta["session_id"] self._resume_token = meta["token"] # Mid-session ticket with expire==0 means immediate rekey (1.5 convention). if int(meta["expire_unix"]) == 0: self.rekey_from_token(meta["token"]) elif ptype == PTYPE_CAPS: meta = unpack_caps(pt) self._caps = meta["caps"] elif ptype == PTYPE_PONG: self.note_pong_rtt(pt) def encrypt_bin( self, ptype: int, plaintext: bytes, stream_id: int = STREAM_DEFAULT, ) -> bytes: """Сервер → клиент: binary кадр (s2c).""" return self._seal(self._s2c, ptype, plaintext, stream_id=stream_id) def decrypt_bin(self, frame: bytes) -> DecodedFrame: """Клиент → сервер: binary кадр (c2s).""" return self._open(self._c2s, frame) def encrypt_app_json( self, obj: Dict[str, Any], stream_id: int = STREAM_DEFAULT ) -> bytes: pt = json.dumps(obj, separators=(",", ":"), ensure_ascii=False).encode("utf-8") return self.encrypt_bin(PTYPE_APP_JSON, pt, stream_id=stream_id) def encrypt_app_bin( self, data: bytes, stream_id: int = STREAM_DEFAULT ) -> bytes: return self.encrypt_bin(PTYPE_APP_BIN, data, stream_id=stream_id) def encrypt_text(self, text: str, stream_id: int = STREAM_DEFAULT) -> bytes: return self.encrypt_bin(PTYPE_TEXT, text.encode("utf-8"), stream_id=stream_id) def encrypt_ping( self, payload: bytes = b"", stream_id: int = STREAM_DEFAULT ) -> bytes: return self.encrypt_bin(PTYPE_PING, payload, stream_id=stream_id) def encrypt_pong( self, payload: bytes = b"", stream_id: int = STREAM_DEFAULT ) -> bytes: return self.encrypt_bin(PTYPE_PONG, payload, stream_id=stream_id) def encrypt_close( self, code: int = 0, reason: bytes = b"", stream_id: int = STREAM_DEFAULT, ) -> bytes: return self.encrypt_bin( PTYPE_CLOSE, pack_close_payload(code, reason), stream_id=stream_id ) def encrypt_window_update( self, credit: int, stream_id: int = STREAM_DEFAULT ) -> bytes: if not self.has_aad_fc: raise ValueError("WINDOW_UPDATE requires wire 1.3+") frame = self.encrypt_bin( PTYPE_WINDOW_UPDATE, pack_window_update(credit), stream_id=stream_id ) # Track how much receive-side credit we advertised (informational). sid = int(stream_id) & 0xFFFF self._recv_credit_given[sid] = self._recv_credit_given.get(sid, 0) + ( int(credit) & 0xFFFFFFFF ) return frame def encrypt_ping_rtt( self, opaque: bytes = b"", stream_id: int = STREAM_DEFAULT ) -> bytes: pt = pack_ping_timestamp(opaque=opaque) self._pending_ping_ms = unpack_ping_timestamp(pt)["timestamp_ms"] return self.encrypt_bin(PTYPE_PING, pt, stream_id=stream_id) def encrypt_stream_priority( self, priority: int, stream_id: int = STREAM_DEFAULT ) -> bytes: if not self.is_v14: raise ValueError("STREAM_PRIORITY requires wire 1.4") self.set_stream_priority(stream_id, priority) return self.encrypt_bin( PTYPE_STREAM_PRIORITY, pack_stream_priority(priority), stream_id=stream_id ) def encrypt_stream_cancel( self, reason: int = 0, note: bytes = b"", stream_id: int = STREAM_DEFAULT, ) -> bytes: if not self.is_v14: raise ValueError("STREAM_CANCEL requires wire 1.4") frame = self.encrypt_bin( PTYPE_STREAM_CANCEL, pack_stream_cancel(reason, note), stream_id=stream_id, ) self._stream_cancelled[int(stream_id) & 0xFFFF] = True return frame def encrypt_data_ack( self, last_seq: Optional[int] = None, stream_id: int = STREAM_DEFAULT ) -> bytes: if not self.is_v14: raise ValueError("DATA_ACK requires wire 1.4") sid = int(stream_id) & 0xFFFF seq = self._pkt_seq_recv if last_seq is None else int(last_seq) self._last_ack_sent[sid] = seq return self.encrypt_bin(PTYPE_DATA_ACK, pack_data_ack(seq), stream_id=sid) def issue_session_ticket( self, ttl_s: int = DEFAULT_TICKET_TTL_S, stream_id: int = STREAM_DEFAULT, *, rekey: bool = False, ) -> Tuple[bytes, bytes, bytes]: """Server helper: create ticket, store on session, return encrypted frame + ids. If rekey=True, switch AEAD keys to the new token after sealing the ticket (caller must ensure peer applies the same rekey after decrypting the ticket). """ if not self.is_v14: raise ValueError("SESSION_TICKET requires wire 1.4+") ttl = max(MIN_TICKET_TTL_S, min(int(ttl_s), MAX_TICKET_TTL_S)) session_id = os.urandom(16) token = os.urandom(32) expire = int(time.time()) + ttl self._session_id = session_id self._resume_token = token frame = self.encrypt_bin( PTYPE_SESSION_TICKET, pack_session_ticket(session_id, token, expire), stream_id=stream_id, ) if rekey: self.rekey_from_token(token) return frame, session_id, token def encrypt_caps(self, caps: Optional[int] = None, stream_id: int = STREAM_DEFAULT) -> bytes: if not self.is_v14: raise ValueError("CAPS requires wire 1.4") value = self._caps if caps is None else int(caps) self._caps = value return self.encrypt_bin(PTYPE_CAPS, pack_caps(value), stream_id=stream_id) def encrypt_file_begin( self, stream_id: int = STREAM_DEFAULT, **kwargs: Any ) -> bytes: return self.encrypt_bin( PTYPE_FILE_BEGIN, pack_file_begin(**kwargs), stream_id=stream_id ) def encrypt_file_chunk( self, transfer_id: bytes, chunk_index: int, data: bytes, stream_id: int = STREAM_DEFAULT, ) -> bytes: return self.encrypt_bin( PTYPE_FILE_CHUNK, pack_file_chunk(transfer_id, chunk_index, data), stream_id=stream_id, ) def encrypt_file_end( self, transfer_id: bytes, sha256: bytes, stream_id: int = STREAM_DEFAULT, ) -> bytes: return self.encrypt_bin( PTYPE_FILE_END, pack_file_end(transfer_id, sha256), stream_id=stream_id ) def encrypt_file_abort( self, transfer_id: bytes, reason_code: int = 0, stream_id: int = STREAM_DEFAULT, ) -> bytes: return self.encrypt_bin( PTYPE_FILE_ABORT, pack_file_abort(transfer_id, reason_code), stream_id=stream_id, ) def encrypt_file_ack( self, transfer_id: bytes, status: int, last_ok: int, stream_id: int = STREAM_DEFAULT, ) -> bytes: return self.encrypt_bin( PTYPE_FILE_ACK, pack_file_ack(transfer_id, status, last_ok), stream_id=stream_id, ) def encrypt_file_resume( self, transfer_id: bytes, from_chunk: int, stream_id: int = STREAM_DEFAULT, ) -> bytes: return self.encrypt_bin( PTYPE_FILE_RESUME, pack_file_resume(transfer_id, from_chunk), stream_id=stream_id, ) # --- Unified API used by ws_server --------------------------------------- def encrypt(self, obj: Dict[str, Any]) -> Union[str, bytes]: """Шифрование объекта приложения (JSON plaintext) в формате выбранного wire.""" if self.is_binary: return self.encrypt_app_json(obj) return self.encrypt_json(obj) def decrypt(self, message: Union[str, bytes, bytearray]) -> Dict[str, Any]: """ Расшифровка кадра приложения. Для 1.1/1.2 принимает только binary; APP_JSON → dict. Файловые / control кадры — через decrypt_raw / decrypt_bin. """ if self.is_binary: if isinstance(message, str): raise ValueError("DVeProto binary wire rejects text frames after handshake") frame = self.decrypt_bin(message) if frame.ptype != PTYPE_APP_JSON: raise ValueError( f"expected APP_JSON, got ptype=0x{frame.ptype:02x}" ) return frame.as_json() if isinstance(message, (bytes, bytearray)): raise ValueError("DVeProto 1.0 expects text JSON frames") return self.decrypt_json(message) def decrypt_raw( self, message: Union[str, bytes, bytearray] ) -> Union[Dict[str, Any], DecodedFrame]: """Как decrypt, но для binary возвращает DecodedFrame (любой ptype).""" if self.is_binary: if isinstance(message, str): raise ValueError("DVeProto binary wire rejects text frames after handshake") return self.decrypt_bin(message) return self.decrypt_json(message) def server_hello_payload( server_private: X25519PrivateKey, offer: Iterable[str] = DEFAULT_OFFER, caps: Optional[int] = None, ) -> str: pub = server_private.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw, ) # Preserve explicit "1.5" LTS label in offer (normalize_wire maps 1.5→1.4 binary). offer_list: List[str] = [] for item in offer: raw = str(item).strip() if item is not None else "" if raw in ("1.5", "v1.5", "lts") or item == WIRE_V15: if WIRE_V15 not in offer_list: offer_list.append(WIRE_V15) continue w = normalize_wire(item) if w and w not in offer_list: offer_list.append(w) if not offer_list: offer_list = [WIRE_V10] payload: Dict[str, Any] = { "type": "dve_hello", "proto": PROTO_NAME, "v": PROTO_HANDSHAKE_VERSION, "server_pk": _b64e(pub), "offer": offer_list, } if caps is None and (WIRE_V14 in offer_list or WIRE_V15 in offer_list): caps = DEFAULT_CAPS_14 if caps is not None: payload["caps"] = int(caps) & 0xFFFFFFFF return json.dumps(payload, separators=(",", ":"), ensure_ascii=False) def parse_client_ack( text: str, offer: Iterable[str] = DEFAULT_OFFER ) -> Optional[ClientAck]: try: data = json.loads(text) except (json.JSONDecodeError, TypeError): return None if data.get("type") != "dve_client_ack": return None if data.get("proto") != PROTO_NAME: return None if int(data.get("v", 0)) != PROTO_HANDSHAKE_VERSION: return None if "select" not in data: wire = WIRE_V10 else: wire = pick_wire(offer, data.get("select")) caps = int(data["caps"]) & 0xFFFFFFFF if "caps" in data else 0 resume = data.get("resume") if isinstance(resume, dict): try: sid = _b64d(resume["session_id"]) token = _b64d(resume["token"]) except (KeyError, ValueError, TypeError): return None if len(sid) != 16 or len(token) != 32: return None return ClientAck( client_pk=None, select=wire, resume_session_id=sid, resume_token=token, caps=caps, ) try: client_pk = _b64d(data["client_pk"]) except (KeyError, ValueError): return None if len(client_pk) != 32: return None return ClientAck(client_pk=client_pk, select=wire, caps=caps) def resume_reject_payload(reason: str = "invalid_ticket") -> str: return json.dumps( { "type": "dve_resume_reject", "proto": PROTO_NAME, "v": PROTO_HANDSHAKE_VERSION, "reason": reason, }, separators=(",", ":"), ensure_ascii=False, ) @dataclass class TicketRecord: token: bytes expire_unix: int identity: str = "" caps: int = 0 class DVeTicketStore: """ Session Ticket Store for wire 1.4 Resume (1.5 LTS). Maps session_id(16) → TicketRecord. Compare tokens with constant-time equality. Bind tickets to an authenticated identity when available. """ __slots__ = ("_by_id", "_default_ttl") def __init__(self, default_ttl_s: int = DEFAULT_TICKET_TTL_S) -> None: self._by_id: Dict[bytes, TicketRecord] = {} self._default_ttl = max( MIN_TICKET_TTL_S, min(int(default_ttl_s), MAX_TICKET_TTL_S) ) def put( self, session_id: bytes, token: bytes, *, identity: str = "", caps: int = 0, ttl_s: Optional[int] = None, expire_unix: Optional[int] = None, ) -> int: if len(session_id) != 16 or len(token) != 32: raise ValueError("invalid ticket fields") if expire_unix is not None: exp = int(expire_unix) else: ttl = self._default_ttl if ttl_s is None else int(ttl_s) ttl = max(MIN_TICKET_TTL_S, min(ttl, MAX_TICKET_TTL_S)) exp = int(time.time()) + ttl self._by_id[bytes(session_id)] = TicketRecord( token=bytes(token), expire_unix=exp, identity=str(identity or ""), caps=int(caps) & 0xFFFFFFFF, ) return exp def get(self, session_id: bytes) -> Optional[TicketRecord]: return self._by_id.get(bytes(session_id)) def pop(self, session_id: bytes) -> Optional[TicketRecord]: return self._by_id.pop(bytes(session_id), None) def purge_expired(self, now: Optional[float] = None) -> int: t = int(now if now is not None else time.time()) dead = [sid for sid, rec in self._by_id.items() if rec.expire_unix <= t] for sid in dead: del self._by_id[sid] return len(dead) def validate( self, session_id: bytes, token: bytes, *, identity: Optional[str] = None, ) -> Tuple[bool, str, Optional[TicketRecord]]: """ Returns (ok, reason, record). reason: ok | not_found | expired | bad_token | identity_mismatch """ self.purge_expired() rec = self.get(session_id) if rec is None: return False, "not_found", None if rec.expire_unix <= int(time.time()): self.pop(session_id) return False, "expired", None if not hmac_compare(rec.token, token): return False, "bad_token", None if identity is not None and rec.identity and rec.identity != identity: return False, "identity_mismatch", None return True, "ok", rec def hmac_compare(a: bytes, b: bytes) -> bool: """Constant-time compare; lengths must match for True.""" if not isinstance(a, (bytes, bytearray)) or not isinstance(b, (bytes, bytearray)): return False if len(a) != len(b): return False try: import hmac as _hmac return _hmac.compare_digest(bytes(a), bytes(b)) except Exception: diff = 0 for x, y in zip(a, b): diff |= x ^ y return diff == 0 # Обратная совместимость имени константы в старом коде. PROTO_VERSION = PROTO_HANDSHAKE_VERSION # --- Client / developer helpers ---------------------------------------------- class DVeSeq: __slots__ = ("_n",) def __init__(self) -> None: self._n = 0 def next(self) -> int: self._n += 1 return self._n def dve_app_message(op: str, body=None, *, seq=None): msg = {"dve_op": op} if seq is not None: msg["dve_seq"] = int(seq) if body: msg.update(body) return msg class DVeClientSession: """Клиент: исходящие c2s, входящие s2c. Production API: send/receive/stats/resume/close.""" __slots__ = ("_inner",) def __init__( self, c2s_key: bytes, s2c_key: bytes, wire_version: str = WIRE_V10, caps: int = 0 ): self._inner = DVeSession(c2s_key, s2c_key, wire_version=wire_version, caps=caps) @property def wire_version(self): return self._inner.wire_version @wire_version.setter def wire_version(self, v): self._inner.wire_version = normalize_wire(v) or WIRE_V10 @property def is_binary(self): return self._inner.is_binary @property def is_mux(self): return self._inner.is_mux def stats(self): return self._inner.stats() def send(self, obj, stream_id: int = STREAM_DEFAULT): """Client→server APP_JSON.""" return self.pack_outgoing(obj, stream_id=stream_id) def receive(self, message): """Decrypt inbound frame (DecodedFrame for binary).""" if self.is_binary: return self.unpack_bin(message) return self.unpack_incoming(message) def close(self, code: int = 0, reason: bytes = b""): frame = self.pack_bin(PTYPE_CLOSE, pack_close_payload(code, reason)) self._inner._closed = True return frame @classmethod def resume(cls, hello_text: str, session_id: bytes, token: bytes, prefer: str = WIRE_V15): return cls.from_resume(hello_text, session_id, token, prefer=prefer) @classmethod def from_server_hello_text(cls, hello_text: str, prefer: str = WIRE_V15): data = json.loads(hello_text) if data.get("type") != "dve_hello" or data.get("proto") != PROTO_NAME: raise ValueError("ожидался dve_hello") offer = data.get("offer") or [WIRE_V10] selected = pick_wire(offer, prefer) caps = int(data.get("caps") or 0) server_pub = X25519PublicKey.from_public_bytes(_b64d(data["server_pk"])) client_priv = X25519PrivateKey.generate() shared = client_priv.exchange(server_pub) c2s, s2c = derive_aes_keys(shared) client_pub = client_priv.public_key().public_bytes( encoding=serialization.Encoding.Raw, format=serialization.PublicFormat.Raw) ack_obj: Dict[str, Any] = { "type": "dve_client_ack", "proto": PROTO_NAME, "v": PROTO_HANDSHAKE_VERSION, "client_pk": _b64e(client_pub), } if selected != WIRE_V10 or "offer" in data: if selected == WIRE_V14 and WIRE_V15 in list(offer): ack_obj["select"] = WIRE_V15 else: ack_obj["select"] = selected if selected == WIRE_V14: ack_obj["caps"] = caps & DEFAULT_CAPS_14 if caps else DEFAULT_CAPS_14 return cls(c2s, s2c, wire_version=selected, caps=ack_obj.get("caps", 0)), json.dumps( ack_obj, separators=(",", ":"), ensure_ascii=False ) @classmethod def from_resume( cls, hello_text: str, session_id: bytes, token: bytes, prefer: str = WIRE_V15, ): data = json.loads(hello_text) if data.get("type") != "dve_hello" or data.get("proto") != PROTO_NAME: raise ValueError("ожидался dve_hello") offer = data.get("offer") or [WIRE_V10] selected = pick_wire(offer, prefer) if selected != WIRE_V14: raise ValueError("resume requires wire 1.4+") c2s, s2c = derive_resume_keys(token) caps = int(data.get("caps") or DEFAULT_CAPS_14) select_label = WIRE_V15 if WIRE_V15 in offer else WIRE_V14 ack_obj = { "type": "dve_client_ack", "proto": PROTO_NAME, "v": PROTO_HANDSHAKE_VERSION, "select": select_label, "resume": {"session_id": _b64e(session_id), "token": _b64e(token)}, "caps": caps & DEFAULT_CAPS_14, } sess = cls(c2s, s2c, wire_version=WIRE_V14, caps=ack_obj["caps"]) sess._inner._session_id = session_id sess._inner._resume_token = token return sess, json.dumps(ack_obj, separators=(",", ":"), ensure_ascii=False) def pack_outgoing(self, obj, stream_id: int = STREAM_DEFAULT): if self.is_binary: pt = json.dumps(obj, separators=(",",":"), ensure_ascii=False).encode() return self._inner._seal(self._inner._c2s, PTYPE_APP_JSON, pt, stream_id=stream_id) nonce = os.urandom(NONCE_LEN) pt = json.dumps(obj, separators=(",",":"), ensure_ascii=False).encode() ct = self._inner._c2s.encrypt(nonce, pt, None) return json.dumps({"type":"dve","proto":PROTO_NAME,"v":PROTO_HANDSHAKE_VERSION,"n":_b64e(nonce),"c":_b64e(ct)}, separators=(",",":"), ensure_ascii=False) def unpack_incoming(self, message): if self.is_binary: if isinstance(message, str): raise ValueError("binary wire rejects text") frame = self._inner._open(self._inner._s2c, message) if frame.ptype != PTYPE_APP_JSON: raise ValueError(f"expected APP_JSON, got 0x{frame.ptype:02x}") return frame.as_json() data = json.loads(message) pt = self._inner._s2c.decrypt(_b64d(data["n"]), _b64d(data["c"]), None) return json.loads(pt.decode()) def pack_bin(self, ptype, plaintext, stream_id=STREAM_DEFAULT): return self._inner._seal(self._inner._c2s, ptype, plaintext, stream_id=stream_id) def unpack_bin(self, frame): return self._inner._open(self._inner._s2c, frame) def pack_ping(self, payload=b"", stream_id=STREAM_DEFAULT): return self.pack_bin(PTYPE_PING, payload, stream_id) def pack_ping_rtt(self, opaque=b"", stream_id=STREAM_DEFAULT): return self.pack_bin(PTYPE_PING, pack_ping_timestamp(opaque=opaque), stream_id) def pack_window_update(self, credit, stream_id=STREAM_DEFAULT): return self.pack_bin(PTYPE_WINDOW_UPDATE, pack_window_update(credit), stream_id) def developer_server_session_from_ack(server_private, client_ack_text, ticket_store=None): ack = parse_client_ack(client_ack_text) if ack is None: raise ValueError("bad ack") if ack.is_resume: if ticket_store is None: raise ValueError("resume requires ticket_store") ok, reason, rec = ticket_store.validate( ack.resume_session_id, ack.resume_token ) if not ok or rec is None: raise ValueError(f"resume rejected: {reason}") return DVeSession.from_resume_token( ack.resume_token, wire_version=WIRE_V14, caps=rec.caps or DEFAULT_CAPS_14 ) if ack.client_pk is None: raise ValueError("missing client_pk") return DVeSession.from_server_keys( server_private, ack.client_pk, wire_version=ack.select, caps=ack.caps )