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 binascii
36import datetime
37import enum
38import logging
39import os
40import statistics
41import struct
42import time
43import traceback
44from collections import deque
45from dataclasses import dataclass, field
46from typing import Any, Callable, Dict, Optional, Protocol, Type, TypeVar, Union
48import pylibsrtp
49from cryptography import x509
50from cryptography.hazmat.backends import default_backend
51from cryptography.hazmat.primitives import hashes
52from cryptography.hazmat.primitives.asymmetric import ec
53from OpenSSL import SSL
54from pyee.asyncio import AsyncIOEventEmitter
55from pylibsrtp import Policy, Session
57from . import clock, rtp
58from .pacer import (
59 BRAKE_LOSS_FRACTION,
60 BRAKE_LOSS_RUNS,
61 CLASS_DC,
62 CLASS_RTCP,
63 CLASS_VIDEO,
64 FRAME_ENDS_MAX,
65 MIN_GOODPUT_SAMPLE_BYTES,
66 RtpPacer,
67)
68from .rtcicetransport import RTCIceTransport
69from .rtcrtpparameters import RTCRtpReceiveParameters, RTCRtpSendParameters
70from .rtp import (
71 AnyRtcpPacket,
72 RtcpByePacket,
73 RtcpPacket,
74 RtcpPsfbPacket,
75 RtcpRrPacket,
76 RtcpRtpfbPacket,
77 pack_twcc_fci,
78 RtcpSrPacket,
79 RtcpXrPacket,
80 RtpPacket,
81 is_rtcp,
82)
83from .stats import RTCStatsReport, RTCTransportStats
85CERTIFICATE_T = TypeVar("CERTIFICATE_T", bound="RTCCertificate")
86K = TypeVar("K")
87V = TypeVar("V")
89# A gap between two arrivals this long is silence, not delivery.
90TWCC_IDLE_US = 250_000
91# How long a sent packet waits in the transport-cc history for its feedback.
92TWCC_HISTORY_S = 2.0
93# How long the transport-wide sequence numbers transport-cc reported received are kept
94# (`twcc_arrived`): a sender reads from them which frames its peer holds, and a peer on a
95# narrow link with a standing queue reports a packet seconds after it left.
96TWCC_ARRIVED_S = 10.0
97# How far back the least one-way delay is the path's own: short enough that
98# the drift between the two ends' clocks stays a few milliseconds inside it.
99TWCC_DELAY_FLOOR_S = 30.0
100# How fast the path's own delay is let rise by the clock alone: the two ends'
101# clocks drift apart by tens of microseconds a second, a few hundred where one
102# is being slewed, and a queue stands tens of milliseconds past it at once.
103TWCC_DRIFT_S_PER_S = 0.00025
104# What tells a longer path from a queue once one-way delay has stood past the
105# path's own: intervals in a row that lost nothing and sent under this share
106# of what the path delivered while the delay stood, with the least delay of
107# each within the band of the others. A queue the stream filled empties by a
108# tenth of a second every second at such a rate; delay that holds is the path's.
109TWCC_PATH_INTERVALS = 3
110TWCC_PATH_SENT = 0.9
111TWCC_PATH_BAND_MS = 5.0
112TWCC_PATH_LOSS = 0.02
113# How far past the path's own delay one-way delay stands before congestion
114# control reads a queue on it.
115TWCC_QUEUE_MS = 25.0
116# The fewest feedback packets a queue is read over, which a still screen's
117# trickle takes a second or more to send; twice as many read one over the
118# newest half of an interval.
119TWCC_STAND_FEEDBACKS = 4
120# Packets sent within this of the one before belong to one burst, and a burst
121# of fewer packets measures nothing.
122TWCC_BURST_GAP_S = 0.001
123TWCC_BURST_MIN_PACKETS = 4
124# The resolution of transport-cc arrival times.
125TWCC_TICK_US = 250.0
127logger = logging.getLogger(__name__)
129# Mapping of supported `RTCDtlsFingerprint` algorithms to the
130# corresponding argument for `x509.Certificate.fingerprint`.
131X509_DIGEST_ALGORITHMS = {
132 "sha-256": hashes.SHA256(),
133 "sha-384": hashes.SHA384(),
134 "sha-512": hashes.SHA512(),
135}
138def burst_delivery_bps(packets: list, probe_bytes: float) -> Optional[float]:
139 """The median rate the wire delivered bursts' leading bytes at.
141 A frame's packets are sent back to back, and the pacer lets the first
142 `probe_bytes` of one out at once whenever it holds nothing back, so their
143 arrivals are spread by the slowest hop alone: the bytes after the first
144 over the time from the first arrival to the last is that hop's rate. The
145 rest of a burst may leave at the pace and would measure the pace instead.
146 Arrivals sit on the feedback's grid, so a burst arriving within one step
147 counts as arriving over one.
149 Args:
150 packets: `(send instant in seconds, arrival in microseconds, size)`
151 per packet a feedback acknowledged, in send order.
152 probe_bytes: How many leading bytes of a burst leave unpaced.
154 Returns:
155 The median of the bursts' rates in bits per second, or None when no
156 burst had `TWCC_BURST_MIN_PACKETS` packets.
157 """
158 rates = []
159 count = 0
160 last_sent = first_size = size_sum = lo = hi = 0.0
161 for sent_at, at_us, size in packets:
162 if not count or sent_at - last_sent > TWCC_BURST_GAP_S:
163 if count >= TWCC_BURST_MIN_PACKETS:
164 rates.append((size_sum - first_size) * 8e6 / max(hi - lo, TWCC_TICK_US))
165 count, first_size, size_sum, lo, hi = 1, size, size, at_us, at_us
166 elif size_sum + size <= probe_bytes:
167 count += 1
168 size_sum += size
169 lo, hi = min(lo, at_us), max(hi, at_us)
170 last_sent = sent_at
171 if count >= TWCC_BURST_MIN_PACKETS:
172 rates.append((size_sum - first_size) * 8e6 / max(hi - lo, TWCC_TICK_US))
173 return statistics.median(rates) if rates else None
176@dataclass(frozen=True)
177class SRTPProtectionProfile:
178 libsrtp_profile: int
179 openssl_profile: bytes
180 key_length: int
181 salt_length: int
183 def get_key_and_salt(self, src: bytes, idx: int) -> bytes:
184 key_start = idx * self.key_length
185 salt_start = 2 * self.key_length + idx * self.salt_length
186 return (
187 src[key_start : key_start + self.key_length]
188 + src[salt_start : salt_start + self.salt_length]
189 )
192SRTP_AEAD_AES_256_GCM = SRTPProtectionProfile(
193 libsrtp_profile=Policy.SRTP_PROFILE_AEAD_AES_256_GCM,
194 openssl_profile=b"SRTP_AEAD_AES_256_GCM",
195 key_length=32,
196 salt_length=12,
197)
198SRTP_AEAD_AES_128_GCM = SRTPProtectionProfile(
199 libsrtp_profile=Policy.SRTP_PROFILE_AEAD_AES_128_GCM,
200 openssl_profile=b"SRTP_AEAD_AES_128_GCM",
201 key_length=16,
202 salt_length=12,
203)
204SRTP_AES128_CM_SHA1_80 = SRTPProtectionProfile(
205 libsrtp_profile=Policy.SRTP_PROFILE_AES128_CM_SHA1_80,
206 openssl_profile=b"SRTP_AES128_CM_SHA1_80",
207 key_length=16,
208 salt_length=14,
209)
211# AES-GCM may not be available depending on how libsrtp2 was built.
212SRTP_PROFILES: list[SRTPProtectionProfile] = []
213for srtp_profile in [
214 SRTP_AEAD_AES_256_GCM,
215 SRTP_AEAD_AES_128_GCM,
216 SRTP_AES128_CM_SHA1_80,
217]:
218 try:
219 Policy(srtp_profile=srtp_profile.libsrtp_profile)
220 except pylibsrtp.Error: # pragma: no cover
221 pass
222 else:
223 SRTP_PROFILES.append(srtp_profile)
226def certificate_digest(certificate: x509.Certificate, algorithm: str) -> str:
227 hexstring = certificate.fingerprint(X509_DIGEST_ALGORITHMS[algorithm]).hex().upper()
228 return ":".join(hexstring[x : x + 2] for x in range(0, len(hexstring), 2))
231def generate_certificate(key: ec.EllipticCurvePrivateKey) -> x509.Certificate:
232 name = x509.Name(
233 [
234 x509.NameAttribute(
235 x509.NameOID.COMMON_NAME,
236 binascii.hexlify(os.urandom(16)).decode("ascii"),
237 )
238 ]
239 )
240 now = datetime.datetime.now(tz=datetime.timezone.utc)
241 builder = (
242 x509.CertificateBuilder()
243 .subject_name(name)
244 .issuer_name(name)
245 .public_key(key.public_key())
246 .serial_number(x509.random_serial_number())
247 .not_valid_before(now - datetime.timedelta(days=1))
248 .not_valid_after(now + datetime.timedelta(days=30))
249 )
250 return builder.sign(key, hashes.SHA256(), default_backend())
253class State(enum.Enum):
254 NEW = 0
255 CONNECTING = 1
256 CONNECTED = 2
257 CLOSED = 3
258 FAILED = 4
261@dataclass
262class RTCDtlsFingerprint:
263 """
264 The :class:`RTCDtlsFingerprint` dictionary includes the hash function
265 algorithm and certificate fingerprint.
266 """
268 algorithm: str
269 "The hash function name, for instance `'sha-256'`."
271 value: str
272 "The fingerprint value."
275class RTCCertificate:
276 """
277 The :class:`RTCCertificate` interface enables the certificates used by an
278 :class:`RTCDtlsTransport`.
280 To generate a certificate and the corresponding private key use
281 :func:`generateCertificate`.
282 """
284 def __init__(self, key: ec.EllipticCurvePrivateKey, cert: x509.Certificate) -> None:
285 self._key = key
286 self._cert = cert
288 @property
289 def expires(self) -> datetime.datetime:
290 """
291 The date and time after which the certificate will be considered invalid.
292 """
293 return self._cert.not_valid_after_utc
295 def getFingerprints(self) -> list[RTCDtlsFingerprint]:
296 """
297 Returns the list of certificate fingerprints, one of which is computed
298 with the digest algorithm used in the certificate signature.
299 """
300 return [
301 RTCDtlsFingerprint(
302 algorithm=algorithm,
303 value=certificate_digest(self._cert, algorithm),
304 )
305 for algorithm in X509_DIGEST_ALGORITHMS.keys()
306 ]
308 @classmethod
309 def generateCertificate(cls: Type[CERTIFICATE_T]) -> CERTIFICATE_T:
310 """
311 Create and return an X.509 certificate and corresponding private key.
313 :rtype: RTCCertificate
314 """
315 key = ec.generate_private_key(ec.SECP256R1(), default_backend())
316 cert = generate_certificate(key)
317 return cls(key=key, cert=cert)
319 def _create_ssl_context(
320 self, srtp_profiles: list[SRTPProtectionProfile]
321 ) -> SSL.Context:
322 ctx = SSL.Context(SSL.DTLS_METHOD)
323 ctx.set_verify(
324 SSL.VERIFY_PEER | SSL.VERIFY_FAIL_IF_NO_PEER_CERT, lambda *args: True
325 )
326 ctx.use_certificate(self._cert)
327 ctx.use_privatekey(self._key)
328 ctx.set_cipher_list(
329 b"ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-ECDSA-AES128-SHA:ECDHE-ECDSA-AES256-SHA"
330 )
331 ctx.set_tlsext_use_srtp(b":".join(x.openssl_profile for x in srtp_profiles))
333 return ctx
336@dataclass
337class RTCDtlsParameters:
338 """
339 The :class:`RTCDtlsParameters` dictionary includes information relating to
340 DTLS configuration.
341 """
343 fingerprints: list[RTCDtlsFingerprint] = field(default_factory=list)
344 "List of :class:`RTCDtlsFingerprint`, one fingerprint for each certificate."
346 role: str = "auto"
347 "The DTLS role, with a default of auto."
350class DataReceiver(Protocol):
351 async def _handle_data(self, data: bytes) -> None: ...
354class RtpReceiver(Protocol):
355 def _handle_disconnect(self) -> None: ...
356 async def _handle_rtcp_packet(self, packet: AnyRtcpPacket) -> None: ...
357 async def _handle_rtp_packet(
358 self, packet: RtpPacket, arrival_time_ms: int
359 ) -> None: ...
362class RtpSender(Protocol):
363 _ssrc: int
365 @property
366 def kind(self) -> str: ...
368 async def _handle_rtcp_packet(self, packet: AnyRtcpPacket) -> None: ...
371class RtpRouter:
372 """
373 Router to associate RTP/RTCP packets with streams.
375 https://tools.ietf.org/html/draft-ietf-mmusic-sdp-bundle-negotiation-53
376 """
378 def __init__(self) -> None:
379 self.receivers: set[RtpReceiver] = set()
380 self.senders: dict[int, RtpSender] = {}
381 self.mid_table: dict[str, RtpReceiver] = {}
382 self.ssrc_table: dict[int, RtpReceiver] = {}
383 self.payload_type_table: dict[int, set[RtpReceiver]] = {}
385 def register_receiver(
386 self,
387 receiver: RtpReceiver,
388 ssrcs: list[int],
389 payload_types: list[int],
390 mid: Optional[str] = None,
391 ) -> None:
392 self.receivers.add(receiver)
393 if mid is not None:
394 self.mid_table[mid] = receiver
395 for ssrc in ssrcs:
396 self.ssrc_table[ssrc] = receiver
397 for payload_type in payload_types:
398 if payload_type not in self.payload_type_table:
399 self.payload_type_table[payload_type] = set()
400 self.payload_type_table[payload_type].add(receiver)
402 def register_sender(self, sender: RtpSender, ssrc: int) -> None:
403 self.senders[ssrc] = sender
405 def route_rtcp(self, packet: AnyRtcpPacket) -> set[Union[RtpReceiver, RtpSender]]:
406 recipients: set[Union[RtpReceiver, RtpSender]] = set()
408 def add_recipient(recipient: Optional[Union[RtpReceiver, RtpSender]]) -> None:
409 if recipient is not None:
410 recipients.add(recipient)
412 # route to RTP receiver
413 if isinstance(packet, RtcpSrPacket):
414 add_recipient(self.ssrc_table.get(packet.ssrc))
415 elif isinstance(packet, RtcpByePacket):
416 for source in packet.sources:
417 add_recipient(self.ssrc_table.get(source))
419 # route to RTP sender
420 if isinstance(packet, (RtcpRrPacket, RtcpSrPacket)):
421 for report in packet.reports:
422 add_recipient(self.senders.get(report.ssrc))
423 elif isinstance(packet, (RtcpPsfbPacket, RtcpRtpfbPacket)):
424 # Transport-cc feedback is association-wide (media_ssrc=0), so keying
425 # by media_ssrc alone discards it entirely. Route it to ONE sender
426 # (video first: this is where congestion control input matters).
427 if isinstance(packet, RtcpRtpfbPacket) and packet.fmt == rtp.RTCP_RTPFB_TWCC:
428 if self.senders:
429 add_recipient(min(self.senders.values(),
430 key=lambda s: s.kind != "video"))
431 add_recipient(self.senders.get(packet.media_ssrc))
433 # for REMB packets, media_ssrc is always 0, we need to look into the FCI
434 if isinstance(packet, RtcpPsfbPacket) and packet.fmt == rtp.RTCP_PSFB_APP:
435 try:
436 for ssrc in rtp.unpack_remb_fci(packet.fci)[1]:
437 add_recipient(self.senders.get(ssrc))
438 except ValueError:
439 pass
440 elif isinstance(packet, RtcpXrPacket) and packet.rrtr is not None:
441 # A reference time names no media source: every sender answers it.
442 for sender in self.senders.values():
443 add_recipient(sender)
445 return recipients
447 def route_rtp(self, packet: RtpPacket) -> Optional[RtpReceiver]:
448 ssrc_receiver = self.ssrc_table.get(packet.ssrc)
449 pt_receivers = self.payload_type_table.get(packet.payload_type, set())
451 # the SSRC and payload type are known and match
452 if ssrc_receiver is not None and ssrc_receiver in pt_receivers:
453 return ssrc_receiver
455 # the SSRC is unknown but the payload type matches, update the SSRC table
456 if ssrc_receiver is None and len(pt_receivers) == 1:
457 pt_receiver = list(pt_receivers)[0]
458 self.ssrc_table[packet.ssrc] = pt_receiver
459 return pt_receiver
461 # discard the packet
462 return None
464 def unregister_receiver(self, receiver: RtpReceiver) -> None:
465 self.receivers.discard(receiver)
466 self.__discard(self.mid_table, receiver)
467 self.__discard(self.ssrc_table, receiver)
468 for pt, receivers in self.payload_type_table.items():
469 receivers.discard(receiver)
471 def unregister_sender(self, sender: RtpSender) -> None:
472 self.__discard(self.senders, sender)
474 def __discard(self, d: dict[K, V], value: V) -> None:
475 for k, v in list(d.items()):
476 if v == value:
477 d.pop(k)
480class RTCDtlsTransport(AsyncIOEventEmitter):
481 """
482 The :class:`RTCDtlsTransport` object includes information relating to
483 Datagram Transport Layer Security (DTLS) transport.
485 :param transport: An :class:`RTCIceTransport`.
486 :param certificates: A list of :class:`RTCCertificate` (only one is allowed
487 currently).
488 """
490 def __init__(
491 self, transport: RTCIceTransport, certificates: list[RTCCertificate]
492 ) -> None:
493 assert len(certificates) == 1
494 certificate = certificates[0]
496 super().__init__()
497 self.encrypted = False
498 self._data_receiver: Optional[DataReceiver] = None
499 self._role = "auto"
500 self._rtp_header_extensions_map = rtp.HeaderExtensionsMap()
501 self._rtp_router = RtpRouter()
502 self._state = State.NEW
503 self._stats_id = "transport_" + str(id(self))
504 self._task: Optional[asyncio.Future[None]] = None
505 self._transport = transport
507 # transport-wide congestion control: one sequence-number space shared by
508 # every sender on this transport, a bounded send history for matching the
509 # receiver's feedback, and the latest loss/delay estimate derived from it.
510 self._twcc_seq = 0
511 self._twcc_history: dict[int, tuple[int, float]] = {}
512 self._twcc_missing: dict[int, int] = {}
513 self._twcc_arrived: dict[int, float] = {}
514 self._twcc_window_id = 0
515 self._twcc_pruned_at = 0.0
516 self.twcc_estimate: Optional[dict] = None
517 self._twcc_window = self._twcc_window_zero()
518 self._twcc_reference: Optional[int] = None
519 self._twcc_delay_floor: deque = deque()
520 self._twcc_delay_base: Optional[tuple] = None
521 # The one-way delay of the newest arrival the newest feedback reported, and the
522 # least any reported, on the feedback's clock as `_twcc_delay_base` is; and when
523 # a feedback last reported a packet the wire lost.
524 self._twcc_delay_newest: Optional[float] = None
525 self._twcc_delay_least: Optional[float] = None
526 self._twcc_lost_at = 0.0
527 self._twcc_received_at = 0.0
528 self._twcc_sent_at = 0.0
529 self._twcc_stand: Optional[dict] = None
530 # Receive side of transport-wide congestion control: a sender that
531 # negotiates transport-cc runs its bandwidth estimation on this
532 # feedback alone and starves at its floor bitrate without it.
533 self._twcc_rx_arrivals: dict[int, float] = {}
534 self._twcc_rx_last_ext: Optional[int] = None
535 self._twcc_rx_next: Optional[int] = None
536 self._twcc_rx_last_fb = 0.0
537 self._twcc_rx_fb_count = 0
539 # Optional strict-priority packet pacer (SELKIES_WEBRTC_PACER): queues
540 # video behind audio/RTCP/data instead of blasting frames to the wire,
541 # with an IDR-aware budget and GOP-reset recovery. Off unless enabled by
542 # the application with enable_pacer().
543 self._pacer: Optional[RtpPacer] = None
544 # `frame_end` without a pacer: the packets it waits on, by sequence number.
545 self._frame_ends: Dict[int, tuple] = {}
547 # counters
548 self.__rx_bytes = 0
549 self.__rx_packets = 0
550 self.__tx_bytes = 0
551 self.__tx_packets = 0
553 # SRTP
554 self._rx_srtp: Session = None
555 self._tx_srtp: Session = None
557 # SSL
558 self._srtp_profiles = SRTP_PROFILES
559 self._ssl: Optional[SSL.Connection] = None
560 self.__local_certificate = certificate
562 @property
563 def state(self) -> str:
564 """
565 The current state of the DTLS transport.
567 One of `'new'`, `'connecting'`, `'connected'`, `'closed'`, or `'failed'`.
568 """
569 return str(self._state)[6:].lower()
571 @property
572 def transport(self) -> RTCIceTransport:
573 """
574 The associated :class:`RTCIceTransport` instance.
575 """
576 return self._transport
578 def getLocalParameters(self) -> RTCDtlsParameters:
579 """
580 Get the local parameters of the DTLS transport.
582 :rtype: :class:`RTCDtlsParameters`
583 """
584 return RTCDtlsParameters(
585 fingerprints=self.__local_certificate.getFingerprints()
586 )
588 async def _do_handshake(self) -> None:
589 """
590 Attempt to complete the DTLS handshake.
591 """
592 try:
593 while not self.encrypted:
594 try:
595 self._ssl.do_handshake()
596 except SSL.WantReadError:
597 await self._write_ssl()
598 await self._recv_next()
599 except SSL.Error as exc:
600 self.__log_debug("x DTLS handshake failed (error %s)", exc)
601 self._set_state(State.FAILED)
602 return
603 else:
604 self.encrypted = True
605 except ConnectionError:
606 self.__log_debug("x DTLS handshake failed (connection error)")
607 self._set_state(State.FAILED)
608 return
610 def _validate_peer_identity(self, remoteParameters: RTCDtlsParameters) -> None:
611 """
612 Check remote fingerprints. There must be at least one fingerprint
613 with a supported algorithm, and all supported fingerprints must
614 match.
615 """
616 certificate = self._ssl.get_peer_certificate(as_cryptography=True)
617 fingerprint_supported = 0
618 fingerprint_valid = 0
619 for f in remoteParameters.fingerprints:
620 algorithm = f.algorithm.lower()
621 if algorithm in X509_DIGEST_ALGORITHMS:
622 fingerprint_supported += 1
623 if f.value.upper() == certificate_digest(certificate, algorithm):
624 fingerprint_valid += 1
625 if not fingerprint_supported or fingerprint_valid != fingerprint_supported:
626 self.__log_debug("x DTLS handshake failed (fingerprint mismatch)")
627 self._set_state(State.FAILED)
628 return
630 def _setup_srtp(self) -> None:
631 """
632 Extract the SRTP keying material and setup the SRTP sessions.
633 """
634 openssl_profile = self._ssl.get_selected_srtp_profile()
635 for srtp_profile in self._srtp_profiles:
636 if srtp_profile.openssl_profile == openssl_profile:
637 self.__log_debug(
638 "x DTLS handshake negotiated %s",
639 srtp_profile.openssl_profile.decode(),
640 )
641 break
642 else:
643 self.__log_debug("x DTLS handshake failed (no SRTP profile negotiated)")
644 self._set_state(State.FAILED)
645 return
646 view = self._ssl.export_keying_material(
647 b"EXTRACTOR-dtls_srtp",
648 2 * (srtp_profile.key_length + srtp_profile.salt_length),
649 )
650 if self._role == "server":
651 srtp_tx_key = srtp_profile.get_key_and_salt(view, 1)
652 srtp_rx_key = srtp_profile.get_key_and_salt(view, 0)
653 else:
654 srtp_tx_key = srtp_profile.get_key_and_salt(view, 0)
655 srtp_rx_key = srtp_profile.get_key_and_salt(view, 1)
657 rx_policy = Policy(
658 key=srtp_rx_key,
659 ssrc_type=Policy.SSRC_ANY_INBOUND,
660 srtp_profile=srtp_profile.libsrtp_profile,
661 )
662 rx_policy.allow_repeat_tx = True
663 rx_policy.window_size = 1024
664 self._rx_srtp = Session(rx_policy)
666 tx_policy = Policy(
667 key=srtp_tx_key,
668 ssrc_type=Policy.SSRC_ANY_OUTBOUND,
669 srtp_profile=srtp_profile.libsrtp_profile,
670 )
671 tx_policy.allow_repeat_tx = True
672 tx_policy.window_size = 1024
673 self._tx_srtp = Session(tx_policy)
675 async def start(self, remoteParameters: RTCDtlsParameters) -> None:
676 """
677 Start DTLS transport negotiation with the parameters of the remote
678 DTLS transport.
680 :param remoteParameters: An :class:`RTCDtlsParameters`.
681 """
682 assert self._state == State.NEW
683 assert len(remoteParameters.fingerprints)
685 # For WebRTC, the DTLS role is explicitly determined as part of the
686 # offer / answer exchange.
687 #
688 # For ORTC however, we determine the DTLS role based on the ICE role.
689 if self._role == "auto":
690 if self.transport.role == "controlling":
691 self._set_role("server")
692 else:
693 self._set_role("client")
695 # Initialise SSL.
696 self._ssl = SSL.Connection(
697 self.__local_certificate._create_ssl_context(
698 srtp_profiles=self._srtp_profiles
699 )
700 )
701 if self._role == "server":
702 self._ssl.set_accept_state()
703 else:
704 self._ssl.set_connect_state()
706 # Start the DTLS handshake.
707 self._set_state(State.CONNECTING)
708 await self._do_handshake()
709 if self._state == State.FAILED:
710 return
712 # Validate the peer identity.
713 self._validate_peer_identity(remoteParameters)
714 if self._state == State.FAILED:
715 return
717 # Generate keying material.
718 self._setup_srtp()
719 if self._state == State.FAILED:
720 return
722 # start data pump
723 self.__log_debug("- DTLS handshake complete")
724 self._set_state(State.CONNECTED)
725 self._task = asyncio.ensure_future(self.__run())
727 async def stop(self) -> None:
728 """
729 Stop and close the DTLS transport.
730 """
731 if self._task is not None:
732 self._task.cancel()
733 self._task = None
735 if self._pacer is not None:
736 try:
737 await self._pacer.close()
738 finally:
739 self._pacer = None
741 if self._ssl and self._state in [State.CONNECTING, State.CONNECTED]:
742 try:
743 self._ssl.shutdown()
744 except SSL.Error:
745 pass
746 try:
747 await self._write_ssl()
748 except ConnectionError:
749 pass
750 self.__log_debug("- DTLS shutdown complete")
752 async def __run(self) -> None:
753 try:
754 while True:
755 await self._recv_next()
756 except ConnectionError:
757 for receiver in self._rtp_router.receivers:
758 receiver._handle_disconnect()
759 except Exception as exc:
760 if not isinstance(exc, asyncio.CancelledError):
761 self.__log_warning(traceback.format_exc())
762 raise exc
763 finally:
764 self._set_state(State.CLOSED)
766 def _get_stats(self) -> RTCStatsReport:
767 report = RTCStatsReport()
768 report.add(
769 RTCTransportStats(
770 # RTCStats
771 timestamp=clock.current_datetime(),
772 type="transport",
773 id=self._stats_id,
774 # RTCTransportStats,
775 packetsSent=self.__tx_packets,
776 packetsReceived=self.__rx_packets,
777 bytesSent=self.__tx_bytes,
778 bytesReceived=self.__rx_bytes,
779 iceRole=self.transport.role,
780 dtlsState=self.state,
781 )
782 )
783 return report
785 async def _handle_rtcp_data(self, data: bytes) -> None:
786 try:
787 packets = RtcpPacket.parse(data)
788 except (ValueError, struct.error) as exc:
789 self.__log_debug("x RTCP parsing failed: %s", exc)
790 return
792 for packet in packets:
793 # route RTCP packet
794 for recipient in self._rtp_router.route_rtcp(packet):
795 await recipient._handle_rtcp_packet(packet)
797 async def _handle_rtp_data(self, data: bytes, arrival_time_ms: int) -> None:
798 try:
799 packet = RtpPacket.parse(data, self._rtp_header_extensions_map)
800 except (ValueError, struct.error) as exc:
801 self.__log_debug("x RTP parsing failed: %s", exc)
802 return
804 # route RTP packet
805 receiver = self._rtp_router.route_rtp(packet)
806 if receiver is not None:
807 await receiver._handle_rtp_packet(packet, arrival_time_ms=arrival_time_ms)
809 async def _recv_next(self) -> None:
810 """Receive a datagram and drain its complete DTLS application records.
812 OpenSSL can retain unprocessed records while pending() reports zero.
813 Read until it needs input so a complete SCTP packet never waits for
814 an unrelated later datagram.
815 """
816 # get timeout
817 timeout = None
818 if not self.encrypted:
819 timeout = self._ssl.DTLSv1_get_timeout()
821 # receive next datagram
822 if timeout is not None:
823 try:
824 data = await asyncio.wait_for(self.transport._recv(), timeout=timeout)
825 except asyncio.TimeoutError:
826 self.__log_debug("x DTLS handling timeout")
827 self._ssl.DTLSv1_handle_timeout()
828 await self._write_ssl()
829 return
830 else:
831 data = await self.transport._recv()
833 self.__rx_bytes += len(data)
834 self.__rx_packets += 1
836 first_byte = data[0]
837 if first_byte > 19 and first_byte < 64:
838 # DTLS
839 self._ssl.bio_write(data)
840 while True:
841 try:
842 data = self._ssl.recv(1500)
843 except SSL.ZeroReturnError:
844 data = None
845 except SSL.Error:
846 data = b""
847 await self._write_ssl()
848 if data is None:
849 self.__log_debug("- DTLS shutdown by remote party")
850 raise ConnectionError
851 elif not data:
852 break
853 elif self._data_receiver:
854 await self._data_receiver._handle_data(data)
855 elif first_byte > 127 and first_byte < 192 and self._rx_srtp:
856 # SRTP / SRTCP
857 arrival_time_ms = clock.current_ms()
858 try:
859 if is_rtcp(data):
860 data = self._rx_srtp.unprotect_rtcp(data)
861 await self._handle_rtcp_data(data)
862 else:
863 data = self._rx_srtp.unprotect(data)
864 await self._handle_rtp_data(data, arrival_time_ms=arrival_time_ms)
865 except pylibsrtp.Error as exc:
866 self.__log_debug("x SRTP unprotect failed: %s", exc)
868 def _register_data_receiver(self, receiver: DataReceiver) -> None:
869 assert self._data_receiver is None
870 self._data_receiver = receiver
872 def _register_rtp_receiver(
873 self, receiver: RtpReceiver, parameters: RTCRtpReceiveParameters
874 ) -> None:
875 ssrcs = set()
876 for encoding in parameters.encodings:
877 ssrcs.add(encoding.ssrc)
879 self._rtp_header_extensions_map.configure(parameters)
880 self._rtp_router.register_receiver(
881 receiver,
882 ssrcs=list(ssrcs),
883 payload_types=[codec.payloadType for codec in parameters.codecs],
884 mid=parameters.muxId,
885 )
887 def _register_rtp_sender(
888 self, sender: RtpSender, parameters: RTCRtpSendParameters
889 ) -> None:
890 self._rtp_header_extensions_map.configure(parameters)
891 self._rtp_router.register_sender(sender, ssrc=sender._ssrc)
893 async def _send_data(self, data: bytes, rtc_class: Optional[int] = None) -> None:
894 if self._state != State.CONNECTED:
895 raise ConnectionError("Cannot send encrypted data, not connected")
897 # Data-channel traffic rides the pacer as class DC unless the caller
898 # names one: ahead of queued video, behind audio/RTCP.
899 if self._pacer is not None:
900 await self._pacer.send(data, rtc_class if rtc_class is not None else CLASS_DC)
901 return
902 self._ssl.send(data)
903 await self._write_ssl()
905 def pacer_enabled(self) -> bool:
906 return self._pacer is not None
908 def set_video_resync(self, resync: Optional[Callable[[Optional[int]], bool]]) -> None:
909 """The video sender's hook for a pacer GOP reset (`RtpPacer._resync`), kept for a
910 pacer enabled later."""
911 self._video_resync = resync
912 if self._pacer is not None:
913 self._pacer._resync = resync
915 def enable_pacer(
916 self,
917 encoder_bps: int,
918 request_keyframe: Optional[Callable[[], None]] = None,
919 ) -> None:
920 """Attach the packet pacer to this transport (idempotent)."""
921 if self._pacer is None:
922 async def _send_now_data(payload: bytes) -> None:
923 self._ssl.send(payload)
924 await self._write_ssl()
925 self._pacer = RtpPacer(
926 encoder_bps=encoder_bps,
927 # Bound coroutine fn directly: the RTP hot path cannot afford
928 # a wrapper frame around every packet.
929 send_now=self.transport._send,
930 send_now_data=_send_now_data,
931 request_keyframe=request_keyframe,
932 on_dropped=self._twcc_dropped,
933 resync=getattr(self, "_video_resync", None),
934 )
935 elif request_keyframe is not None:
936 self._pacer._request_keyframe = request_keyframe
938 def set_pacer_encoder_bps(self, bps: int) -> None:
939 if self._pacer is not None:
940 self._pacer.set_encoder_bps(bps)
942 def note_video_keyframe(self, total_payload_bytes: int, natural: bool = True) -> None:
943 """Video senders report keyframe emissions here so the pacer can keep
944 its queue budget above the (content-variable) IDR size and resurrect
945 the video class after a GOP reset. `natural=False` marks forced
946 recovery keyframes: they resurrect the stream but must not shrink
947 the IDR budget floor."""
948 if self._pacer is not None:
949 self._pacer.note_keyframe(total_payload_bytes, natural)
951 def _peer_heard_at(self) -> float:
952 """Monotonic time the peer was last heard from on the ICE path: media, DTLS,
953 SCTP, or one of its consent checks."""
954 return self.transport._connection.last_received
956 def _send_delay(self) -> float:
957 """Seconds until a packet handed over now is on the wire."""
958 return self._pacer.drain_s() if self._pacer is not None else 0.0
960 def video_backlog(self) -> tuple[int, float, float]:
961 """The frames whose last packet waits in the pacer; how long a packet handed
962 over now waits to reach the peer: what the pacer holds takes to leave, and the
963 queue the newest transport-cc feedback found standing on the path, its newest
964 arrival's delay over the path's own (`_twcc_path_delay`; the least delay
965 reported until the first control interval dates that); and the seconds since
966 a feedback last reported a packet the wire lost."""
967 frames, wait = (0, 0.0) if self._pacer is None else (
968 len(self._pacer.frame_ends), self._pacer.drain_s())
969 base = self._twcc_delay_base[1] if self._twcc_delay_base is not None else self._twcc_delay_least
970 if self._twcc_delay_newest is not None and base is not None:
971 wait += max(0.0, self._twcc_delay_newest - base)
972 return frames, wait, time.monotonic() - self._twcc_lost_at
974 def video_room(self) -> Optional[int]:
975 """Bytes of video the pacer queues before it abandons the GOP; None without a pacer."""
976 return None if self._pacer is None else self._pacer.video_room()
978 def pacer_snapshot(self) -> Optional[dict]:
979 return self._pacer.snapshot() if self._pacer is not None else None
981 def pacer_room_bps(self) -> Optional[float]:
982 """The rate a still screen's cleanup may take through the pacer (`RtpPacer.room_bps`)."""
983 return self._pacer.room_bps() if self._pacer is not None else None
985 def set_pacer_frame_floor(self, size: int) -> None:
986 """Let the pacer's video queue take a cleanup frame of `size` bytes whole."""
987 if self._pacer is not None:
988 self._pacer.set_frame_floor(size)
990 async def _send_rtp(self, data: bytes, rtc_class: Optional[int] = None,
991 twcc_seq: Optional[int] = None) -> bool:
992 """Protect and send one RTP or RTCP packet; False when the pacer dropped
993 it. `twcc_seq` names the packet in the transport-cc history, so a drop
994 is accounted as one."""
995 if self._state != State.CONNECTED:
996 raise ConnectionError("Cannot send encrypted RTP, not connected")
998 if is_rtcp(data):
999 data = self._tx_srtp.protect_rtcp(data)
1000 cls = CLASS_RTCP
1001 else:
1002 data = self._tx_srtp.protect(data)
1003 cls = rtc_class if rtc_class is not None else CLASS_VIDEO
1004 sent = True
1005 if self._pacer is not None:
1006 sent = await self._pacer.send(data, cls, twcc_seq)
1007 else:
1008 await self.transport._send(data)
1009 end = self._frame_ends.pop(twcc_seq, None) if self._frame_ends else None
1010 if end is not None:
1011 end[0](*end[1])
1012 self.__tx_bytes += len(data)
1013 self.__tx_packets += 1
1014 return sent
1016 def frame_end(self, twcc_seq: int, sink: Callable[..., None], *args: Any) -> None:
1017 """Call `sink(*args)` once the RTP packet with transport-wide sequence number
1018 `twcc_seq`, the last of its frame, is on the wire: past the pacer where one
1019 paces, else as it is written to the socket."""
1020 if self._pacer is not None:
1021 self._pacer.frame_end(twcc_seq, sink, *args)
1022 return
1023 ends = self._frame_ends
1024 ends[twcc_seq] = (sink, args)
1025 if len(ends) > FRAME_ENDS_MAX:
1026 del ends[next(iter(ends))]
1028 def forget_frame_end(self, twcc_seq: int) -> None:
1029 """Forget the frame whose last packet, `twcc_seq`, will not be sent."""
1030 (self._pacer.frame_ends if self._pacer is not None else self._frame_ends).pop(twcc_seq, None)
1032 async def _twcc_received(self, seq: int, arrival_ms: float, media_ssrc: int,
1033 rtcp_ssrc: Optional[int]) -> None:
1034 """Record one received packet's transport-wide sequence and answer with
1035 feedback at most every 100 ms, built inline on arrival: while packets
1036 flow the estimator is fed at the cadence browsers use, and when they
1037 stop there is nothing left to report."""
1038 last = self._twcc_rx_last_ext
1039 if last is None:
1040 ext = seq
1041 else:
1042 delta = (seq - last) & 0xFFFF
1043 ext = last + delta - (0x10000 if delta >= 0x8000 else 0)
1044 if ext < 0:
1045 return
1046 if last is None or ext > last:
1047 self._twcc_rx_last_ext = ext
1048 if self._twcc_rx_next is not None and ext < self._twcc_rx_next:
1049 return
1050 self._twcc_rx_arrivals[ext] = arrival_ms
1051 now = time.time()
1052 if now - self._twcc_rx_last_fb < 0.1:
1053 return
1054 base = self._twcc_rx_next
1055 if base is None:
1056 base = min(self._twcc_rx_arrivals)
1057 end = self._twcc_rx_last_ext
1058 if end - base + 1 > 0x7FFF:
1059 base = end - 1024
1060 arrivals = [self._twcc_rx_arrivals.get(i) for i in range(base, end + 1)]
1061 fci = pack_twcc_fci(base, arrivals, self._twcc_rx_fb_count)
1062 if not fci:
1063 return
1064 self._twcc_rx_last_fb = now
1065 self._twcc_rx_fb_count = (self._twcc_rx_fb_count + 1) & 0xFF
1066 for key in [k for k in self._twcc_rx_arrivals if k <= end]:
1067 del self._twcc_rx_arrivals[key]
1068 self._twcc_rx_next = end + 1
1069 packet = RtcpRtpfbPacket(
1070 fmt=rtp.RTCP_RTPFB_TWCC,
1071 ssrc=rtcp_ssrc if rtcp_ssrc is not None else 1,
1072 media_ssrc=media_ssrc,
1073 fci=fci,
1074 )
1075 try:
1076 await self._send_rtp(bytes(packet))
1077 except ConnectionError:
1078 pass
1080 def _twcc_next(self, size: int) -> int:
1081 """Allocate the next transport-wide sequence number and record the packet's
1082 size and send time for matching against the receiver's transport-cc feedback."""
1083 seq = self._twcc_seq
1084 self._twcc_seq = (self._twcc_seq + 1) & 0xFFFF
1085 now = self._twcc_sent_at = time.monotonic()
1086 self._twcc_history.pop(seq, None)
1087 self._twcc_missing.pop(seq, None)
1088 self._twcc_history[seq] = (size, now)
1089 self._twcc_window["bytes_sent"] += size
1090 # Bounded by age, not count: a retransmission storm allocates thousands
1091 # of numbers a second, and one let go before its feedback arrives would
1092 # read as wire loss. Insertion order is allocation order.
1093 if len(self._twcc_history) > 2048 and now - self._twcc_pruned_at > 0.25:
1094 self._twcc_pruned_at = now
1095 for old, (_size, at) in list(self._twcc_history.items()):
1096 if now - at < TWCC_HISTORY_S:
1097 break
1098 del self._twcc_history[old]
1099 self._twcc_missing.pop(old, None)
1100 return seq
1102 def twcc_arrived(self, seq: int) -> bool:
1103 """Whether transport-cc feedback reported the packet sent under `seq` received
1104 within TWCC_ARRIVED_S."""
1105 return seq in self._twcc_arrived
1107 def _twcc_acked(self, seq: int) -> bool:
1108 """Whether transport-cc feedback reported the packet sent under `seq`, sent
1109 within TWCC_HISTORY_S, received."""
1110 return seq not in self._twcc_history
1112 def _twcc_unmark_missing(self, seq: int) -> None:
1113 """Retract a provisional loss only while its control interval is open.
1115 A positive acknowledgment or a local discard resolves the negative.
1116 Intervals already consumed by the controller remain unchanged.
1117 """
1118 if self._twcc_missing.pop(seq, None) == self._twcc_window_id:
1119 self._twcc_window["lost"] -= 1
1121 def _twcc_dropped(self, seq: int) -> None:
1122 """Record that the pacer dropped the packet sent under `seq`, so the
1123 receiver reporting it missing counts as nothing the wire lost."""
1124 if seq in self._twcc_history:
1125 self._twcc_history[seq] = (0, self._twcc_history[seq][1])
1126 self._twcc_unmark_missing(seq)
1128 def _twcc_process_feedback(self, fci: bytes) -> None:
1129 """Decode a transport-cc feedback FCI (draft-holmer-rmcat-transport-wide-cc):
1130 walk the packet-status chunks and receive deltas, join them against the send
1131 history, and publish a loss / throughput / delay estimate.
1133 A missing status is provisional: the packet may still be in transit.
1134 Retain its history until a positive acknowledgment or normal expiry,
1135 and count the negative once. A late positive contributes its bytes
1136 once and retracts a negative only from the current control interval.
1137 Missing markers carry interval numbers rather than retaining drained
1138 windows, and expire or reset with their send-history entries. Unknown
1139 packets and local discards contribute neither loss nor delivery;
1140 feedback containing no new observations cannot steer the pacer.
1142 The deltas chain arrival times: the first is the first arrival's offset
1143 from the feedback's reference time, each later one the gap from the
1144 arrival before, so the interval the bytes were delivered over runs from
1145 the earliest arrival to the latest, and the earliest packet's bytes were
1146 not delivered inside it. The reference time, 64 ms units on the
1147 receiver's clock that wrap at 24 bits, puts every feedback's arrivals on
1148 that one clock, so an arrival less its send time is the packet's one-way
1149 delay plus a constant offset between the two clocks, comparable across
1150 feedback packets; the least of those in a control interval is what
1151 `take_twcc_window` measures a standing queue from. Only arrivals still in the send history take
1152 part: a receiver that moved its window back over a late packet reports
1153 packets it already reported, which would stretch the interval and add
1154 nothing to it. A silence of TWCC_IDLE_US or more between two arrivals
1155 is left out of that interval, with the packet ending it: it is an
1156 outage or an idle sender, and a rate measured across it would say the
1157 wire carries almost nothing. A feedback whose chunks or deltas run
1158 short, or that carries the reserved status symbol, is dropped whole,
1159 so a malformed one consumes no history. Each feedback also tells the
1160 pacer how fast the wire delivered frames' leading bursts
1161 (`burst_delivery_bps`), or that it lost a packet, which sizes the
1162 pacer's burst budget."""
1163 if len(fci) < 8:
1164 return
1165 base_seq, status_count = struct.unpack("!HH", fci[0:4])
1166 reference = int.from_bytes(fci[4:7], "big")
1167 if self._twcc_reference is not None:
1168 step = (reference - self._twcc_reference) & 0xFFFFFF
1169 reference = self._twcc_reference + (step - 0x1000000 if step & 0x800000 else step)
1170 pos = 8
1171 statuses: list[int] = []
1172 while len(statuses) < status_count and pos + 2 <= len(fci):
1173 chunk = struct.unpack("!H", fci[pos : pos + 2])[0]
1174 pos += 2
1175 if chunk & 0x8000 == 0:
1176 symbol = (chunk >> 13) & 0x3
1177 run = chunk & 0x1FFF
1178 statuses.extend([symbol] * run)
1179 elif chunk & 0x4000 == 0:
1180 for i in range(14):
1181 statuses.append((chunk >> (13 - i)) & 0x1)
1182 else:
1183 for i in range(7):
1184 statuses.append((chunk >> (12 - 2 * i)) & 0x3)
1185 statuses = statuses[:status_count]
1186 if len(statuses) < status_count or 3 in statuses:
1187 return
1189 arrivals: list[tuple[int, float]] = []
1190 missing: list[int] = []
1191 at_us = reference * 64_000.0
1192 for i, symbol in enumerate(statuses):
1193 if symbol == 1:
1194 if pos >= len(fci):
1195 return
1196 at_us += fci[pos] * 250
1197 pos += 1
1198 elif symbol == 2:
1199 if pos + 2 > len(fci):
1200 return
1201 at_us += struct.unpack("!h", fci[pos : pos + 2])[0] * 250
1202 pos += 2
1203 else:
1204 missing.append((base_seq + i) & 0xFFFF)
1205 continue
1206 arrivals.append(((base_seq + i) & 0xFFFF, at_us))
1207 if not statuses:
1208 return
1209 self._twcc_reference = reference
1210 received = 0
1211 # A packet the pacer dropped was never on the wire to lose. Loss the
1212 # wire spread through the window says it is short of room; one run of
1213 # it is an outage, which says nothing about the room there is.
1214 lost = runs = 0
1215 previous = None
1216 for seq in missing:
1217 sent = self._twcc_history.get(seq)
1218 if sent is None or sent[0] == 0 or seq in self._twcc_missing:
1219 continue
1220 self._twcc_missing[seq] = self._twcc_window_id
1221 lost += 1
1222 if previous is None or (seq - previous) & 0xFFFF != 1:
1223 runs += 1
1224 previous = seq
1225 # The delivery interval is the time between arrivals of packets in the
1226 # history, less any silence long enough to be an outage or an idle
1227 # sender rather than a rate; the packet opening each stretch was not
1228 # delivered inside it.
1229 bytes_acked = spanned = 0
1230 matched: list[tuple[float, int]] = []
1231 in_order: list[tuple[float, float, int]] = []
1232 delay_min = delay_last = None
1233 now = time.monotonic()
1234 arrived = self._twcc_arrived
1235 for seq, _at in arrivals:
1236 arrived.pop(seq, None)
1237 arrived[seq] = now
1238 while arrived and now - next(iter(arrived.values())) > TWCC_ARRIVED_S:
1239 del arrived[next(iter(arrived))]
1240 for seq, at in arrivals:
1241 sent = self._twcc_history.pop(seq, None)
1242 self._twcc_unmark_missing(seq)
1243 if sent is not None and sent[0] > 0:
1244 received += 1
1245 bytes_acked += sent[0]
1246 window = self._twcc_window
1247 key = (at, seq)
1248 if window["first_arrival"] is None or key < window["first_arrival"]:
1249 window["first_arrival"] = key
1250 window["first_bytes"] = sent[0]
1251 last = window["last_arrival_us"]
1252 window["last_arrival_us"] = at if last is None else max(last, at)
1253 matched.append((at, sent[0]))
1254 in_order.append((sent[1], at, sent[0]))
1255 delay_last = at / 1e6 - sent[1]
1256 if delay_min is None or delay_last < delay_min:
1257 delay_min = delay_last
1258 if not received and not lost:
1259 return
1260 matched.sort(key=lambda m: m[0])
1261 span_us = 0.0
1262 for (t0, _size), (t1, size) in zip(matched, matched[1:]):
1263 if t1 - t0 <= TWCC_IDLE_US:
1264 span_us += t1 - t0
1265 spanned += size
1266 span_s = span_us / 1e6
1267 goodput = int(spanned * 8 / span_s) if span_s > 0 else 0
1268 self.twcc_estimate = {
1269 "received": received,
1270 "lost": lost,
1271 "loss_fraction": lost / max(1, received + lost),
1272 "bytes_acked": bytes_acked,
1273 "recv_span_s": span_s,
1274 "goodput_bps": goodput,
1275 }
1276 if lost:
1277 self._twcc_lost_at = time.monotonic()
1278 if received:
1279 self._twcc_received_at = time.monotonic()
1280 window = self._twcc_window
1281 window["received"] += received
1282 window["lost"] += lost
1283 window["bytes_acked"] += bytes_acked
1284 if delay_min is not None:
1285 if window["delay_min"] is None or delay_min < window["delay_min"]:
1286 window["delay_min"] = delay_min
1287 window["delay_last"] = delay_last
1288 self._twcc_delay_newest = delay_last
1289 if self._twcc_delay_least is None or delay_min < self._twcc_delay_least:
1290 self._twcc_delay_least = delay_min
1291 window["feedback_mins"].append(delay_min)
1292 if self._pacer is not None:
1293 # A brake sizes itself from a rate the wire limited. A window the wire
1294 # delivered whole, or one carrying almost no data, measured the pacer's
1295 # own output instead, which a braked pacer can only ever read back.
1296 if (goodput and spanned >= MIN_GOODPUT_SAMPLE_BYTES
1297 and runs >= BRAKE_LOSS_RUNS and lost >= BRAKE_LOSS_FRACTION * (received + lost)):
1298 self._pacer.set_goodput_bps(goodput)
1299 self._pacer.set_link_bps(
1300 0.0 if lost else burst_delivery_bps(in_order, self._pacer.burst_probe_bytes))
1301 logger.debug(
1302 "TWCC feedback: recv=%d lost=%d goodput=%s bps",
1303 received,
1304 lost,
1305 goodput,
1306 )
1308 @staticmethod
1309 def _twcc_window_zero() -> dict:
1310 return {"received": 0, "lost": 0, "bytes_acked": 0,
1311 "first_arrival": None, "first_bytes": 0, "last_arrival_us": None,
1312 "delay_min": None, "delay_last": None, "feedback_mins": [], "bytes_sent": 0,
1313 "opened": time.monotonic()}
1315 def take_twcc_window(self) -> Optional[dict]:
1316 """Drain the transport-cc feedback accumulated since the last call.
1318 One feedback packet covers a few tens of RTP packets over a few tens of
1319 milliseconds, which is too little to measure either quantity: five lost
1320 out of twenty-two reads as 22.7% loss, and a 17 ms receive span turns a
1321 static screen's trickle into tens of Mbps. A control interval's worth of
1322 feedback is the measurement. Draining also means an interval that
1323 carried no feedback yields nothing to steer from, rather than the last
1324 window to apply again.
1326 Aggregate goodput spans the earliest through the latest unique known
1327 arrival, excluding only the earliest packet's bytes. Gaps between
1328 feedback reports and idle periods count as elapsed time: this is
1329 delivered throughput, not a measure of available capacity. Summing the
1330 spans inside individual reports omits their intervening gaps and makes
1331 batching inflate the rate. Equal-time arrivals use the transport
1332 sequence as a deterministic tie-breaker; late positives can extend the
1333 envelope, while duplicates add nothing. The three envelope fields are
1334 constant-size and reset on drain. Per-report pacer rates keep their
1335 separate idle-exclusion semantics.
1337 The interval's least one-way delay over the path's own
1338 (`_twcc_path_delay`) is the queue that stood through all of it: a key
1339 frame's burst delays the packets behind it for a moment, while a queue
1340 the rate has outgrown delays every one of them. A queue still building
1341 shows before it stands a whole interval as the least delay of the
1342 newest half of its feedback packets standing past the path's own, the
1343 least of a few hundred packets: jitter that spreads one-way delay over
1344 tens of milliseconds, as a Wi-Fi hop does, never lifts all of those,
1345 where a row of rising feedback packets comes from it once in 24
1346 intervals, and the packets behind a burst arrive ever sooner, so a
1347 burst shorter than half an interval never reads that way either. An
1348 interval of fewer than `TWCC_STAND_FEEDBACKS` feedback packets, a still
1349 screen's trickle, reads no queue: jitter lifts the least of one or two
1350 of them past any allowance, and a verdict there would take the
1351 trickle's goodput for the path's capacity. The newest arrival's delay
1352 over that floor is how deep the queue is by the interval's end, which a
1353 growing one has grown into.
1355 Returns:
1356 Loss, goodput and the rate sent over the drained interval, with the
1357 standing queue, the queue still building (None unless the newest
1358 half of the interval stood past `TWCC_QUEUE_MS`), and the depth, in
1359 milliseconds (None without a delay measured or with too few
1360 feedback packets to read one), or None when no feedback arrived in
1361 it.
1362 """
1363 window = self._twcc_window
1364 packets = window["received"] + window["lost"]
1365 if not packets:
1366 return None
1367 self._twcc_window = self._twcc_window_zero()
1368 self._twcc_window_id += 1
1369 queue_ms = rising_ms = depth_ms = None
1370 now = time.monotonic()
1371 first = window["first_arrival"]
1372 span_s = (window["last_arrival_us"] - first[0]) / 1e6 if first is not None else 0.0
1373 goodput_bps = (int((window["bytes_acked"] - window["first_bytes"]) * 8 / span_s)
1374 if span_s > 0 else 0)
1375 sent_bps = (int(window["bytes_sent"] * 8 / (now - window["opened"]))
1376 if now > window["opened"] else 0)
1377 if window["delay_min"] is not None:
1378 mins = window["feedback_mins"]
1379 least = self._twcc_path_delay(
1380 now, window["delay_min"], window["lost"] / packets, sent_bps, goodput_bps,
1381 len(mins) >= TWCC_STAND_FEEDBACKS)
1382 if len(mins) >= TWCC_STAND_FEEDBACKS:
1383 queue_ms = (window["delay_min"] - least) * 1000.0
1384 depth_ms = (window["delay_last"] - least) * 1000.0
1385 if len(mins) >= 2 * TWCC_STAND_FEEDBACKS:
1386 newest_ms = (min(mins[len(mins) // 2:]) - least) * 1000.0
1387 if newest_ms > TWCC_QUEUE_MS:
1388 rising_ms = newest_ms
1389 return {
1390 "received": window["received"],
1391 "lost": window["lost"],
1392 "loss_fraction": window["lost"] / packets,
1393 "bytes_acked": window["bytes_acked"],
1394 "goodput_bps": goodput_bps,
1395 "queue_ms": queue_ms,
1396 "queue_rising_ms": rising_ms,
1397 "queue_depth_ms": depth_ms,
1398 "sent_bps": sent_bps,
1399 }
1401 def _twcc_path_delay(self, now: float, delay: float, loss: float,
1402 sent_bps: int, goodput_bps: int, measured: bool) -> float:
1403 """The path's own one-way delay, in seconds on the feedback's clock,
1404 after an interval whose least delay was `delay`.
1406 It is the least delay seen, let rise only as fast as the two ends'
1407 clocks drift apart (`TWCC_DRIFT_S_PER_S`): the least of the last
1408 `TWCC_DELAY_FLOOR_S` alone takes a queue that stands that long for the
1409 path, and reads nothing of it from then on, while the loss it overflows
1410 into goes on. Delay standing `TWCC_QUEUE_MS` past it is then a queue or
1411 a longer path, and what the stream does tells them apart: over
1412 `TWCC_PATH_INTERVALS` intervals in a row that lost nothing and sent
1413 under `TWCC_PATH_SENT` of the most the path delivered while the delay
1414 stood, a queue the stream filled drains by a tenth of a second each,
1415 so delay that held within `TWCC_PATH_BAND_MS` through them is the
1416 path's and becomes its own from there. A rate cut answering the first
1417 reading provides those intervals, so a route change costs the few
1418 cuts until they have passed. Where nothing cuts the rate, or the
1419 delay jitters past the band, delay that has stood with nothing lost for
1420 `TWCC_DELAY_FLOOR_S` gives way to the least of that time, as before. An interval of too few feedback packets to
1421 read a queue from (`measured` unset) can only lower it.
1422 """
1423 floor = self._twcc_delay_floor
1424 floor.append((now, delay))
1425 while floor[0][0] < now - TWCC_DELAY_FLOOR_S:
1426 floor.popleft()
1427 if self._twcc_delay_base is None:
1428 base = delay
1429 else:
1430 at, base = self._twcc_delay_base
1431 base = min(delay, base + TWCC_DRIFT_S_PER_S * max(now - at, 0.0))
1432 band = TWCC_PATH_BAND_MS / 1000.0
1433 if (delay - base) * 1000.0 <= TWCC_QUEUE_MS:
1434 self._twcc_stand = None
1435 elif measured:
1436 stand = self._twcc_stand
1437 if stand is None:
1438 stand = self._twcc_stand = {"delivered": 0, "quiet": [], "lossless": now}
1439 stand["delivered"] = max(stand["delivered"], goodput_bps)
1440 lossless = loss <= TWCC_PATH_LOSS
1441 if lossless and sent_bps <= stand["delivered"] * TWCC_PATH_SENT:
1442 stand["quiet"].append(delay)
1443 else:
1444 stand["quiet"].clear()
1445 if not lossless:
1446 stand["lossless"] = now
1447 quiet = stand["quiet"][-TWCC_PATH_INTERVALS:]
1448 if len(quiet) == TWCC_PATH_INTERVALS and max(quiet) - min(quiet) <= band:
1449 base = min(quiet)
1450 self._twcc_stand = None
1451 elif now - stand["lossless"] >= TWCC_DELAY_FLOOR_S:
1452 base = min(d for _, d in floor)
1453 self._twcc_stand = None
1454 self._twcc_delay_base = (now, base)
1455 return base
1457 def _set_role(self, role: str) -> None:
1458 self._role = role
1460 def _set_state(self, state: State) -> None:
1461 if state != self._state:
1462 self.__log_debug("- %s -> %s", self._state, state)
1463 self._state = state
1464 self.emit("statechange")
1466 def _unregister_data_receiver(self, receiver: DataReceiver) -> None:
1467 if self._data_receiver == receiver:
1468 self._data_receiver = None
1470 def _unregister_rtp_receiver(self, receiver: RtpReceiver) -> None:
1471 self._rtp_router.unregister_receiver(receiver)
1473 def _unregister_rtp_sender(self, sender: RtpSender) -> None:
1474 self._rtp_router.unregister_sender(sender)
1476 async def _write_ssl(self) -> None:
1477 """
1478 Flush outgoing data which OpenSSL put in our BIO to the transport.
1479 """
1480 try:
1481 data = self._ssl.bio_read(1500)
1482 except SSL.Error:
1483 data = b""
1484 if data:
1485 await self.transport._send(data)
1486 self.__tx_bytes += len(data)
1487 self.__tx_packets += 1
1489 def __log_debug(self, msg: str, *args: object) -> None:
1490 logger.debug(f"RTCDtlsTransport(%s) {msg}", self._role, *args)
1492 def __log_warning(self, msg: str, *args: object) -> None:
1493 logger.warning(f"RTCDtlsTransport(%s) {msg}", self._role, *args)