Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# This Source Code Form is subject to the terms of the Mozilla Public
2# License, v. 2.0. If a copy of the MPL was not distributed with this
3# file, You can obtain one at https://mozilla.org/MPL/2.0/.
4#
5# This file incorporates work covered by the following copyright and
6# permission notice:
7#
8# Copyright (c) Jeremy Lainé.
9# All rights reserved.
10#
11# Redistribution and use in source and binary forms, with or without
12# modification, are permitted provided that the following conditions are met:
13#
14# * Redistributions of source code must retain the above copyright notice,
15# this list of conditions and the following disclaimer.
16# * Redistributions in binary form must reproduce the above copyright notice,
17# this list of conditions and the following disclaimer in the documentation
18# and/or other materials provided with the distribution.
19# * Neither the name of aiortc nor the names of its contributors may
20# be used to endorse or promote products derived from this software without
21# specific prior written permission.
22#
23# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
24# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
25# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
26# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
27# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
28# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
29# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
30# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
31# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
32# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
34import asyncio
35import bisect
36import enum
37import hmac
38import logging
39import math
40import os
41import time
42from collections import deque
43from collections.abc import Callable, Iterator
44from dataclasses import dataclass, field
45from struct import error as StructError, pack, unpack_from
46from typing import Deque, Optional, Union, cast
48from google_crc32c import value as crc32c
49from pyee.asyncio import AsyncIOEventEmitter
51from .exceptions import InvalidStateError
52from .rtcdatachannel import RTCDataChannel, RTCDataChannelParameters
53from .rtcdtlstransport import RTCDtlsTransport
54from .utils import random32, uint16_add, uint16_gt, uint32_gt, uint32_gte
56logger = logging.getLogger(__name__)
58# local constants
59COOKIE_LENGTH = 24
60COOKIE_LIFETIME = 60
61MAX_STREAMS = 65535
62USERDATA_MAX_LENGTH = 1200
64# protocol constants
65SCTP_CAUSE_INVALID_STREAM = 0x0001
66SCTP_CAUSE_STALE_COOKIE = 0x0003
68SCTP_DATA_LAST_FRAG = 0x01
69SCTP_DATA_FIRST_FRAG = 0x02
70SCTP_DATA_UNORDERED = 0x04
71# RFC 7053 I-bit: the sender asks the receiver to acknowledge without delay.
72SCTP_DATA_SACK_IMMEDIATELY = 0x08
74SCTP_MAX_ASSOCIATION_RETRANS = 10
75SCTP_MAX_BURST = 4
76SCTP_MAX_INIT_RETRANS = 8
77SCTP_RTO_ALPHA = 1 / 8
78SCTP_RTO_BETA = 1 / 4
79# A lost message nothing follows (a cursor, a clipboard, a control reply) is
80# repaired by T3 alone, so the floor is its whole cost. Chrome's dcSCTP floor:
81# it stays above a peer's delayed SACK (200 ms in dcSCTP) plus a round trip, so
82# a peer that ignores the I-bit is not sent spurious retransmissions.
83SCTP_RTO_INITIAL = 1.0
84SCTP_RTO_MIN = 0.4
85SCTP_RTO_MAX = 60
86SCTP_TSN_MODULO = 2**32
88RECONFIG_MAX_STREAMS = 135
90# parameters
91SCTP_STATE_COOKIE = 0x0007
92SCTP_STR_RESET_OUT_REQUEST = 0x000D
93SCTP_STR_RESET_RESPONSE = 0x0010
94SCTP_STR_RESET_ADD_OUT_STREAMS = 0x0011
95SCTP_SUPPORTED_CHUNK_EXT = 0x8008
96SCTP_PRSCTP_SUPPORTED = 0xC000
98# data channel constants
99DATA_CHANNEL_ACK = 2
100DATA_CHANNEL_OPEN = 3
102DATA_CHANNEL_RELIABLE = 0x00
103DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT = 0x01
104DATA_CHANNEL_PARTIAL_RELIABLE_TIMED = 0x02
105DATA_CHANNEL_RELIABLE_UNORDERED = 0x80
106DATA_CHANNEL_PARTIAL_RELIABLE_REXMIT_UNORDERED = 0x81
107DATA_CHANNEL_PARTIAL_RELIABLE_TIMED_UNORDERED = 0x82
109WEBRTC_DCEP = 50
110WEBRTC_STRING = 51
111WEBRTC_BINARY = 53
112WEBRTC_STRING_EMPTY = 56
113WEBRTC_BINARY_EMPTY = 57
115DataChannelQueue = Deque[tuple[RTCDataChannel, int, bytes]]
118def chunk_type(chunk: "Chunk") -> str:
119 return chunk.__class__.__name__
122def decode_params(body: bytes) -> list[tuple[int, bytes]]:
123 params = []
124 pos = 0
125 while pos <= len(body) - 4:
126 param_type, param_length = unpack_from("!HH", body, pos)
127 if param_length < 4:
128 raise ValueError(f"SCTP parameter has an invalid length of {param_length} bytes")
129 params.append((param_type, body[pos + 4 : pos + param_length]))
130 pos += param_length + padl(param_length)
131 return params
134def encode_params(params: list[tuple[int, bytes]]) -> bytes:
135 body = b""
136 padding = b""
137 for param_type, param_value in params:
138 param_length = len(param_value) + 4
139 body += padding
140 body += pack("!HH", param_type, param_length) + param_value
141 padding = b"\x00" * padl(param_length)
142 return body
145def padl(length: int) -> int:
146 m = length % 4
147 return 4 - m if m else 0
150def tsn_minus_one(a: int) -> int:
151 return (a - 1) % SCTP_TSN_MODULO
154def tsn_plus_one(a: int) -> int:
155 return (a + 1) % SCTP_TSN_MODULO
158class Chunk:
159 type = -1
161 def __init__(self, flags: int = 0, body: bytes = b"") -> None:
162 self.flags = flags
163 self.body = body
165 def __bytes__(self) -> bytes:
166 body = self.body
167 data = pack("!BBH", self.type, self.flags, len(body) + 4) + body
168 data += b"\x00" * padl(len(body))
169 return data
171 def __repr__(self) -> str:
172 return f"{chunk_type(self)}(flags={self.flags})"
175class BaseParamsChunk(Chunk):
176 def __init__(self, flags: int = 0, body: Optional[bytes] = None) -> None:
177 self.flags = flags
178 if body:
179 self.params = decode_params(body)
180 else:
181 self.params = []
183 @property
184 def body(self) -> bytes: # type: ignore
185 return encode_params(self.params)
188class AbortChunk(BaseParamsChunk):
189 type = 6
192class CookieAckChunk(Chunk):
193 type = 11
196class CookieEchoChunk(Chunk):
197 type = 10
200class DataChunk(Chunk):
201 type = 0
203 _acked: bool
204 _abandoned: bool
205 _book_size: int
206 _expiry: Optional[float]
207 _max_retransmits: Optional[int]
208 _misses: int
209 _retransmit: bool
210 _sent_count: int
211 _sent_time: Optional[float]
213 def __init__(self, flags: int = 0, body: Optional[bytes] = None) -> None:
214 self.flags = flags
215 if body:
216 (self.tsn, self.stream_id, self.stream_seq, self.protocol) = unpack_from(
217 "!LHHL", body
218 )
219 self.user_data = body[12:]
220 else:
221 self.tsn = 0
222 self.stream_id = 0
223 self.stream_seq = 0
224 self.protocol = 0
225 self.user_data = b""
227 def __bytes__(self) -> bytes:
228 length = 16 + len(self.user_data)
229 data = (
230 pack(
231 "!BBHLHHL",
232 self.type,
233 self.flags,
234 length,
235 self.tsn,
236 self.stream_id,
237 self.stream_seq,
238 self.protocol,
239 )
240 + self.user_data
241 )
242 if length % 4:
243 data += b"\x00" * padl(length)
244 return data
246 def __repr__(self) -> str:
247 return (
248 f"DataChunk(flags={self.flags}, tsn={self.tsn}, "
249 f"stream_id={self.stream_id}, stream_seq={self.stream_seq})"
250 )
253class ErrorChunk(BaseParamsChunk):
254 type = 9
257class ForwardTsnChunk(Chunk):
258 type = 192
260 def __init__(self, flags: int = 0, body: Optional[bytes] = None) -> None:
261 self.flags = flags
262 self.streams: list[tuple[int, int]] = []
263 if body:
264 self.cumulative_tsn = unpack_from("!L", body, 0)[0]
265 pos = 4
266 while pos < len(body):
267 self.streams.append(
268 cast(tuple[int, int], unpack_from("!HH", body, pos))
269 )
270 pos += 4
271 else:
272 self.cumulative_tsn = 0
274 @property
275 def body(self) -> bytes: # type: ignore
276 body = pack("!L", self.cumulative_tsn)
277 for stream_id, stream_seq in self.streams:
278 body += pack("!HH", stream_id, stream_seq)
279 return body
281 def __repr__(self) -> str:
282 return (
283 f"ForwardTsnChunk(cumulative_tsn={self.cumulative_tsn}, "
284 f"streams={self.streams})"
285 )
288class HeartbeatChunk(BaseParamsChunk):
289 type = 4
292class HeartbeatAckChunk(BaseParamsChunk):
293 type = 5
296class BaseInitChunk(Chunk):
297 def __init__(self, flags: int = 0, body: Optional[bytes] = None) -> None:
298 self.flags = flags
299 if body:
300 (
301 self.initiate_tag,
302 self.advertised_rwnd,
303 self.outbound_streams,
304 self.inbound_streams,
305 self.initial_tsn,
306 ) = unpack_from("!LLHHL", body)
307 self.params = decode_params(body[16:])
308 else:
309 self.initiate_tag = 0
310 self.advertised_rwnd = 0
311 self.outbound_streams = 0
312 self.inbound_streams = 0
313 self.initial_tsn = 0
314 self.params = []
316 @property
317 def body(self) -> bytes: # type: ignore
318 body = pack(
319 "!LLHHL",
320 self.initiate_tag,
321 self.advertised_rwnd,
322 self.outbound_streams,
323 self.inbound_streams,
324 self.initial_tsn,
325 )
326 body += encode_params(self.params)
327 return body
330class InitChunk(BaseInitChunk):
331 type = 1
334class InitAckChunk(BaseInitChunk):
335 type = 2
338class ReconfigChunk(BaseParamsChunk):
339 type = 130
342class SackChunk(Chunk):
343 type = 3
345 def __init__(self, flags: int = 0, body: Optional[bytes] = None) -> None:
346 self.flags = flags
347 self.gaps = []
348 self.duplicates = []
349 if body:
350 (
351 self.cumulative_tsn,
352 self.advertised_rwnd,
353 nb_gaps,
354 nb_duplicates,
355 ) = unpack_from("!LLHH", body)
356 pos = 12
357 for i in range(nb_gaps):
358 self.gaps.append(unpack_from("!HH", body, pos))
359 pos += 4
360 for i in range(nb_duplicates):
361 self.duplicates.append(unpack_from("!L", body, pos)[0])
362 pos += 4
363 else:
364 self.cumulative_tsn = 0
365 self.advertised_rwnd = 0
367 def __bytes__(self) -> bytes:
368 length = 16 + 4 * (len(self.gaps) + len(self.duplicates))
369 data = pack(
370 "!BBHLLHH",
371 self.type,
372 self.flags,
373 length,
374 self.cumulative_tsn,
375 self.advertised_rwnd,
376 len(self.gaps),
377 len(self.duplicates),
378 )
379 for gap in self.gaps:
380 data += pack("!HH", *gap)
381 for tsn in self.duplicates:
382 data += pack("!L", tsn)
383 return data
385 def __repr__(self) -> str:
386 return (
387 f"SackChunk(flags={self.flags}, advertised_rwnd={self.advertised_rwnd}, "
388 f"cumulative_tsn={self.cumulative_tsn}, gaps={self.gaps})"
389 )
392class ShutdownChunk(Chunk):
393 type = 7
395 def __init__(self, flags: int = 0, body: Optional[bytes] = None) -> None:
396 self.flags = flags
397 if body:
398 self.cumulative_tsn = unpack_from("!L", body)[0]
399 else:
400 self.cumulative_tsn = 0
402 @property
403 def body(self) -> bytes: # type: ignore
404 return pack("!L", self.cumulative_tsn)
406 def __repr__(self) -> str:
407 return (
408 f"ShutdownChunk(flags={self.flags}, cumulative_tsn={self.cumulative_tsn})"
409 )
412class ShutdownAckChunk(Chunk):
413 type = 8
416class ShutdownCompleteChunk(Chunk):
417 type = 14
420CHUNK_CLASSES = [
421 DataChunk,
422 InitChunk,
423 InitAckChunk,
424 SackChunk,
425 HeartbeatChunk,
426 HeartbeatAckChunk,
427 AbortChunk,
428 ShutdownChunk,
429 ShutdownAckChunk,
430 ErrorChunk,
431 CookieEchoChunk,
432 CookieAckChunk,
433 ShutdownCompleteChunk,
434 ReconfigChunk,
435 ForwardTsnChunk,
436]
437CHUNK_TYPES = dict((cls.type, cls) for cls in CHUNK_CLASSES)
440# SCTP packet and chunk header sizes.
441SCTP_COMMON_HEADER_LENGTH = 12
442SCTP_CHUNK_HEADER_LENGTH = 4
443# An SCTP packet must contain the common header plus at least one chunk header.
444SCTP_PACKET_MINIMUM_LENGTH = SCTP_COMMON_HEADER_LENGTH + SCTP_CHUNK_HEADER_LENGTH
447def parse_packet(data: bytes) -> tuple[int, int, int, list[Chunk]]:
448 length = len(data)
450 # Check the packet is long enough for the common header and one chunk header.
451 if length < SCTP_PACKET_MINIMUM_LENGTH:
452 raise ValueError(
453 f"SCTP packet length is less than {SCTP_PACKET_MINIMUM_LENGTH} bytes"
454 )
456 source_port, destination_port, verification_tag = unpack_from("!HHL", data)
458 # verify checksum
459 checksum = unpack_from("<L", data, 8)[0]
460 if checksum != crc32c(data[0:8] + b"\x00\x00\x00\x00" + data[12:]):
461 raise ValueError("SCTP packet has invalid checksum")
463 chunks: list[Chunk] = []
464 pos = SCTP_COMMON_HEADER_LENGTH
465 while pos <= length - SCTP_CHUNK_HEADER_LENGTH:
466 chunk_type, chunk_flags, chunk_length = unpack_from("!BBH", data, pos)
468 # Validate the chunk length: at least a chunk header, and within the packet.
469 if chunk_length < SCTP_CHUNK_HEADER_LENGTH or pos + chunk_length > length:
470 raise ValueError(
471 f"SCTP chunk has an invalid length of {chunk_length} bytes"
472 )
473 chunk_body = data[pos + SCTP_CHUNK_HEADER_LENGTH : pos + chunk_length]
474 chunk_cls = CHUNK_TYPES.get(chunk_type)
475 if chunk_cls:
476 chunks.append(chunk_cls(flags=chunk_flags, body=chunk_body))
477 pos += chunk_length + padl(chunk_length)
478 return source_port, destination_port, verification_tag, chunks
481def serialize_packet(
482 source_port: int, destination_port: int, verification_tag: int, chunk: Chunk
483) -> bytes:
484 header = pack("!HHL", source_port, destination_port, verification_tag)
485 data = bytes(chunk)
486 checksum = crc32c(header + b"\x00\x00\x00\x00" + data)
487 return header + pack("<L", checksum) + data
490# RFC 6525
493@dataclass
494class StreamResetOutgoingParam:
495 request_sequence: int
496 response_sequence: int
497 last_tsn: int
498 streams: list[int] = field(default_factory=list)
500 def __bytes__(self) -> bytes:
501 data = pack(
502 "!LLL", self.request_sequence, self.response_sequence, self.last_tsn
503 )
504 for stream in self.streams:
505 data += pack("!H", stream)
506 return data
508 @classmethod
509 def parse(cls, data: bytes) -> "StreamResetOutgoingParam":
510 request_sequence, response_sequence, last_tsn = unpack_from("!LLL", data)
511 streams = []
512 for pos in range(12, len(data), 2):
513 streams.append(unpack_from("!H", data, pos)[0])
514 return cls(
515 request_sequence=request_sequence,
516 response_sequence=response_sequence,
517 last_tsn=last_tsn,
518 streams=streams,
519 )
522@dataclass
523class StreamAddOutgoingParam:
524 request_sequence: int
525 new_streams: int
527 def __bytes__(self) -> bytes:
528 data = pack("!LHH", self.request_sequence, self.new_streams, 0)
529 return data
531 @classmethod
532 def parse(cls, data: bytes) -> "StreamAddOutgoingParam":
533 request_sequence, new_streams, reserved = unpack_from("!LHH", data)
534 return cls(request_sequence=request_sequence, new_streams=new_streams)
537@dataclass
538class StreamResetResponseParam:
539 response_sequence: int
540 result: int
542 def __bytes__(self) -> bytes:
543 return pack("!LL", self.response_sequence, self.result)
545 @classmethod
546 def parse(cls, data: bytes) -> "StreamResetResponseParam":
547 response_sequence, result = unpack_from("!LL", data)
548 return cls(response_sequence=response_sequence, result=result)
551RECONFIG_PARAM_TYPES: dict[
552 int,
553 Union[
554 type[StreamResetOutgoingParam],
555 type[StreamResetResponseParam],
556 type[StreamAddOutgoingParam],
557 ],
558] = {
559 13: StreamResetOutgoingParam,
560 16: StreamResetResponseParam,
561 17: StreamAddOutgoingParam,
562}
565class InboundStream:
566 def __init__(self) -> None:
567 self.reassembly: list[DataChunk] = []
568 self.sequence_number = 0
569 # Bytes currently buffered for reassembly; the transport enforces a
570 # ceiling on this so a peer that ignores flow control cannot grow the
571 # buffer without bound (receive-side max-message-size enforcement).
572 self.buffered_bytes = 0
573 # Counts of FIRST_FRAG and LAST_FRAG chunks currently buffered. A message can
574 # only become deliverable once BOTH its first and final fragments are present,
575 # so pop_messages skips its whole-buffer scan while either count is zero.
576 # Tracking both (not just LAST) is what bounds the unordered path: a large
577 # message whose LAST arrives before its earlier fragments would otherwise make
578 # every subsequent fragment re-walk the growing buffer looking for a start that
579 # is not there yet (O(n^2)); with first_frag_count == 0 those scans are skipped.
580 self.first_frag_count = 0
581 self.last_frag_count = 0
583 def add_chunk(self, chunk: DataChunk) -> None:
584 self.buffered_bytes += len(chunk.user_data)
585 if chunk.flags & SCTP_DATA_FIRST_FRAG:
586 self.first_frag_count += 1
587 if chunk.flags & SCTP_DATA_LAST_FRAG:
588 self.last_frag_count += 1
589 if not self.reassembly or uint32_gt(chunk.tsn, self.reassembly[-1].tsn):
590 self.reassembly.append(chunk)
591 return
593 # Binary search for the insertion point instead of a linear scan, so a
594 # burst of reordered fragments reassembles in O(n log n) rather than
595 # O(n^2). The buffered window is bounded well under 2^31 by the per-stream
596 # reassembly cap, so modular TSN order is a total order over it and the
597 # `uint32_gt(chunk.tsn, reassembly[mid].tsn)` predicate is monotonic across
598 # the (already TSN-sorted) list — the exact precondition the linear insert
599 # relied on, so this is equivalent, just faster.
600 lo, hi = 0, len(self.reassembly)
601 while lo < hi:
602 mid = (lo + hi) // 2
603 if uint32_gt(chunk.tsn, self.reassembly[mid].tsn):
604 lo = mid + 1
605 else:
606 hi = mid
607 # should never happen, the chunk should have been eliminated
608 # as a duplicate when _mark_received() is called
609 assert (
610 lo >= len(self.reassembly) or self.reassembly[lo].tsn != chunk.tsn
611 ), "duplicate chunk in reassembly"
612 self.reassembly.insert(lo, chunk)
614 def pop_messages(self) -> Iterator[tuple[int, int, bytes]]:
615 # A complete message needs BOTH a first and a last fragment buffered; if
616 # either is absent nothing can be delivered, so skip the whole-buffer scan.
617 # This is what keeps reassembly O(n) not O(n^2) — in particular for the
618 # unordered path, where an early-arriving LAST would otherwise trigger a
619 # full re-walk on every subsequent fragment until the FIRST shows up.
620 if self.first_frag_count == 0 or self.last_frag_count == 0:
621 return
622 pos = 0
623 start_pos = None
624 expected_tsn = None
625 ordered = True
626 consumed_tsns = set()
628 while pos < len(self.reassembly):
629 chunk = self.reassembly[pos]
630 if chunk.tsn in consumed_tsns:
631 pos += 1
632 continue
633 if start_pos is None:
634 ordered = not (chunk.flags & SCTP_DATA_UNORDERED)
635 if not (chunk.flags & SCTP_DATA_FIRST_FRAG):
636 if ordered:
637 # Head-of-line fragment whose FIRST is still in flight:
638 # WAIT for the retransmit. Skipping here permanently
639 # loses ordered messages under burst reordering.
640 break
641 pos += 1
642 continue
643 if ordered and uint16_gt(chunk.stream_seq, self.sequence_number):
644 # A future ordered message; its predecessor must deliver first.
645 break
646 expected_tsn = chunk.tsn
647 start_pos = pos
648 else:
649 if chunk.tsn != expected_tsn:
650 # Gap inside the message being reassembled.
651 if ordered:
652 # The missing TSN is guaranteed to arrive; wait.
653 break
654 # Unordered: rescan this position as a fresh head.
655 start_pos = None
656 continue
658 if chunk.flags & SCTP_DATA_LAST_FRAG:
659 message_chunks = []
660 for i in range(start_pos, pos + 1):
661 c = self.reassembly[i]
662 message_chunks.append(c.user_data)
663 consumed_tsns.add(c.tsn)
664 user_data = b"".join(message_chunks)
665 if ordered and chunk.stream_seq == self.sequence_number:
666 self.sequence_number = uint16_add(self.sequence_number, 1)
667 # Each delivered message consumes exactly one FIRST (at start_pos,
668 # which is only ever set on a FIRST_FRAG chunk) and one LAST (here).
669 self.first_frag_count -= 1
670 self.last_frag_count -= 1
671 yield (chunk.stream_id, chunk.protocol, user_data)
672 start_pos = None
673 pos += 1
674 if start_pos is not None:
675 expected_tsn = tsn_plus_one(expected_tsn)
677 if consumed_tsns:
678 kept = []
679 for c in self.reassembly:
680 if c.tsn in consumed_tsns:
681 self.buffered_bytes -= len(c.user_data)
682 else:
683 kept.append(c)
684 self.reassembly = kept
686 def prune_chunks(self, tsn: int) -> int:
687 """
688 Prune chunks up to the given TSN.
689 """
690 pos = -1
691 size = 0
692 for i, chunk in enumerate(self.reassembly):
693 # Not ahead is old, as in is_obsolete: exactly 2^31 away is neither.
694 if not uint32_gt(chunk.tsn, tsn):
695 pos = i
696 size += len(chunk.user_data)
697 if chunk.flags & SCTP_DATA_FIRST_FRAG:
698 self.first_frag_count -= 1
699 if chunk.flags & SCTP_DATA_LAST_FRAG:
700 self.last_frag_count -= 1
701 else:
702 break
704 self.reassembly = self.reassembly[pos + 1 :]
705 self.buffered_bytes -= size
706 return size
709@dataclass
710class RTCSctpCapabilities:
711 """
712 The :class:`RTCSctpCapabilities` dictionary provides information about the
713 capabilities of the :class:`RTCSctpTransport`.
714 """
716 maxMessageSize: int
717 """
718 The maximum size of data that the implementation can send or
719 0 if the implementation can handle messages of any size.
721 """
724class RTCSctpTransport(AsyncIOEventEmitter):
725 """
726 The :class:`RTCSctpTransport` interface includes information relating to
727 Stream Control Transmission Protocol (SCTP) transport.
729 :param transport: An :class:`RTCDtlsTransport`.
730 """
732 def __init__(self, transport: RTCDtlsTransport, port: int = 5000) -> None:
733 if transport.state == "closed":
734 raise InvalidStateError
736 super().__init__()
737 # 256 KiB when the peer doesn't advertise a=max-message-size: Chromium
738 # answers legacy DTLS/SCTP m-lines without the attribute yet receives
739 # up to 256 KiB, and Firefox/WebKit accept more. RFC 8841's 64 KiB
740 # fallback would needlessly cap clipboard/file messages we send.
741 self.remote_max_message_size = 262144
742 # Receive-side ceiling on what the inbound streams buffer together: a
743 # memory-safety backstop, not per-message policing. Under burst loss an ORDERED stream legitimately
744 # buffers several completed messages behind one head-of-line gap while
745 # retransmits fill it, so the bound must be generous; only a peer that
746 # ignores both our advertised limit and flow control can reach it.
747 self._max_reassembly_bytes = 16 * 1024 * 1024
748 self._reassembly_bytes = 0
749 self._association_state = self.State.CLOSED
750 self.__log_debug: Callable[..., None] = lambda *args: None
751 self.__started = False
752 self.__state = "new"
753 self.__transport = transport
755 self._loop = asyncio.get_running_loop()
756 self._hmac_key = os.urandom(16)
758 self._local_partial_reliability = True
759 self._local_port = port
760 self._local_verification_tag = random32()
762 self._remote_extensions: list[int] = []
763 self._remote_partial_reliability = False
764 self._remote_port: Optional[int] = None
765 self._remote_verification_tag = 0
767 # inbound
768 # 4 MiB receive window: browser->server bulk transfers (uploads) are
769 # window-capped at rwnd/RTT once past slow start, so a small window
770 # (e.g. 1 MiB) throttles any high-RTT path (TURN relays especially) to
771 # a crawl. The cost is a larger worst-case reassembly buffer per
772 # association, bounded by the same flow control.
773 self._advertised_rwnd = 4 * 1024 * 1024
774 self._inbound_streams: dict[int, InboundStream] = {}
775 self._inbound_streams_count = 0
776 self._inbound_streams_max = MAX_STREAMS
777 self._last_received_tsn: Optional[int] = None
778 self._sack_duplicates: list[int] = []
779 self._sack_misordered: set[int] = set()
780 self._sack_needed = False
781 # Delayed-SACK (RFC 4960 §6.2): acknowledge every second in-order packet
782 # rather than every one, halving SACK output on a bulk receive (the pure-
783 # Python receive path is CPU-bound). A short timer flushes a lone pending
784 # SACK so a final or cwnd-limited packet is never left unacknowledged, and
785 # a gap always acks immediately so the peer can fast-retransmit.
786 self._sack_delay = 0.020
787 self._sack_delay_handle: Optional[asyncio.TimerHandle] = None
788 self._sack_packets_since = 0
789 # A received DATA chunk carried the RFC 7053 I-bit: acknowledge at once
790 # (the peer is cwnd-limited and waiting on this SACK to send more).
791 self._sack_immediate = False
793 # outbound
794 # True while at least one sent chunk is flagged for retransmit, so
795 # _transmit can skip its scan of the whole in-flight queue on the common
796 # lossless path (that scan is per-SACK, i.e. O(in-flight) every call).
797 self._retransmit_pending = False
798 self._cwnd = 3 * USERDATA_MAX_LENGTH
799 self._fast_recovery_exit = None
800 self._fast_recovery_transmit = False
801 self._forward_tsn_chunk: Optional[ForwardTsnChunk] = None
802 self._flight_size = 0
803 self._local_tsn = random32()
804 self._last_sacked_tsn = tsn_minus_one(self._local_tsn)
805 self._advanced_peer_ack_tsn = tsn_minus_one(self._local_tsn)
806 self._outbound_queue: Deque[DataChunk] = deque()
807 self._outbound_stream_seq: dict[int, int] = {}
808 self._outbound_streams_count = MAX_STREAMS
809 self._partial_bytes_acked = 0
810 self._sent_queue: Deque[DataChunk] = deque()
812 # reconfiguration
813 self._reconfig_queue: list[int] = []
814 self._reconfig_request: Optional[StreamResetOutgoingParam] = None
815 self._reconfig_request_seq = self._local_tsn
816 self._reconfig_response_seq = 0
818 # rtt calculation
819 self._srtt: Optional[float] = None
820 self._rttvar: Optional[float] = None
822 # timers
823 self._rto = SCTP_RTO_INITIAL
824 self._t1_chunk: Optional[Chunk] = None
825 self._t1_failures = 0
826 self._t1_handle: Optional[asyncio.TimerHandle] = None
827 self._t2_chunk: Optional[Chunk] = None
828 self._t2_failures = 0
829 self._t2_handle: Optional[asyncio.TimerHandle] = None
830 self._t3_handle: Optional[asyncio.TimerHandle] = None
832 # data channels
833 self._data_channel_id: Optional[int] = None
834 self._data_channel_queue: DataChannelQueue = deque()
835 self._data_channels: dict[int, RTCDataChannel] = {}
837 # Owned and driven by RTCPeerConnection's BUNDLE bookkeeping.
838 self._bundled = False
839 self.mid: Optional[str] = None
841 @property
842 def is_server(self) -> bool:
843 return self.transport.transport.role != "controlling"
845 @property
846 def maxChannels(self) -> Optional[int]:
847 """
848 The maximum number of :class:`RTCDataChannel` that can be used simultaneously.
849 """
850 if self._inbound_streams_count:
851 return min(self._inbound_streams_count, self._outbound_streams_count)
852 return None
854 @property
855 def port(self) -> int:
856 """
857 The local SCTP port number used for data channels.
858 """
859 return self._local_port
861 @property
862 def state(self) -> str:
863 """
864 The current state of the SCTP transport.
865 """
866 return self.__state
868 @property
869 def transport(self) -> RTCDtlsTransport:
870 """
871 The :class:`RTCDtlsTransport` over which SCTP data is transmitted.
872 """
873 return self.__transport
875 @classmethod
876 def getCapabilities(cls) -> RTCSctpCapabilities:
877 """
878 Retrieve the capabilities of the transport.
880 :rtype: RTCSctpCapabilities
881 """
882 # 256 KiB: larger DataChannel messages (clipboard/file) with less chunking;
883 # emitted as a=max-message-size and matches the browser default ceiling.
884 return RTCSctpCapabilities(maxMessageSize=262144)
886 def setTransport(self, transport: RTCDtlsTransport) -> None:
887 self.__transport = transport
889 async def start(
890 self, remoteCaps: Optional[RTCSctpCapabilities], remotePort: int
891 ) -> None:
892 """
893 Start the transport.
894 """
895 if not self.__started:
896 self.__started = True
897 self.__state = "connecting"
898 self._remote_port = remotePort
899 # The peer's a=max-message-size bounds what we may send in one
900 # message; absent from the SDP (None caps — Chromium answers
901 # legacy DTLS/SCTP m-lines without it), keep the 256 KiB default
902 # every supported browser accepts.
903 if remoteCaps is not None and remoteCaps.maxMessageSize > 0:
904 self.remote_max_message_size = remoteCaps.maxMessageSize
906 # configure logging
907 if logger.isEnabledFor(logging.DEBUG):
908 prefix = "RTCSctpTransport(%s) " % (
909 self.is_server and "server" or "client"
910 )
911 self.__log_debug = lambda msg, *args: logger.debug(prefix + msg, *args)
913 # initialise local channel ID counter
914 # one side should be using even IDs, the other odd IDs
915 if self.is_server:
916 self._data_channel_id = 0
917 else:
918 self._data_channel_id = 1
920 self.__transport._register_data_receiver(self)
921 if not self.is_server:
922 await self._init()
924 async def stop(self) -> None:
925 """
926 Stop the transport.
927 """
928 if self._association_state != self.State.CLOSED:
929 await self._abort()
930 self.__transport._unregister_data_receiver(self)
931 self._set_state(self.State.CLOSED)
933 async def _abort(self) -> None:
934 """
935 Abort the association.
936 """
937 chunk = AbortChunk()
938 try:
939 await self._send_chunk(chunk)
940 except ConnectionError:
941 pass
943 async def _init(self) -> None:
944 """
945 Initialize the association.
946 """
947 chunk = InitChunk()
948 chunk.initiate_tag = self._local_verification_tag
949 chunk.advertised_rwnd = self._advertised_rwnd
950 chunk.outbound_streams = self._outbound_streams_count
951 chunk.inbound_streams = self._inbound_streams_max
952 chunk.initial_tsn = self._local_tsn
953 self._set_extensions(chunk.params)
954 await self._send_chunk(chunk)
956 # start T1 timer and enter COOKIE-WAIT state
957 self._t1_start(chunk)
958 self._set_state(self.State.COOKIE_WAIT)
960 def _flight_size_decrease(self, chunk: DataChunk) -> None:
961 self._flight_size = max(0, self._flight_size - chunk._book_size)
963 def _flight_size_increase(self, chunk: DataChunk) -> None:
964 self._flight_size += chunk._book_size
966 def _get_extensions(self, params: list[tuple[int, bytes]]) -> None:
967 """
968 Gets what extensions are supported by the remote party.
969 """
970 for k, v in params:
971 if k == SCTP_PRSCTP_SUPPORTED:
972 self._remote_partial_reliability = True
973 elif k == SCTP_SUPPORTED_CHUNK_EXT:
974 self._remote_extensions = list(v)
976 def _set_extensions(self, params: list[tuple[int, bytes]]) -> None:
977 """
978 Sets what extensions are supported by the local party.
979 """
980 extensions = []
981 if self._local_partial_reliability:
982 params.append((SCTP_PRSCTP_SUPPORTED, b""))
983 extensions.append(ForwardTsnChunk.type)
985 extensions.append(ReconfigChunk.type)
986 params.append((SCTP_SUPPORTED_CHUNK_EXT, bytes(extensions)))
988 def _inbound_stream_open(self, stream_id: int) -> bool:
989 """Whether the peer's INIT opened this stream id (any, before it is known)."""
990 return not self._inbound_streams_count or stream_id < self._inbound_streams_count
992 def _get_inbound_stream(self, stream_id: int) -> InboundStream:
993 """
994 Get or create the inbound stream with the specified ID.
995 """
996 if stream_id not in self._inbound_streams:
997 self._inbound_streams[stream_id] = InboundStream()
998 return self._inbound_streams[stream_id]
1000 def _get_timestamp(self) -> int:
1001 return int(time.time())
1003 async def _handle_data(self, data: bytes) -> None:
1004 """
1005 Handle data received from the network.
1006 """
1007 try:
1008 _, _, verification_tag, chunks = parse_packet(data)
1009 except (ValueError, StructError):
1010 return
1012 # is this an init?
1013 init_chunk = len([x for x in chunks if isinstance(x, InitChunk)])
1014 if init_chunk:
1015 # An INIT must be the only chunk in the packet (RFC 4960 3.2.2). A
1016 # peer that bundles more is a protocol violation; drop the packet
1017 # rather than raising (an AssertionError would kill the association,
1018 # and asserts vanish under `python -O`).
1019 if len(chunks) != 1:
1020 self.__log_debug("Dropping packet: INIT bundled with other chunks")
1021 return
1022 expected_tag = 0
1023 else:
1024 expected_tag = self._local_verification_tag
1026 # verify tag
1027 if verification_tag != expected_tag:
1028 self.__log_debug(
1029 "Bad verification tag %d vs %d", verification_tag, expected_tag
1030 )
1031 return
1033 # handle chunks
1034 for chunk in chunks:
1035 await self._receive_chunk(chunk)
1037 # SACK immediately on a gap/duplicate (so the peer can fast-retransmit)
1038 # or on every second in-order packet; otherwise arm a short timer to
1039 # flush the deferred acknowledgement.
1040 if self._sack_needed:
1041 if (self._sack_immediate or self._sack_misordered or self._sack_duplicates
1042 or self._sack_packets_since >= 1):
1043 await self._send_sack()
1044 else:
1045 self._sack_packets_since += 1
1046 if self._sack_delay_handle is None:
1047 self._sack_delay_handle = self._loop.call_later(
1048 self._sack_delay, self._sack_delay_expired
1049 )
1051 def _maybe_abandon(self, chunk: DataChunk) -> bool:
1052 """
1053 Determine if a chunk needs to be marked as abandoned.
1055 If it does, it marks the chunk and any other chunk belong to the same
1056 message as abandoned.
1057 """
1058 if chunk._abandoned:
1059 return True
1061 abandon = (
1062 chunk._max_retransmits is not None
1063 and chunk._sent_count > chunk._max_retransmits
1064 ) or (chunk._expiry is not None and chunk._expiry < time.time())
1065 if not abandon:
1066 return False
1068 chunk_pos = self._sent_queue.index(chunk)
1069 for pos in range(chunk_pos, -1, -1):
1070 ochunk = self._sent_queue[pos]
1071 ochunk._abandoned = True
1072 ochunk._retransmit = False
1073 if ochunk.flags & SCTP_DATA_FIRST_FRAG:
1074 break
1075 for pos in range(chunk_pos, len(self._sent_queue)):
1076 ochunk = self._sent_queue[pos]
1077 ochunk._abandoned = True
1078 ochunk._retransmit = False
1079 if ochunk.flags & SCTP_DATA_LAST_FRAG:
1080 break
1082 return True
1084 def _mark_received(self, tsn: int) -> bool:
1085 """
1086 Mark an incoming data TSN as received.
1087 """
1088 # it's a duplicate unless ahead, as in is_obsolete: exactly 2^31 away is neither
1089 if not uint32_gt(tsn, self._last_received_tsn) or tsn in self._sack_misordered:
1090 self._sack_duplicates.append(tsn)
1091 return True
1093 # consolidate misordered entries
1094 self._sack_misordered.add(tsn)
1095 # Order by SCTP modular (uint32) distance from the cumulative TSN, not raw
1096 # integer value: at the 2^32 wrap raw sorting puts the low post-wrap TSNs
1097 # before 0xFFFFFFFF, so the contiguous run is tested out of order and the
1098 # cumulative ack wedges at the boundary.
1099 base = self._last_received_tsn
1100 for tsn in sorted(self._sack_misordered, key=lambda t: (t - base) % SCTP_TSN_MODULO):
1101 if tsn == tsn_plus_one(self._last_received_tsn):
1102 self._last_received_tsn = tsn
1103 else:
1104 break
1106 # filter out obsolete entries
1107 def is_obsolete(x: int) -> bool:
1108 return uint32_gt(x, self._last_received_tsn)
1110 self._sack_duplicates = list(filter(is_obsolete, self._sack_duplicates))
1111 self._sack_misordered = set(filter(is_obsolete, self._sack_misordered))
1112 return False
1114 async def _receive(self, stream_id: int, pp_id: int, data: bytes) -> None:
1115 """
1116 Receive data stream -> ULP.
1117 """
1118 await self._data_channel_receive(stream_id, pp_id, data)
1120 async def _receive_chunk(self, chunk: Chunk) -> None:
1121 """
1122 Handle an incoming chunk.
1123 """
1124 self.__log_debug("< %s", chunk)
1126 # Neither INIT nor INIT ACK has named the peer's first TSN yet.
1127 if isinstance(chunk, (DataChunk, ForwardTsnChunk)) and self._last_received_tsn is None:
1128 return
1130 # common
1131 if isinstance(chunk, DataChunk):
1132 await self._receive_data_chunk(chunk)
1133 elif isinstance(chunk, SackChunk):
1134 await self._receive_sack_chunk(chunk)
1135 elif isinstance(chunk, ForwardTsnChunk):
1136 await self._receive_forward_tsn_chunk(chunk)
1137 elif isinstance(chunk, HeartbeatChunk):
1138 heartbeat_ack = HeartbeatAckChunk()
1139 heartbeat_ack.params = chunk.params
1140 await self._send_chunk(heartbeat_ack)
1141 elif isinstance(chunk, AbortChunk):
1142 self.__log_debug("x Association was aborted by remote party")
1143 self._set_state(self.State.CLOSED)
1144 elif isinstance(chunk, ShutdownChunk):
1145 self._t2_cancel()
1146 self._set_state(self.State.SHUTDOWN_RECEIVED)
1147 shutdown_ack = ShutdownAckChunk()
1148 await self._send_chunk(shutdown_ack)
1149 self._t2_start(shutdown_ack)
1150 self._set_state(self.State.SHUTDOWN_ACK_SENT)
1151 elif (
1152 isinstance(chunk, ShutdownCompleteChunk)
1153 and self._association_state == self.State.SHUTDOWN_ACK_SENT
1154 ):
1155 self._t2_cancel()
1156 self._set_state(self.State.CLOSED)
1157 elif (
1158 isinstance(chunk, ReconfigChunk)
1159 and self._association_state == self.State.ESTABLISHED
1160 ):
1161 for param in chunk.params:
1162 cls = RECONFIG_PARAM_TYPES.get(param[0])
1163 if cls is not None:
1164 try:
1165 parsed = cls.parse(param[1])
1166 except StructError:
1167 continue
1168 await self._receive_reconfig_param(parsed)
1170 # server
1171 elif isinstance(chunk, InitChunk) and self.is_server:
1172 self._last_received_tsn = tsn_minus_one(chunk.initial_tsn)
1173 self._reconfig_response_seq = tsn_minus_one(chunk.initial_tsn)
1174 self._remote_verification_tag = chunk.initiate_tag
1175 self._ssthresh = chunk.advertised_rwnd
1176 self._get_extensions(chunk.params)
1178 self.__log_debug(
1179 "- Peer supports %d outbound streams, %d max inbound streams",
1180 chunk.outbound_streams,
1181 chunk.inbound_streams,
1182 )
1183 self._inbound_streams_count = min(
1184 chunk.outbound_streams, self._inbound_streams_max
1185 )
1186 self._outbound_streams_count = min(
1187 self._outbound_streams_count, chunk.inbound_streams
1188 )
1190 init_ack = InitAckChunk()
1191 init_ack.initiate_tag = self._local_verification_tag
1192 init_ack.advertised_rwnd = self._advertised_rwnd
1193 init_ack.outbound_streams = self._outbound_streams_count
1194 init_ack.inbound_streams = self._inbound_streams_max
1195 init_ack.initial_tsn = self._local_tsn
1196 self._set_extensions(init_ack.params)
1198 # generate state cookie
1199 cookie = pack("!L", self._get_timestamp())
1200 cookie += hmac.new(self._hmac_key, cookie, "sha1").digest()
1201 init_ack.params.append((SCTP_STATE_COOKIE, cookie))
1202 await self._send_chunk(init_ack)
1203 elif isinstance(chunk, CookieEchoChunk) and self.is_server:
1204 # check state cookie MAC
1205 cookie = chunk.body
1206 if (
1207 len(cookie) != COOKIE_LENGTH
1208 or hmac.new(self._hmac_key, cookie[0:4], "sha1").digest() != cookie[4:]
1209 ):
1210 self.__log_debug("x State cookie is invalid")
1211 return
1213 # check state cookie lifetime
1214 now = self._get_timestamp()
1215 stamp = unpack_from("!L", cookie)[0]
1216 if stamp < now - COOKIE_LIFETIME or stamp > now:
1217 self.__log_debug("x State cookie has expired")
1218 error = ErrorChunk()
1219 error.params.append((SCTP_CAUSE_STALE_COOKIE, b"\x00" * 8))
1220 await self._send_chunk(error)
1221 return
1223 cookie_ack = CookieAckChunk()
1224 await self._send_chunk(cookie_ack)
1225 self._set_state(self.State.ESTABLISHED)
1227 # client
1228 elif (
1229 isinstance(chunk, InitAckChunk)
1230 and self._association_state == self.State.COOKIE_WAIT
1231 ):
1232 # cancel T1 timer and process chunk
1233 self._t1_cancel()
1234 self._last_received_tsn = tsn_minus_one(chunk.initial_tsn)
1235 self._reconfig_response_seq = tsn_minus_one(chunk.initial_tsn)
1236 self._remote_verification_tag = chunk.initiate_tag
1237 self._ssthresh = chunk.advertised_rwnd
1238 self._get_extensions(chunk.params)
1240 self.__log_debug(
1241 "- Peer supports %d outbound streams, %d max inbound streams",
1242 chunk.outbound_streams,
1243 chunk.inbound_streams,
1244 )
1245 self._inbound_streams_count = min(
1246 chunk.outbound_streams, self._inbound_streams_max
1247 )
1248 self._outbound_streams_count = min(
1249 self._outbound_streams_count, chunk.inbound_streams
1250 )
1252 echo = CookieEchoChunk()
1253 for k, v in chunk.params:
1254 if k == SCTP_STATE_COOKIE:
1255 echo.body = v
1256 break
1257 await self._send_chunk(echo)
1259 # start T1 timer and enter COOKIE-ECHOED state
1260 self._t1_start(echo)
1261 self._set_state(self.State.COOKIE_ECHOED)
1262 elif (
1263 isinstance(chunk, CookieAckChunk)
1264 and self._association_state == self.State.COOKIE_ECHOED
1265 ):
1266 # cancel T1 timer and enter ESTABLISHED state
1267 self._t1_cancel()
1268 self._set_state(self.State.ESTABLISHED)
1269 elif isinstance(chunk, ErrorChunk) and self._association_state in [
1270 self.State.COOKIE_WAIT,
1271 self.State.COOKIE_ECHOED,
1272 ]:
1273 self._t1_cancel()
1274 self._set_state(self.State.CLOSED)
1275 self.__log_debug("x Could not establish association")
1276 return
1278 async def _receive_data_chunk(self, chunk: DataChunk) -> None:
1279 """
1280 Handle a DATA chunk.
1281 """
1282 self._sack_needed = True
1283 if chunk.flags & SCTP_DATA_SACK_IMMEDIATELY:
1284 self._sack_immediate = True
1286 # A gap ack block reaches only 65535 TSNs past the cumulative one.
1287 if (
1288 uint32_gt(chunk.tsn, self._last_received_tsn)
1289 and (chunk.tsn - self._last_received_tsn) % SCTP_TSN_MODULO > 0xFFFF
1290 ):
1291 return
1293 # mark as received
1294 if self._mark_received(chunk.tsn):
1295 return
1297 # A stream the association did not open is acknowledged and its data
1298 # discarded (RFC 4960 6.5).
1299 if not self._inbound_stream_open(chunk.stream_id):
1300 return
1302 # find stream
1303 inbound_stream = self._get_inbound_stream(chunk.stream_id)
1305 # defragment data
1306 inbound_stream.add_chunk(chunk)
1307 self._advertised_rwnd -= len(chunk.user_data)
1308 self._reassembly_bytes += len(chunk.user_data)
1309 if self._reassembly_bytes > self._max_reassembly_bytes:
1310 # The peer ignored our advertised max-message-size and flow
1311 # control, on one stream or spread over many; abort instead of
1312 # buffering without bound.
1313 logger.warning(
1314 "RTCSctpTransport aborting: reassembly exceeded %d bytes",
1315 self._max_reassembly_bytes,
1316 )
1317 await self.stop()
1318 return
1319 for message in inbound_stream.pop_messages():
1320 self._advertised_rwnd += len(message[2])
1321 self._reassembly_bytes -= len(message[2])
1322 await self._receive(*message)
1324 async def _receive_forward_tsn_chunk(self, chunk: ForwardTsnChunk) -> None:
1325 """
1326 Handle a FORWARD TSN chunk.
1327 """
1328 self._sack_needed = True
1330 # it's a duplicate unless ahead: exactly 2^31 away is neither
1331 if not uint32_gt(chunk.cumulative_tsn, self._last_received_tsn):
1332 return
1334 def is_obsolete(x: int) -> bool:
1335 return uint32_gt(x, self._last_received_tsn)
1337 # advance cumulative TSN
1338 self._last_received_tsn = chunk.cumulative_tsn
1339 self._sack_misordered = set(filter(is_obsolete, self._sack_misordered))
1340 base = self._last_received_tsn
1341 for tsn in sorted(self._sack_misordered, key=lambda t: (t - base) % SCTP_TSN_MODULO):
1342 if tsn == tsn_plus_one(self._last_received_tsn):
1343 self._last_received_tsn = tsn
1344 else:
1345 break
1347 # filter out obsolete entries
1348 self._sack_duplicates = list(filter(is_obsolete, self._sack_duplicates))
1349 self._sack_misordered = set(filter(is_obsolete, self._sack_misordered))
1351 # update reassembly
1352 for stream_id, stream_seq in chunk.streams:
1353 if not self._inbound_stream_open(stream_id):
1354 continue
1355 inbound_stream = self._get_inbound_stream(stream_id)
1357 # advance sequence number and perform delivery
1358 inbound_stream.sequence_number = uint16_add(stream_seq, 1)
1359 for message in inbound_stream.pop_messages():
1360 self._advertised_rwnd += len(message[2])
1361 self._reassembly_bytes -= len(message[2])
1362 await self._receive(*message)
1364 # prune obsolete chunks
1365 for stream_id, inbound_stream in self._inbound_streams.items():
1366 pruned = inbound_stream.prune_chunks(self._last_received_tsn)
1367 self._advertised_rwnd += pruned
1368 self._reassembly_bytes -= pruned
1370 async def _receive_sack_chunk(self, chunk: SackChunk) -> None:
1371 """
1372 Handle a SACK chunk.
1373 """
1374 if uint32_gt(self._last_sacked_tsn, chunk.cumulative_tsn):
1375 return
1377 received_time = time.time()
1378 self._last_sacked_tsn = chunk.cumulative_tsn
1379 cwnd_fully_utilized = self._flight_size >= self._cwnd
1380 done = 0
1381 done_bytes = 0
1383 # handle acknowledged data
1384 while self._sent_queue and uint32_gte(
1385 self._last_sacked_tsn, self._sent_queue[0].tsn
1386 ):
1387 schunk = self._sent_queue.popleft()
1388 done += 1
1389 if not schunk._acked:
1390 done_bytes += schunk._book_size
1391 self._flight_size_decrease(schunk)
1393 # update RTO estimate
1394 if done == 1 and schunk._sent_count == 1:
1395 self._update_rto(received_time - schunk._sent_time)
1397 # handle gap blocks
1398 loss = False
1399 if chunk.gaps:
1400 # The gap ranges, offsets from the cumulative TSN, merged: each chunk
1401 # sent is looked up in them, never each offset they span (a packet of
1402 # blocks names millions). Empty ranges leave the cumulative TSN as the
1403 # highest seen.
1404 ranges: list = []
1405 for start, end in sorted(g for g in chunk.gaps if g[0] <= g[1]):
1406 if ranges and start <= ranges[-1][1] + 1:
1407 ranges[-1][1] = max(ranges[-1][1], end)
1408 else:
1409 ranges.append([start, end])
1410 starts = [r[0] for r in ranges]
1411 highest_seen_tsn = chunk.cumulative_tsn
1412 if ranges:
1413 highest_seen_tsn = (chunk.cumulative_tsn + ranges[-1][1]) % SCTP_TSN_MODULO
1415 def gap_acked(tsn: int) -> bool:
1416 offset = (tsn - chunk.cumulative_tsn) % SCTP_TSN_MODULO
1417 i = bisect.bisect_right(starts, offset) - 1
1418 return i >= 0 and offset <= ranges[i][1]
1420 # determined Highest TSN Newly Acked (HTNA)
1421 highest_newly_acked = chunk.cumulative_tsn
1422 for schunk in self._sent_queue:
1423 if uint32_gt(schunk.tsn, highest_seen_tsn):
1424 break
1425 if not schunk._acked and gap_acked(schunk.tsn):
1426 done_bytes += schunk._book_size
1427 schunk._acked = True
1428 self._flight_size_decrease(schunk)
1429 highest_newly_acked = schunk.tsn
1431 # strike missing chunks prior to HTNA
1432 for schunk in self._sent_queue:
1433 if uint32_gt(schunk.tsn, highest_newly_acked):
1434 break
1435 if not gap_acked(schunk.tsn):
1436 schunk._misses += 1
1437 if schunk._misses == 3:
1438 schunk._misses = 0
1439 if not self._maybe_abandon(schunk):
1440 schunk._retransmit = True
1441 self._retransmit_pending = True
1443 schunk._acked = False
1444 self._flight_size_decrease(schunk)
1446 loss = True
1448 # adjust congestion window
1449 if self._fast_recovery_exit is None:
1450 if done and cwnd_fully_utilized:
1451 if self._cwnd <= self._ssthresh:
1452 # slow start
1453 self._cwnd += min(done_bytes, USERDATA_MAX_LENGTH)
1454 else:
1455 # congestion avoidance
1456 self._partial_bytes_acked += done_bytes
1457 if self._partial_bytes_acked >= self._cwnd:
1458 self._partial_bytes_acked -= self._cwnd
1459 self._cwnd += USERDATA_MAX_LENGTH
1460 if loss:
1461 self._ssthresh = max(self._cwnd // 2, 4 * USERDATA_MAX_LENGTH)
1462 self._cwnd = self._ssthresh
1463 self._partial_bytes_acked = 0
1464 self._fast_recovery_exit = self._sent_queue[-1].tsn
1465 self._fast_recovery_transmit = True
1466 elif uint32_gte(chunk.cumulative_tsn, self._fast_recovery_exit):
1467 self._fast_recovery_exit = None
1469 if not self._sent_queue:
1470 # there is no outstanding data, stop T3
1471 self._t3_cancel()
1472 elif done:
1473 # the earliest outstanding chunk was acknowledged, restart T3
1474 self._t3_restart()
1476 self._update_advanced_peer_ack_point()
1477 await self._data_channel_flush()
1478 await self._transmit()
1480 async def _receive_reconfig_param(
1481 self,
1482 param: Union[
1483 StreamAddOutgoingParam, StreamResetOutgoingParam, StreamResetResponseParam
1484 ],
1485 ) -> None:
1486 """
1487 Handle a RE-CONFIG parameter.
1488 """
1489 self.__log_debug("<< %s", param)
1491 if isinstance(param, StreamResetOutgoingParam):
1492 # mark closed inbound streams
1493 for stream_id in param.streams:
1494 closed = self._inbound_streams.pop(stream_id, None)
1495 if closed is not None:
1496 self._advertised_rwnd += closed.buffered_bytes
1497 self._reassembly_bytes -= closed.buffered_bytes
1499 # close data channel
1500 channel = self._data_channels.get(stream_id)
1501 if channel:
1502 self._data_channel_close(channel)
1504 # send response
1505 response_param = StreamResetResponseParam(
1506 response_sequence=param.request_sequence, result=1
1507 )
1508 self._reconfig_response_seq = param.request_sequence
1510 await self._send_reconfig_param(response_param)
1511 elif isinstance(param, StreamAddOutgoingParam):
1512 # increase inbound streams
1513 self._inbound_streams_count += param.new_streams
1515 # send response
1516 response_param = StreamResetResponseParam(
1517 response_sequence=param.request_sequence, result=1
1518 )
1519 self._reconfig_response_seq = param.request_sequence
1521 await self._send_reconfig_param(response_param)
1522 elif isinstance(param, StreamResetResponseParam):
1523 if (
1524 self._reconfig_request
1525 and param.response_sequence == self._reconfig_request.request_sequence
1526 ):
1527 # mark closed streams
1528 for stream_id in self._reconfig_request.streams:
1529 self._outbound_stream_seq.pop(stream_id, None)
1530 self._data_channel_closed(stream_id)
1532 self._reconfig_request = None
1533 await self._transmit_reconfig()
1535 async def _send(
1536 self,
1537 stream_id: int,
1538 pp_id: int,
1539 user_data: bytes,
1540 expiry: Optional[float] = None,
1541 max_retransmits: Optional[int] = None,
1542 ordered: bool = True,
1543 ) -> None:
1544 """
1545 Send data ULP -> stream.
1546 """
1547 if ordered:
1548 stream_seq = self._outbound_stream_seq.get(stream_id, 0)
1549 else:
1550 stream_seq = 0
1552 fragments = math.ceil(len(user_data) / USERDATA_MAX_LENGTH)
1553 pos = 0
1554 for fragment in range(0, fragments):
1555 chunk = DataChunk()
1556 chunk.flags = 0
1557 if not ordered:
1558 chunk.flags = SCTP_DATA_UNORDERED
1559 if fragment == 0:
1560 chunk.flags |= SCTP_DATA_FIRST_FRAG
1561 if fragment == fragments - 1:
1562 chunk.flags |= SCTP_DATA_LAST_FRAG
1563 chunk.tsn = self._local_tsn
1564 chunk.stream_id = stream_id
1565 chunk.stream_seq = stream_seq
1566 chunk.protocol = pp_id
1567 chunk.user_data = user_data[pos : pos + USERDATA_MAX_LENGTH]
1569 # Per-chunk bookkeeping attached dynamically (not part of the Chunk type);
1570 # initialize counters
1571 chunk._abandoned = False
1572 chunk._acked = False
1573 chunk._book_size = len(chunk.user_data)
1574 chunk._expiry = expiry
1575 chunk._max_retransmits = max_retransmits
1576 chunk._misses = 0
1577 chunk._retransmit = False
1578 chunk._sent_count = 0
1579 chunk._sent_time = None
1581 pos += USERDATA_MAX_LENGTH
1582 self._local_tsn = tsn_plus_one(self._local_tsn)
1583 self._outbound_queue.append(chunk)
1585 if ordered:
1586 self._outbound_stream_seq[stream_id] = uint16_add(stream_seq, 1)
1588 # transmit outbound data
1589 await self._transmit()
1591 async def _send_chunk(self, chunk: Chunk) -> None:
1592 """
1593 Transmit a chunk (no bundling for now).
1594 """
1595 self.__log_debug("> %s", chunk)
1596 await self.__transport._send_data(
1597 serialize_packet(
1598 self._local_port,
1599 self._remote_port,
1600 self._remote_verification_tag,
1601 chunk,
1602 )
1603 )
1605 async def _send_reconfig_param(
1606 self,
1607 param: Union[
1608 StreamAddOutgoingParam, StreamResetOutgoingParam, StreamResetResponseParam
1609 ],
1610 ) -> None:
1611 chunk = ReconfigChunk()
1612 for k, cls in RECONFIG_PARAM_TYPES.items():
1613 if isinstance(param, cls):
1614 param_type = k
1615 break
1616 else:
1617 raise ValueError(f"Unsupported reconfig parameter type: {type(param)!r}")
1618 chunk.params.append((param_type, bytes(param)))
1620 self.__log_debug(">> %s", param)
1621 await self._send_chunk(chunk)
1623 async def _send_sack(self) -> None:
1624 """
1625 Build and send a selective acknowledgement (SACK) chunk.
1626 """
1627 gaps: list[list[int]] = []
1628 gap_next = None
1629 # In modular order, so the blocks stay ascending and merged across the wrap.
1630 base = self._last_received_tsn
1631 for tsn in sorted(self._sack_misordered, key=lambda t: (t - base) % SCTP_TSN_MODULO):
1632 pos = (tsn - self._last_received_tsn) % SCTP_TSN_MODULO
1633 if tsn == gap_next:
1634 gaps[-1][1] = pos
1635 else:
1636 gaps.append([pos, pos])
1637 gap_next = tsn_plus_one(tsn)
1639 sack = SackChunk()
1640 sack.cumulative_tsn = self._last_received_tsn
1641 sack.advertised_rwnd = max(0, self._advertised_rwnd)
1642 sack.duplicates = self._sack_duplicates[:]
1643 sack.gaps = [tuple(x) for x in gaps]
1645 await self._send_chunk(sack)
1647 self._sack_duplicates.clear()
1648 self._sack_needed = False
1649 self._sack_immediate = False
1650 self._sack_packets_since = 0
1651 if self._sack_delay_handle is not None:
1652 self._sack_delay_handle.cancel()
1653 self._sack_delay_handle = None
1655 def _sack_delay_expired(self) -> None:
1656 self._sack_delay_handle = None
1657 if self._sack_needed:
1658 asyncio.ensure_future(self._send_sack())
1660 def _set_state(self, state: "RTCSctpTransport.State") -> None:
1661 """
1662 Transition the SCTP association to a new state.
1663 """
1664 if state != self._association_state:
1665 self.__log_debug("- %s -> %s", self._association_state, state)
1666 self._association_state = state
1668 if state == self.State.ESTABLISHED:
1669 self.__state = "connected"
1670 for channel in list(self._data_channels.values()):
1671 if channel.negotiated and channel.readyState != "open":
1672 channel._setReadyState("open")
1673 asyncio.ensure_future(self._data_channel_flush())
1674 elif state == self.State.CLOSED:
1675 self._t1_cancel()
1676 self._t2_cancel()
1677 self._t3_cancel()
1678 if self._sack_delay_handle is not None:
1679 self._sack_delay_handle.cancel()
1680 self._sack_delay_handle = None
1681 self.__state = "closed"
1683 # close data channels
1684 for stream_id in list(self._data_channels.keys()):
1685 self._data_channel_closed(stream_id)
1687 # no more events will be emitted, so remove all event listeners
1688 # to facilitate garbage collection.
1689 self.remove_all_listeners()
1691 # timers
1693 def _t1_cancel(self) -> None:
1694 if self._t1_handle is not None:
1695 self.__log_debug("- T1(%s) cancel", chunk_type(self._t1_chunk))
1696 self._t1_handle.cancel()
1697 self._t1_handle = None
1698 self._t1_chunk = None
1700 def _t1_expired(self) -> None:
1701 self._t1_failures += 1
1702 self._t1_handle = None
1703 self.__log_debug(
1704 "x T1(%s) expired %d", chunk_type(self._t1_chunk), self._t1_failures
1705 )
1706 if self._t1_failures > SCTP_MAX_INIT_RETRANS:
1707 self._set_state(self.State.CLOSED)
1708 else:
1709 asyncio.ensure_future(self._send_chunk(self._t1_chunk))
1710 self._t1_handle = self._loop.call_later(self._rto, self._t1_expired)
1712 def _t1_start(self, chunk: Chunk) -> None:
1713 assert self._t1_handle is None
1714 self._t1_chunk = chunk
1715 self._t1_failures = 0
1716 self.__log_debug("- T1(%s) start", chunk_type(self._t1_chunk))
1717 self._t1_handle = self._loop.call_later(self._rto, self._t1_expired)
1719 def _t2_cancel(self) -> None:
1720 if self._t2_handle is not None:
1721 self.__log_debug("- T2(%s) cancel", chunk_type(self._t2_chunk))
1722 self._t2_handle.cancel()
1723 self._t2_handle = None
1724 self._t2_chunk = None
1726 def _t2_expired(self) -> None:
1727 self._t2_failures += 1
1728 self._t2_handle = None
1729 self.__log_debug(
1730 "x T2(%s) expired %d", chunk_type(self._t2_chunk), self._t2_failures
1731 )
1732 if self._t2_failures > SCTP_MAX_ASSOCIATION_RETRANS:
1733 self._set_state(self.State.CLOSED)
1734 else:
1735 asyncio.ensure_future(self._send_chunk(self._t2_chunk))
1736 self._t2_handle = self._loop.call_later(self._rto, self._t2_expired)
1738 def _t2_start(self, chunk: ShutdownAckChunk) -> None:
1739 assert self._t2_handle is None
1740 self._t2_chunk = chunk
1741 self._t2_failures = 0
1742 self.__log_debug("- T2(%s) start", chunk_type(self._t2_chunk))
1743 self._t2_handle = self._loop.call_later(self._rto, self._t2_expired)
1745 def _t3_expired(self) -> None:
1746 self._t3_handle = None
1747 self.__log_debug("x T3 expired")
1749 # mark retransmit or abandoned chunks
1750 for chunk in self._sent_queue:
1751 if not self._maybe_abandon(chunk):
1752 chunk._retransmit = True
1753 self._retransmit_pending = True
1754 self._update_advanced_peer_ack_point()
1756 # adjust congestion window
1757 self._fast_recovery_exit = None
1758 self._flight_size = 0
1759 self._partial_bytes_acked = 0
1761 self._ssthresh = max(self._cwnd // 2, 4 * USERDATA_MAX_LENGTH)
1762 self._cwnd = USERDATA_MAX_LENGTH
1764 asyncio.ensure_future(self._transmit())
1766 def _t3_restart(self) -> None:
1767 self.__log_debug("- T3 restart")
1768 if self._t3_handle is not None:
1769 self._t3_handle.cancel()
1770 self._t3_handle = None
1771 self._t3_handle = self._loop.call_later(self._rto, self._t3_expired)
1773 def _t3_start(self) -> None:
1774 assert self._t3_handle is None
1775 self.__log_debug("- T3 start")
1776 self._t3_handle = self._loop.call_later(self._rto, self._t3_expired)
1778 def _t3_cancel(self) -> None:
1779 if self._t3_handle is not None:
1780 self.__log_debug("- T3 cancel")
1781 self._t3_handle.cancel()
1782 self._t3_handle = None
1784 async def _transmit(self) -> None:
1785 """
1786 Transmit outbound data.
1787 """
1788 # send FORWARD TSN
1789 if self._forward_tsn_chunk is not None:
1790 await self._send_chunk(self._forward_tsn_chunk)
1791 self._forward_tsn_chunk = None
1793 # ensure T3 is running
1794 if not self._t3_handle:
1795 self._t3_start()
1797 # limit burst size
1798 if self._fast_recovery_exit is not None:
1799 burst_size = 2 * USERDATA_MAX_LENGTH
1800 else:
1801 burst_size = 4 * USERDATA_MAX_LENGTH
1802 cwnd = min(self._flight_size + burst_size, self._cwnd)
1804 # retransmit — skip the whole-queue scan unless a chunk is actually
1805 # flagged (no loss on the common path). An early return on a full cwnd
1806 # leaves the flag set so the remaining marked chunks are retried later.
1807 if self._retransmit_pending:
1808 retransmit_earliest = True
1809 for chunk in self._sent_queue:
1810 if chunk._retransmit:
1811 if self._fast_recovery_transmit:
1812 self._fast_recovery_transmit = False
1813 elif self._flight_size >= cwnd:
1814 return
1815 self._flight_size_increase(chunk)
1817 chunk._misses = 0
1818 chunk._retransmit = False
1819 chunk._sent_count += 1
1820 await self._send_chunk(chunk)
1821 if retransmit_earliest:
1822 # restart the T3 timer as the earliest outstanding TSN
1823 # is being retransmitted
1824 self._t3_restart()
1825 retransmit_earliest = False
1826 self._retransmit_pending = False
1828 while self._outbound_queue and self._flight_size < cwnd:
1829 chunk = self._outbound_queue.popleft()
1830 self._sent_queue.append(chunk)
1831 self._flight_size_increase(chunk)
1833 # update counters
1834 chunk._sent_count += 1
1835 chunk._sent_time = time.time()
1837 # RFC 7053: ask the receiver to acknowledge at once (I-bit) when
1838 # nothing follows this chunk until its SACK: once this send makes us
1839 # cwnd-limited, so cwnd reopens without waiting a delayed-ack
1840 # interval, and when it empties the queue, so the round trip the
1841 # lone message's timer is set from is the path's, not the peer's
1842 # ack delay.
1843 if (self._flight_size + USERDATA_MAX_LENGTH >= self._cwnd
1844 or not self._outbound_queue):
1845 chunk.flags |= SCTP_DATA_SACK_IMMEDIATELY
1847 await self._send_chunk(chunk)
1848 if not self._t3_handle:
1849 self._t3_start()
1851 async def _transmit_reconfig(self) -> None:
1852 if (
1853 self._association_state == self.State.ESTABLISHED
1854 and self._reconfig_queue
1855 and not self._reconfig_request
1856 ):
1857 streams = self._reconfig_queue[0:RECONFIG_MAX_STREAMS]
1858 self._reconfig_queue = self._reconfig_queue[RECONFIG_MAX_STREAMS:]
1859 param = StreamResetOutgoingParam(
1860 request_sequence=self._reconfig_request_seq,
1861 response_sequence=self._reconfig_response_seq,
1862 last_tsn=tsn_minus_one(self._local_tsn),
1863 streams=streams,
1864 )
1865 self._reconfig_request = param
1866 self._reconfig_request_seq = tsn_plus_one(self._reconfig_request_seq)
1868 await self._send_reconfig_param(param)
1870 def _update_advanced_peer_ack_point(self) -> None:
1871 """
1872 Try to advance "Advanced.Peer.Ack.Point" according to RFC 3758.
1873 """
1874 if uint32_gt(self._last_sacked_tsn, self._advanced_peer_ack_tsn):
1875 self._advanced_peer_ack_tsn = self._last_sacked_tsn
1877 done = 0
1878 streams = {}
1879 while self._sent_queue and self._sent_queue[0]._abandoned:
1880 chunk = self._sent_queue.popleft()
1881 self._advanced_peer_ack_tsn = chunk.tsn
1882 done += 1
1883 if not (chunk.flags & SCTP_DATA_UNORDERED):
1884 streams[chunk.stream_id] = chunk.stream_seq
1886 if done:
1887 # build FORWARD TSN
1888 self._forward_tsn_chunk = ForwardTsnChunk()
1889 self._forward_tsn_chunk.cumulative_tsn = self._advanced_peer_ack_tsn
1890 self._forward_tsn_chunk.streams = list(streams.items())
1892 def _update_rto(self, R: float) -> None:
1893 """
1894 Update RTO given a new roundtrip measurement R.
1895 """
1896 if self._srtt is None:
1897 self._rttvar = R / 2
1898 self._srtt = R
1899 else:
1900 self._rttvar = (1 - SCTP_RTO_BETA) * self._rttvar + SCTP_RTO_BETA * abs(
1901 self._srtt - R
1902 )
1903 self._srtt = (1 - SCTP_RTO_ALPHA) * self._srtt + SCTP_RTO_ALPHA * R
1904 self._rto = max(SCTP_RTO_MIN, min(self._srtt + 4 * self._rttvar, SCTP_RTO_MAX))
1906 def _data_channel_close(self, channel: RTCDataChannel) -> None:
1907 """
1908 Request closing the datachannel by sending an Outgoing Stream Reset Request.
1909 """
1910 if channel.readyState not in ["closing", "closed"]:
1911 channel._setReadyState("closing")
1913 if self._association_state == self.State.ESTABLISHED:
1914 # queue a stream reset
1915 self._reconfig_queue.append(channel.id)
1916 if len(self._reconfig_queue) == 1:
1917 asyncio.ensure_future(self._transmit_reconfig())
1918 else:
1919 # remove any queued messages for the datachannel
1920 new_queue: DataChannelQueue = deque()
1921 for queue_item in self._data_channel_queue:
1922 if queue_item[0] != channel:
1923 new_queue.append(queue_item)
1924 self._data_channel_queue = new_queue
1926 # mark the datachannel as closed
1927 if channel.id is not None:
1928 self._data_channels.pop(channel.id)
1929 channel._setReadyState("closed")
1931 def _data_channel_closed(self, stream_id: int) -> None:
1932 channel = self._data_channels.pop(stream_id)
1933 channel._setReadyState("closed")
1935 async def _data_channel_flush(self) -> None:
1936 """
1937 Try to flush buffered data to the SCTP layer.
1939 We wait until the association is established, as we need to know
1940 whether we are a client or a server to correctly assign an odd/even ID
1941 to the data channels.
1942 """
1943 if self._association_state != self.State.ESTABLISHED:
1944 return
1946 while self._data_channel_queue and not self._outbound_queue:
1947 channel, protocol, user_data = self._data_channel_queue.popleft()
1949 # register channel if necessary
1950 stream_id = channel.id
1951 if stream_id is None:
1952 stream_id = self._data_channel_id
1953 while stream_id in self._data_channels:
1954 stream_id += 2
1955 self._data_channels[stream_id] = channel
1956 channel._setId(stream_id)
1958 # send data
1959 if protocol == WEBRTC_DCEP:
1960 await self._send(stream_id, protocol, user_data)
1961 else:
1962 if channel.maxPacketLifeTime:
1963 expiry = time.time() + (channel.maxPacketLifeTime / 1000)
1964 else:
1965 expiry = None
1966 await self._send(
1967 stream_id,
1968 protocol,
1969 user_data,
1970 expiry=expiry,
1971 max_retransmits=channel.maxRetransmits,
1972 ordered=channel.ordered,
1973 )
1974 channel._addBufferedAmount(-len(user_data))
1976 def _data_channel_add_negotiated(self, channel: RTCDataChannel) -> None:
1977 if channel.id in self._data_channels:
1978 raise ValueError(f"Data channel with ID {channel.id} already registered")
1980 self._data_channels[channel.id] = channel
1982 if self._association_state == self.State.ESTABLISHED:
1983 channel._setReadyState("open")
1985 def _data_channel_open(self, channel: RTCDataChannel) -> None:
1986 if channel.id is not None:
1987 if channel.id in self._data_channels:
1988 raise ValueError(
1989 f"Data channel with ID {channel.id} already registered"
1990 )
1991 else:
1992 self._data_channels[channel.id] = channel
1994 channel_type = DATA_CHANNEL_RELIABLE
1995 priority = 0
1996 reliability = 0
1998 if not channel.ordered:
1999 channel_type |= 0x80
2000 if channel.maxRetransmits is not None:
2001 channel_type |= 1
2002 reliability = channel.maxRetransmits
2003 elif channel.maxPacketLifeTime is not None:
2004 channel_type |= 2
2005 reliability = channel.maxPacketLifeTime
2007 data = pack(
2008 "!BBHLHH",
2009 DATA_CHANNEL_OPEN,
2010 channel_type,
2011 priority,
2012 reliability,
2013 len(channel.label),
2014 len(channel.protocol),
2015 )
2016 data += channel.label.encode("utf8")
2017 data += channel.protocol.encode("utf8")
2018 self._data_channel_queue.append((channel, WEBRTC_DCEP, data))
2019 asyncio.ensure_future(self._data_channel_flush())
2021 async def _data_channel_receive(
2022 self, stream_id: int, pp_id: int, data: bytes
2023 ) -> None:
2024 if pp_id == WEBRTC_DCEP and len(data):
2025 msg_type = data[0]
2026 if msg_type == DATA_CHANNEL_OPEN and len(data) >= 12:
2027 # A DCEP OPEN reusing a live stream id is a protocol violation;
2028 # ignore it rather than raising (an AssertionError here would tear
2029 # down the whole association, and asserts vanish under `python -O`).
2030 if stream_id in self._data_channels:
2031 self.__log_debug(
2032 "Ignoring DATA_CHANNEL_OPEN for existing stream %d", stream_id
2033 )
2034 return
2036 (
2037 msg_type,
2038 channel_type,
2039 priority,
2040 reliability,
2041 label_length,
2042 protocol_length,
2043 ) = unpack_from("!BBHLHH", data)
2044 pos = 12
2045 # A peer's label or protocol need not be UTF-8 (libwebrtc takes the
2046 # bytes as they come); a decode error here would end the DTLS loop.
2047 label = data[pos : pos + label_length].decode("utf8", "replace")
2048 pos += label_length
2049 protocol = data[pos : pos + protocol_length].decode("utf8", "replace")
2051 # check channel type
2052 maxPacketLifeTime = None
2053 maxRetransmits = None
2054 if (channel_type & 0x03) == 1:
2055 maxRetransmits = reliability
2056 elif (channel_type & 0x03) == 2:
2057 maxPacketLifeTime = reliability
2059 # register channel
2060 parameters = RTCDataChannelParameters(
2061 label=label,
2062 ordered=(channel_type & 0x80) == 0,
2063 maxPacketLifeTime=maxPacketLifeTime,
2064 maxRetransmits=maxRetransmits,
2065 protocol=protocol,
2066 id=stream_id,
2067 )
2068 channel = RTCDataChannel(self, parameters, False)
2069 channel._setReadyState("open")
2070 self._data_channels[stream_id] = channel
2072 # send ack
2073 self._data_channel_queue.append(
2074 (channel, WEBRTC_DCEP, pack("!B", DATA_CHANNEL_ACK))
2075 )
2076 await self._data_channel_flush()
2078 # emit channel
2079 self.emit("datachannel", channel)
2080 elif msg_type == DATA_CHANNEL_ACK:
2081 # An ACK for an unknown stream is a protocol violation; ignore it
2082 # rather than raising (see the OPEN handler above re: -O / teardown).
2083 if stream_id not in self._data_channels:
2084 self.__log_debug(
2085 "Ignoring DATA_CHANNEL_ACK for unknown stream %d", stream_id
2086 )
2087 return
2088 channel = self._data_channels[stream_id]
2089 channel._setReadyState("open")
2090 elif pp_id == WEBRTC_STRING and stream_id in self._data_channels:
2091 # emit message
2092 self._data_channels[stream_id].emit("message", data.decode("utf8", "replace"))
2093 elif pp_id == WEBRTC_STRING_EMPTY and stream_id in self._data_channels:
2094 # emit message
2095 self._data_channels[stream_id].emit("message", "")
2096 elif pp_id == WEBRTC_BINARY and stream_id in self._data_channels:
2097 # emit message
2098 self._data_channels[stream_id].emit("message", data)
2099 elif pp_id == WEBRTC_BINARY_EMPTY and stream_id in self._data_channels:
2100 # emit message
2101 self._data_channels[stream_id].emit("message", b"")
2103 def _data_channel_send(
2104 self, channel: RTCDataChannel, data: Union[bytes, str]
2105 ) -> None:
2106 if data == "":
2107 pp_id, user_data = WEBRTC_STRING_EMPTY, b"\x00"
2108 elif isinstance(data, str):
2109 pp_id, user_data = WEBRTC_STRING, data.encode("utf8")
2110 elif data == b"":
2111 pp_id, user_data = WEBRTC_BINARY_EMPTY, b"\x00"
2112 else:
2113 pp_id, user_data = WEBRTC_BINARY, data
2115 # Mirror browser RTCDataChannel.send(): refuse messages the peer
2116 # declared it cannot receive instead of letting it kill the channel.
2117 if len(user_data) > self.remote_max_message_size:
2118 raise ValueError(
2119 f"Cannot send {len(user_data)} bytes: exceeds the peer's "
2120 f"maximum message size of {self.remote_max_message_size} bytes"
2121 )
2123 channel._addBufferedAmount(len(user_data))
2124 self._data_channel_queue.append((channel, pp_id, user_data))
2125 asyncio.ensure_future(self._data_channel_flush())
2127 class State(enum.Enum):
2128 CLOSED = 1
2129 COOKIE_WAIT = 2
2130 COOKIE_ECHOED = 3
2131 ESTABLISHED = 4
2132 SHUTDOWN_PENDING = 5
2133 SHUTDOWN_SENT = 6
2134 SHUTDOWN_RECEIVED = 7
2135 SHUTDOWN_ACK_SENT = 8
2138## 1. not sending heartBeat chunk if we are data sender