Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/selkies/ice/ice.py: 17%
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) 2018-2019 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 aioice 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 copy
36import enum
37import errno
38import ipaddress
39import itertools
40import logging
41import random
42import re
43import secrets
44import socket
45import threading
46import time
47from collections.abc import Callable
48from typing import Optional, Union, cast
50import psutil
52from . import mdns, mux, stun, turn
53from .candidate import Candidate, candidate_foundation, candidate_priority
54from .utils import bind_errno, random_string
56logger = logging.getLogger(__name__)
58ICE_COMPLETED = 1
59ICE_FAILED = 2
61CONSENT_INTERVAL = 5
62# A check's retransmissions (0.5, 1 and 2 s apart) let an answer come back
63# from behind a standing queue; consent lapses only after CONSENT_TIMEOUT
64# without one (RFC 7675).
65CONSENT_RETRANSMISSIONS = 2
66CONSENT_TIMEOUT = 30
67# Seconds a connection waits for a path only the peer can open (its nomination
68# of an ICE-lite agent, its connection to a passive TCP candidate).
69INBOUND_CHECK_TIMEOUT = 30
71# Remote candidates a connection takes, resolved or still being resolved: a browser
72# offers a handful, and each costs a pairing and a check list sort.
73MAX_REMOTE_CANDIDATES = 64
75connection_id = itertools.count()
76protocol_id = itertools.count()
78_mdns = threading.local()
81class TransportPolicy(enum.Enum):
82 ALL = 0
83 """
84 All ICE candidates will be considered.
85 """
87 RELAY = 1
88 """
89 Only ICE candidates whose IP addresses are being relayed,
90 such as those being passed through a STUN or TURN server,
91 will be considered.
92 """
95async def get_or_create_mdns_protocol(subscriber: object) -> mdns.MDnsProtocol:
96 if not hasattr(_mdns, "lock"):
97 _mdns.lock = asyncio.Lock()
98 _mdns.protocol = None
99 _mdns.subscribers = set()
100 async with _mdns.lock:
101 if _mdns.protocol is None:
102 _mdns.protocol = await mdns.create_mdns_protocol()
103 _mdns.subscribers.add(subscriber)
104 return _mdns.protocol
107async def unref_mdns_protocol(subscriber: object) -> None:
108 if hasattr(_mdns, "lock"):
109 async with _mdns.lock:
110 _mdns.subscribers.discard(subscriber)
111 if _mdns.protocol and not _mdns.subscribers:
112 await _mdns.protocol.close()
113 _mdns.protocol = None
116def candidate_pair_priority(
117 local: Candidate, remote: Candidate, ice_controlling: bool
118) -> int:
119 """
120 See RFC 5245 - 5.7.2. Computing Pair Priority and Ordering Pairs
121 """
122 G = ice_controlling and local.priority or remote.priority
123 D = ice_controlling and remote.priority or local.priority
124 return (1 << 32) * min(G, D) + 2 * max(G, D) + (G > D and 1 or 0)
127def get_host_addresses(use_ipv4: bool, use_ipv6: bool) -> list[str]:
128 """
129 Get local IP addresses.
131 An IPv6 address carries a zone (``fe80::1%eth0``) exactly when its scope id is
132 non-zero, so the zone is what marks the link-local addresses to leave out.
133 """
134 addresses = []
135 for adapter in psutil.net_if_addrs().values():
136 for addr in adapter:
137 if addr.family == socket.AF_INET and use_ipv4 and addr.address != "127.0.0.1":
138 addresses.append(addr.address)
139 elif addr.family == socket.AF_INET6 and use_ipv6:
140 ip, _, zone = addr.address.partition("%")
141 if ip != "::1" and not zone:
142 addresses.append(ip)
143 return addresses
146async def relayed_candidate(
147 component: int,
148 protocol_factory: Callable[[], "StunProtocol"],
149 turn_server: tuple[str, int],
150 turn_username: Optional[str],
151 turn_password: Optional[str],
152 turn_ssl: bool,
153 turn_transport: str,
154) -> tuple[Candidate, "StunProtocol"]:
155 """
156 Connect to a TURN server to obtain a relayed candidate.
157 """
158 # Connect to TURN server.
159 _, protocol = await turn.create_turn_endpoint(
160 protocol_factory,
161 server_addr=turn_server,
162 username=turn_username,
163 password=turn_password,
164 ssl=turn_ssl,
165 transport=turn_transport,
166 )
168 # Build relayed candidate.
169 candidate_address = protocol.transport.get_extra_info("sockname")
170 related_address = protocol.transport.get_extra_info("related_address")
171 protocol.local_candidate = Candidate(
172 foundation=candidate_foundation("relay", "udp", candidate_address[0]),
173 component=component,
174 transport="udp",
175 priority=candidate_priority(component, "relay"),
176 host=candidate_address[0],
177 port=candidate_address[1],
178 type="relay",
179 related_address=related_address[0],
180 related_port=related_address[1],
181 )
182 return protocol.local_candidate, protocol
185async def server_reflexive_candidate(
186 protocol: "StunProtocol", stun_server: tuple[str, int]
187) -> tuple[Candidate, None]:
188 """
189 Query STUN server to obtain a server-reflexive candidate.
190 """
191 # lookup address
192 loop = asyncio.get_running_loop()
193 stun_server = (
194 await loop.run_in_executor(None, socket.gethostbyname, stun_server[0]),
195 stun_server[1],
196 )
198 # perform STUN query
199 request = stun.Message(
200 message_method=stun.Method.BINDING, message_class=stun.Class.REQUEST
201 )
202 response, _ = await protocol.request(request, stun_server)
204 local_candidate = protocol.local_candidate
205 return Candidate(
206 foundation=candidate_foundation("srflx", "udp", local_candidate.host),
207 component=local_candidate.component,
208 transport=local_candidate.transport,
209 priority=candidate_priority(local_candidate.component, "srflx"),
210 host=response.attributes["XOR-MAPPED-ADDRESS"][0],
211 port=response.attributes["XOR-MAPPED-ADDRESS"][1],
212 type="srflx",
213 related_address=local_candidate.host,
214 related_port=local_candidate.port,
215 ), None
218def host_local_preference(index: int, tcptype: Optional[str] = None) -> int:
219 """Local preference of the host candidate on the `index`-th address.
221 Unique per address as RFC 8445 5.1.2.1 requires, counting down from the
222 single-address value; a TCP candidate takes the RFC 6544 4.2 form, its
223 direction preference (passive 4, active 6) below every UDP candidate.
224 """
225 if tcptype is None:
226 return 65535 - index
227 direction = {"active": 6, "passive": 4, "so": 2}[tcptype]
228 return (1 << 13) * direction + (8191 - index)
231def sort_candidate_pairs(pairs: list["CandidatePair"], ice_controlling: bool) -> None:
232 """
233 Sort a list of candidate pairs.
234 """
236 def pair_priority(pair: CandidatePair) -> int:
237 return -candidate_pair_priority(
238 pair.local_candidate, pair.remote_candidate, ice_controlling
239 )
241 pairs.sort(key=pair_priority)
244def validate_password(value: str) -> None:
245 """
246 Check the password is well-formed.
248 See RFC 5245 - 15.4. "ice-ufrag" and "ice-pwd" Attributes
249 """
250 if not re.match("^[a-zA-Z0-9+/]{22,256}$", value):
251 raise ValueError("Password must satisfy 22*256ice-char")
254def validate_remote_candidate(candidate: Candidate) -> Candidate:
255 """
256 Check the remote candidate is supported.
257 """
258 if candidate.type not in ["host", "relay", "srflx"]:
259 raise ValueError('Unexpected candidate type "%s"' % candidate.type)
260 ipaddress.ip_address(candidate.host)
261 return candidate
264def validate_username(value: str) -> None:
265 """
266 Check the username is well-formed.
268 See RFC 5245 - 15.4. "ice-ufrag" and "ice-pwd" Attributes
269 """
270 if not re.match("^[a-zA-Z0-9+/]{4,256}$", value):
271 raise ValueError("Username must satisfy 4*256ice-char")
274class CandidatePair:
275 def __init__(self, protocol: "StunProtocol", remote_candidate: Candidate) -> None:
276 self.task: Optional[asyncio.Task] = None
277 self.nominated = False
278 self.protocol = protocol
279 self.remote_candidate = remote_candidate
280 self.remote_nominated = False
281 self.state = CandidatePair.State.FROZEN
283 def __repr__(self) -> str:
284 return "CandidatePair(%s -> %s)" % (self.local_addr, self.remote_addr)
286 @property
287 def component(self) -> int:
288 return self.local_candidate.component
290 @property
291 def local_addr(self) -> tuple[str, int]:
292 return (self.local_candidate.host, self.local_candidate.port)
294 @property
295 def local_candidate(self) -> Candidate:
296 return self.protocol.local_candidate
298 @property
299 def remote_addr(self) -> tuple[str, int]:
300 return (self.remote_candidate.host, self.remote_candidate.port)
302 class State(enum.Enum):
303 FROZEN = 0
304 WAITING = 1
305 IN_PROGRESS = 2
306 SUCCEEDED = 3
307 FAILED = 4
310class StunProtocol(asyncio.DatagramProtocol):
311 def __init__(self, receiver: "Connection") -> None:
312 self.__closed: asyncio.Future[bool] = asyncio.Future()
313 self.id = next(protocol_id)
314 self.local_candidate: Optional[Candidate] = None
315 self.receiver = receiver
316 self.transport: Optional[asyncio.DatagramTransport] = None
317 self.transactions: dict[bytes, stun.Transaction] = {}
319 def connection_lost(self, exc: Exception) -> None: # type: ignore
320 self.__log_debug("connection_lost(%s)", exc)
321 if not self.__closed.done():
322 self.receiver.data_received(None, None)
323 self.__closed.set_result(True)
325 def connection_made(self, transport: asyncio.BaseTransport) -> None:
326 self.__log_debug("connection_made(%s)", transport)
327 self.transport = cast(asyncio.DatagramTransport, transport)
329 def datagram_received(self, data: Union[bytes, str], addr: tuple) -> None:
330 # force IPv6 four-tuple to a two-tuple
331 addr = (addr[0], addr[1])
332 data = cast(bytes, data)
333 if not data:
334 return
336 try:
337 message = stun.parse_message(data)
338 self.__log_debug("< %s %s", addr, message)
339 except ValueError:
340 self.receiver.data_received(data, self.local_candidate.component)
341 return
343 if (
344 message.message_class == stun.Class.RESPONSE
345 or message.message_class == stun.Class.ERROR
346 ) and message.transaction_id in self.transactions:
347 transaction = self.transactions[message.transaction_id]
348 transaction.response_received(message, addr)
349 elif message.message_class == stun.Class.REQUEST:
350 self.receiver.request_received(message, addr, self, data)
352 def error_received(self, exc: Exception) -> None:
353 self.__log_debug("error_received(%s)", exc)
355 # custom
357 async def close(self) -> None:
358 self.transport.close()
359 await self.__closed
361 async def request(
362 self,
363 request: stun.Message,
364 addr: tuple[str, int],
365 integrity_key: Optional[bytes] = None,
366 retransmissions: Optional[int] = None,
367 ) -> tuple[stun.Message, tuple[str, int]]:
368 """
369 Execute a STUN transaction and return the response.
370 """
371 assert request.transaction_id not in self.transactions
373 if integrity_key is not None:
374 request.add_message_integrity(integrity_key)
376 transaction = stun.Transaction(
377 request, addr, self, retransmissions=retransmissions
378 )
379 self.transactions[request.transaction_id] = transaction
380 try:
381 return await transaction.run()
382 finally:
383 del self.transactions[request.transaction_id]
385 async def send_data(self, data: bytes, addr: tuple[str, int]) -> None:
386 self.transport.sendto(data, addr)
388 def send_stun(self, message: stun.Message, addr: tuple[str, int]) -> None:
389 """
390 Send a STUN message.
391 """
392 self.__log_debug("> %s %s", addr, message)
393 self.transport.sendto(bytes(message), addr)
395 def __log_debug(self, msg: str, *args: object) -> None:
396 logger.debug("%s %s " + msg, self.receiver, self, *args)
398 def __repr__(self) -> str:
399 return "protocol(%s)" % self.id
402class MuxedStunProtocol(StunProtocol):
403 """A connection's protocol on a shared socket or listener (`mux`).
405 The mux hands it a transport of its own and routes the answers to the
406 STUN requests it sends back by transaction id; an ICE check's answer also
407 teaches the mux the peer's address, a STUN server's does not.
408 """
410 async def request(
411 self,
412 request: stun.Message,
413 addr: tuple[str, int],
414 integrity_key: Optional[bytes] = None,
415 retransmissions: Optional[int] = None,
416 ) -> tuple[stun.Message, tuple[str, int]]:
417 self.transport.expect_response(
418 request.transaction_id, "USERNAME" in request.attributes
419 )
420 try:
421 return await super().request(
422 request, addr, integrity_key=integrity_key, retransmissions=retransmissions
423 )
424 finally:
425 self.transport.forget_response(request.transaction_id)
427 def peer_authenticated(self, addr: tuple) -> None:
428 """A check from `addr` carried this connection's credentials: the mux
429 routes the address's datagrams here from now on."""
430 self.transport.learn_peer(addr)
432 def __repr__(self) -> str:
433 return "protocol(%s, muxed)" % self.id
436class ConnectionEvent:
437 pass
440class ConnectionClosed(ConnectionEvent):
441 pass
444class Connection:
445 """
446 An ICE connection for a single media stream.
448 :param ice_controlling: Whether the local peer has the controlling role.
449 :param components: The number of components.
450 :param stun_server: The address of the STUN server or `None`.
451 :param turn_server: The address of the TURN server or `None`.
452 :param turn_username: The username for the TURN server.
453 :param turn_password: The password for the TURN server.
454 :param turn_ssl: Whether to use TLS for the TURN server.
455 :param turn_transport: The transport for TURN server, `"udp"` or `"tcp"`.
456 :param use_ipv4: Whether to use IPv4 candidates.
457 :param use_ipv6: Whether to use IPv6 candidates.
458 :param transport_policy: Transport policy.
459 :param local_username: An optional local username, otherwise a random one
460 will be generated.
461 :param local_password: An optional local password, otherwise a random one
462 will be generated.
463 :param nat1to1_ips: Optional public addresses (IPv4 and/or IPv6) to
464 advertise in host candidates of the matching family
465 (static 1:1 NAT), otherwise host candidates use the
466 bound address.
467 :param port_range: Optional inclusive `(min, max)` UDP port window the
468 host candidate sockets bind into.
469 :param udp_mux: Optional `mux.UdpMux` whose shared sockets carry the UDP
470 host candidates instead of sockets of their own; the port
471 range is then unused.
472 :param tcp_mux: Optional `mux.TcpMux` whose shared listeners add a passive
473 ICE-TCP host candidate per address.
474 :param ice_lite: Run as an ICE-lite agent (RFC 8445 appendix A): host
475 candidates only, always the controlled role, answering
476 the peer's connectivity checks instead of sending any.
477 """
479 def __init__(
480 self,
481 ice_controlling: bool,
482 components: int = 1,
483 stun_server: Optional[tuple[str, int]] = None,
484 turn_server: Optional[tuple[str, int]] = None,
485 turn_username: Optional[str] = None,
486 turn_password: Optional[str] = None,
487 turn_ssl: bool = False,
488 turn_transport: str = "udp",
489 use_ipv4: bool = True,
490 use_ipv6: bool = True,
491 transport_policy: TransportPolicy = TransportPolicy.ALL,
492 local_username: Optional[str] = None,
493 local_password: Optional[str] = None,
494 nat1to1_ips: Optional[list[str]] = None,
495 port_range: Optional[tuple[int, int]] = None,
496 udp_mux: Optional[mux.UdpMux] = None,
497 tcp_mux: Optional[mux.TcpMux] = None,
498 ice_lite: bool = False,
499 ) -> None:
500 if ice_lite and (stun_server is not None or turn_server is not None):
501 raise ValueError(
502 "An ICE-lite agent gathers host candidates only; it takes no STUN or TURN server."
503 )
504 self.ice_lite = ice_lite
505 self.ice_controlling = ice_controlling and not ice_lite
506 self._udp_mux = udp_mux
507 self._tcp_mux = tcp_mux
509 if local_username is None:
510 # Shared sockets tell sessions apart by this fragment alone, so it
511 # is longer than the four characters ICE requires.
512 local_username = random_string(8)
513 else:
514 validate_username(local_username)
516 if local_password is None:
517 local_password = random_string(22)
518 else:
519 validate_password(local_password)
521 #: Whether the remote party is an ICE Lite implementation.
522 self.remote_is_lite = False
523 #: Remote username, which you need to set.
524 self.remote_username: Optional[str] = None
525 #: Remote password, which you need to set.
526 self.remote_password: Optional[str] = None
528 self.stun_server = stun_server
529 self.turn_server = turn_server
530 self.turn_username = turn_username
531 self.turn_password = turn_password
532 self.turn_ssl = turn_ssl
533 self.turn_transport = turn_transport
535 # private
536 self._closed = False
537 self._components = set(range(1, components + 1))
538 self._check_list: list[CandidatePair] = []
539 self._check_list_done = False
540 self._check_list_state: asyncio.Queue = asyncio.Queue()
541 self._early_checks: list[
542 tuple[stun.Message, tuple[str, int], StunProtocol]
543 ] = []
544 self._early_checks_done = False
545 self._event_waiter: Optional[asyncio.Future[ConnectionEvent]] = None
546 self._id = next(connection_id)
547 self._local_candidates: list[Candidate] = []
548 self._local_candidates_end = False
549 self._local_candidates_start = False
550 self._local_password = local_password
551 self._local_username = local_username
552 self._nominated: dict[int, CandidatePair] = {}
553 self._nominating: set[int] = set()
554 self._protocols: list[StunProtocol] = []
555 self._remote_candidates: list[Candidate] = []
556 self._remote_candidates_end = False
557 self._remote_candidates_ending = False
558 self._remote_candidate_keys: set[tuple] = set()
559 self._mdns_lookups: set[asyncio.Task] = set()
560 self._mdns_unresolved: set[str] = set()
561 self._query_consent_task: Optional[asyncio.Task] = None
562 # When the peer was last heard from: media, DTLS, SCTP, or its consent checks.
563 self.last_received = time.monotonic()
564 self._queue: asyncio.Queue[tuple[Optional[bytes], Optional[int]]] = (
565 asyncio.Queue()
566 )
567 self._tie_breaker = secrets.randbits(64)
568 self._use_ipv4 = use_ipv4
569 self._use_ipv6 = use_ipv6
571 # Host-candidate UDP confinement: a scheduler-managed deployment hands
572 # a session a small firewalled port window, so the sockets behind
573 # direct host candidates must bind inside it. Bounds below 1024 are
574 # rejected rather than clamped; STUN/TURN/mDNS sockets stay
575 # unconfined, and the TURN relay keeps its own separate
576 # TURN_MIN_PORT/TURN_MAX_PORT.
577 if port_range is not None:
578 range_min, range_max = port_range
579 if not 1024 <= range_min <= range_max <= 65535:
580 raise ValueError(
581 "port_range must satisfy 1024 <= min <= max <= 65535, "
582 f"got {port_range!r}"
583 )
584 self._port_range = port_range
586 # NAT1TO1: validated public addresses keyed by IP version. A host
587 # candidate of a configured family advertises the mapped address in
588 # place of its bound private one; unconfigured families are unchanged.
589 self._nat1to1_ips: dict[int, str] = {}
590 for raw_ip in nat1to1_ips or []:
591 try:
592 addr = ipaddress.ip_address(raw_ip.strip())
593 except ValueError:
594 continue
595 self._nat1to1_ips[addr.version] = str(addr)
597 if (
598 stun_server is None
599 and turn_server is None
600 and transport_policy == TransportPolicy.RELAY
601 ):
602 raise ValueError(
603 "Relay transport policy requires a STUN and/or TURN server."
604 )
606 self._transport_policy = transport_policy
608 @property
609 def local_candidates(self) -> list[Candidate]:
610 """
611 Local candidates, automatically set by :meth:`gather_candidates`.
612 """
613 return self._local_candidates[:]
615 @property
616 def local_password(self) -> str:
617 """
618 Local password, set at construction time.
619 """
620 return self._local_password
622 @property
623 def local_username(self) -> str:
624 """
625 Local username, set at construction time.
626 """
627 return self._local_username
629 @property
630 def remote_candidates(self) -> list[Candidate]:
631 """
632 Remote candidates, which you need to populate using
633 :meth:`add_remote_candidate`.
634 """
635 return self._remote_candidates[:]
637 async def add_remote_candidate(self, remote_candidate: Optional[Candidate]) -> None:
638 """
639 Add a remote candidate or signal end-of-candidates.
641 To signal end-of-candidates, pass `None`.
643 :param remote_candidate: A :class:`Candidate` instance or `None`.
644 """
645 if self._remote_candidates_end or self._remote_candidates_ending:
646 raise ValueError("Cannot add remote candidate after end-of-candidates.")
648 # end-of-candidates, once the mDNS names still being resolved are settled
649 if remote_candidate is None:
650 if self._mdns_lookups:
651 self._remote_candidates_ending = True
652 asyncio.ensure_future(self._end_after_lookups())
653 else:
654 self._end_remote_candidates()
655 return
657 key = (remote_candidate.component, remote_candidate.transport.lower(),
658 remote_candidate.host, remote_candidate.port, remote_candidate.type,
659 remote_candidate.tcptype)
660 if key in self._remote_candidate_keys:
661 return
662 if len(self._remote_candidate_keys) >= MAX_REMOTE_CANDIDATES:
663 self.__log_info(
664 f'Remote candidate "{remote_candidate.host}" dropped: '
665 f"{MAX_REMOTE_CANDIDATES} remote candidates taken"
666 )
667 return
668 self._remote_candidate_keys.add(key)
670 # resolve mDNS candidate, off the caller's await: the signaling that hands
671 # candidates over is shared, and an unanswered name holds a lookup a second
672 if mdns.is_mdns_hostname(remote_candidate.host):
673 if remote_candidate.host in self._mdns_unresolved:
674 return
675 lookup = asyncio.ensure_future(self._resolve_remote_candidate(remote_candidate))
676 self._mdns_lookups.add(lookup)
677 lookup.add_done_callback(self._mdns_lookups.discard)
678 return
680 self._take_remote_candidate(remote_candidate)
682 async def _resolve_remote_candidate(self, remote_candidate: Candidate) -> None:
683 """
684 Resolve an mDNS remote candidate and take it, remembering a name that
685 did not resolve so a repeat of it costs no second lookup.
686 """
687 try:
688 mdns_protocol = await get_or_create_mdns_protocol(self)
689 remote_addr = await mdns_protocol.resolve(remote_candidate.host)
690 if remote_addr is None:
691 self._mdns_unresolved.add(remote_candidate.host)
692 self.__log_info(
693 f'Remote candidate "{remote_candidate.host}" could not be resolved'
694 )
695 return
696 self.__log_info(
697 f'Remote candidate "{remote_candidate.host}" resolved to {remote_addr}'
698 )
699 if not self._closed:
700 copy_candidate = copy.copy(remote_candidate)
701 copy_candidate.host = remote_addr
702 self._take_remote_candidate(copy_candidate)
703 finally:
704 self._mdns_lookups.discard(asyncio.current_task())
705 if not self._mdns_lookups and self._check_list:
706 self._check_exhausted()
708 async def _end_after_lookups(self) -> None:
709 """
710 Signal end-of-candidates once every mDNS lookup still out has settled.
711 """
712 await asyncio.gather(*list(self._mdns_lookups), return_exceptions=True)
713 self._end_remote_candidates()
715 def _end_remote_candidates(self) -> None:
716 self._prune_components()
717 self._remote_candidates_end = True
719 def _take_remote_candidate(self, remote_candidate: Candidate) -> None:
720 """
721 Validate a resolved remote candidate and pair it with every local one.
722 """
723 try:
724 validate_remote_candidate(remote_candidate)
725 except ValueError as e:
726 self.__log_info(
727 f'Remote candidate "{remote_candidate.host}" is not valid: {e}'
728 )
729 return
730 self._remote_candidates.append(remote_candidate)
732 # pair the remote candidate
733 for protocol in self._protocols:
734 if self._pairs_with(
735 protocol, remote_candidate
736 ) and not self._find_pair(protocol, remote_candidate):
737 pair = CandidatePair(protocol, remote_candidate)
738 self._check_list.append(pair)
739 self.sort_check_list()
741 async def gather_candidates(self) -> None:
742 """
743 Gather local candidates.
745 You **must** call this coroutine before calling :meth:`connect`.
746 """
747 if not self._local_candidates_start:
748 self._local_candidates_start = True
749 addresses = get_host_addresses(
750 use_ipv4=self._use_ipv4, use_ipv6=self._use_ipv6
751 )
752 coros = [
753 self.get_component_candidates(component=component, addresses=addresses)
754 for component in self._components
755 ]
756 for candidates in await asyncio.gather(*coros):
757 self._local_candidates += candidates
758 self._local_candidates_end = True
760 def get_default_candidate(self, component: int) -> Optional[Candidate]:
761 """
762 Get the default local candidate for the specified component.
764 :param component: The component whose default candidate is requested.
765 """
766 for candidate in sorted(self._local_candidates, key=lambda x: x.priority):
767 if candidate.component == component:
768 return candidate
769 return None
771 async def connect(self) -> None:
772 """
773 Perform ICE handshake.
775 This coroutine returns if a candidate pair was successfully nominated
776 and raises an exception otherwise.
777 """
778 if not self._local_candidates_end:
779 raise ConnectionError("Local candidates gathering was not performed")
781 if self.remote_username is None or self.remote_password is None:
782 raise ConnectionError("Remote username or password is missing")
784 # 5.7.1. Forming Candidate Pairs
785 for remote_candidate in self._remote_candidates:
786 for protocol in self._protocols:
787 if self._pairs_with(
788 protocol, remote_candidate
789 ) and not self._find_pair(protocol, remote_candidate):
790 pair = CandidatePair(protocol, remote_candidate)
791 self._check_list.append(pair)
792 self.sort_check_list()
794 if not self.ice_lite:
795 self._unfreeze_initial()
797 # handle early checks
798 for early_check in self._early_checks:
799 self.check_incoming(*early_check)
800 self._early_checks = []
801 self._early_checks_done = True
803 # perform checks; an ICE-lite agent only answers the peer's
804 if not self.ice_lite:
805 while True:
806 if not self.check_periodic():
807 break
808 await asyncio.sleep(0.02)
810 # wait for completion; a path only the peer can open is waited for
811 # within a bound, since no local check ever fails it
812 if self._check_list or self._awaits_inbound_checks():
813 timeout = INBOUND_CHECK_TIMEOUT if self._awaits_inbound_checks() else None
814 try:
815 res = await asyncio.wait_for(self._check_list_state.get(), timeout)
816 except asyncio.TimeoutError:
817 self.__log_info(
818 "ICE failed: no check from the peer completed within %ss",
819 INBOUND_CHECK_TIMEOUT,
820 )
821 self._check_list_done = True
822 res = ICE_FAILED
823 else:
824 res = ICE_FAILED
826 # cancel remaining checks
827 for check in self._check_list:
828 if check.task:
829 check.task.cancel()
831 if res != ICE_COMPLETED:
832 raise ConnectionError("ICE negotiation failed")
834 # start consent freshness tests
835 self._query_consent_task = asyncio.create_task(self.query_consent())
837 async def close(self) -> None:
838 """
839 Close the connection.
840 """
841 # stop consent freshness tests
842 if self._query_consent_task and not self._query_consent_task.done():
843 self._query_consent_task.cancel()
844 try:
845 await self._query_consent_task
846 except asyncio.CancelledError:
847 pass
849 # stop check list, and a connect() awaiting the peer's checks
850 if not self._check_list_done:
851 self._check_list_state.put_nowait(ICE_FAILED)
852 self._check_list_done = True
854 # stop mDNS lookups, then unreference mDNS
855 for lookup in list(self._mdns_lookups):
856 lookup.cancel()
857 await unref_mdns_protocol(self)
859 self._nominated.clear()
860 for protocol in self._protocols:
861 await protocol.close()
862 self._protocols.clear()
863 self._local_candidates.clear()
865 # emit event
866 if not self._closed:
867 self._emit_event(ConnectionClosed())
868 self._closed = True
870 async def get_event(self) -> Optional[ConnectionEvent]:
871 """
872 Return the next `ConnectionEvent` or `None` if the connection is
873 already closed.
875 This method may only be called once at a time.
876 """
877 assert self._event_waiter is None, "already awaiting event"
878 if self._closed:
879 return None
880 loop = asyncio.get_running_loop()
881 self._event_waiter = loop.create_future()
882 return await asyncio.shield(self._event_waiter)
884 async def recv(self) -> bytes:
885 """
886 Receive the next datagram.
888 The return value is a `bytes` object representing the data received.
890 If the connection is not established, a `ConnectionError` is raised.
891 """
892 data, component = await self.recvfrom()
893 return data
895 async def recvfrom(self) -> tuple[bytes, int]:
896 """
897 Receive the next datagram.
899 The return value is a `(bytes, component)` tuple where `bytes` is a
900 bytes object representing the data received and `component` is the
901 component on which the data was received.
903 If the connection is not established, a `ConnectionError` is raised.
904 """
905 if not len(self._nominated):
906 raise ConnectionError("Cannot receive data, not connected")
908 result = await self._queue.get()
909 if result[0] is None:
910 raise ConnectionError("Connection lost while receiving data")
911 return result
913 async def send(self, data: bytes) -> None:
914 """
915 Send a datagram on the first component.
917 If the connection is not established, a `ConnectionError` is raised.
919 :param data: The data to be sent.
920 """
921 await self.sendto(data, 1)
923 async def sendto(self, data: bytes, component: int) -> None:
924 """
925 Send a datagram on the specified component.
927 If the connection is not established, a `ConnectionError` is raised.
929 :param data: The data to be sent.
930 :param component: The component on which to send the data.
931 """
932 active_pair = self._nominated.get(component)
933 if active_pair:
934 await active_pair.protocol.send_data(data, active_pair.remote_addr)
935 else:
936 raise ConnectionError("Cannot send data, not connected")
938 def set_selected_pair(
939 self, component: int, local_foundation: str, remote_foundation: str
940 ) -> None:
941 """
942 Force the selected candidate pair.
944 If the remote party does not support ICE, you should using this
945 instead of calling :meth:`connect`.
946 """
947 # find local candidate
948 protocol = None
949 for p in self._protocols:
950 if (
951 p.local_candidate.component == component
952 and p.local_candidate.foundation == local_foundation
953 ):
954 protocol = p
955 break
957 # find remote candidate
958 remote_candidate = None
959 for c in self._remote_candidates:
960 if c.component == component and c.foundation == remote_foundation:
961 remote_candidate = c
963 assert protocol and remote_candidate
964 self._nominated[component] = CandidatePair(protocol, remote_candidate)
966 # private
968 def build_request(self, pair: CandidatePair, nominate: bool) -> stun.Message:
969 tx_username = "%s:%s" % (self.remote_username, self.local_username)
970 request = stun.Message(
971 message_method=stun.Method.BINDING, message_class=stun.Class.REQUEST
972 )
973 request.attributes["USERNAME"] = tx_username
974 request.attributes["PRIORITY"] = candidate_priority(pair.component, "prflx")
975 if self.ice_controlling:
976 request.attributes["ICE-CONTROLLING"] = self._tie_breaker
977 if nominate:
978 request.attributes["USE-CANDIDATE"] = None
979 else:
980 request.attributes["ICE-CONTROLLED"] = self._tie_breaker
981 return request
983 def check_complete(self, pair: CandidatePair) -> None:
984 pair.task = None
986 if pair.state == CandidatePair.State.SUCCEEDED:
987 if pair.nominated and self._selects(pair):
988 self._nominated[pair.component] = pair
990 # 8.1.2. Updating States
991 #
992 # The agent MUST remove all Waiting and Frozen pairs in the check
993 # list and triggered check queue for the same component as the
994 # nominated pairs for that media stream.
995 for p in self._check_list:
996 if p.component == pair.component and p.state in [
997 CandidatePair.State.WAITING,
998 CandidatePair.State.FROZEN,
999 ]:
1000 self.check_state(p, CandidatePair.State.FAILED)
1002 # Once there is at least one nominated pair in the valid list for
1003 # every component of at least one media stream and the state of the
1004 # check list is Running:
1005 if len(self._nominated) == len(self._components):
1006 if not self._check_list_done:
1007 self.__log_info("ICE completed")
1008 self._check_list_state.put_nowait(ICE_COMPLETED)
1009 self._check_list_done = True
1010 return
1012 # 7.1.3.2.3. Updating Pair States
1013 for p in self._check_list:
1014 if (
1015 p.local_candidate.foundation == pair.local_candidate.foundation
1016 and p.state == CandidatePair.State.FROZEN
1017 ):
1018 self.check_state(p, CandidatePair.State.WAITING)
1020 self._check_exhausted()
1022 def _check_exhausted(self) -> None:
1023 """
1024 Fail the check list once every pair has run out, unless a pair may still
1025 come: a remote candidate whose mDNS name is still being resolved, or a
1026 path only the peer can open.
1027 """
1028 for p in self._check_list:
1029 if p.state not in [
1030 CandidatePair.State.SUCCEEDED,
1031 CandidatePair.State.FAILED,
1032 ]:
1033 return
1035 if not self.ice_controlling:
1036 for p in self._check_list:
1037 if p.state == CandidatePair.State.SUCCEEDED:
1038 return
1040 if self._mdns_lookups:
1041 return
1043 # the peer may still open a path of its own; connect() bounds the wait
1044 if self._awaits_inbound_checks():
1045 return
1047 if not self._check_list_done:
1048 self.__log_info("ICE failed")
1049 self._check_list_state.put_nowait(ICE_FAILED)
1050 self._check_list_done = True
1052 def check_incoming(
1053 self, message: stun.Message, addr: tuple[str, int], protocol: StunProtocol
1054 ) -> None:
1055 """
1056 Handle a successful incoming check.
1057 """
1058 component = protocol.local_candidate.component
1060 # find remote candidate
1061 remote_candidate = None
1062 for c in self._remote_candidates:
1063 if c.host == addr[0] and c.port == addr[1]:
1064 remote_candidate = c
1065 assert remote_candidate.component == component
1066 break
1067 if remote_candidate is None:
1068 # 7.2.1.3. Learning Peer Reflexive Candidates
1069 local_candidate = protocol.local_candidate
1070 remote_candidate = Candidate(
1071 foundation=random_string(10),
1072 component=component,
1073 transport=local_candidate.transport,
1074 priority=message.attributes["PRIORITY"],
1075 host=addr[0],
1076 port=addr[1],
1077 type="prflx",
1078 tcptype="active" if local_candidate.transport == "tcp" else None,
1079 )
1080 self._remote_candidates.append(remote_candidate)
1081 self.__log_info("Discovered peer reflexive candidate %s", remote_candidate)
1083 # find pair
1084 pair = self._find_pair(protocol, remote_candidate)
1085 if pair is None:
1086 pair = CandidatePair(protocol, remote_candidate)
1087 pair.state = CandidatePair.State.WAITING
1088 self._check_list.append(pair)
1089 self.sort_check_list()
1091 # triggered check; an ICE-lite agent sends none
1092 if not self.ice_lite and pair.state in [
1093 CandidatePair.State.WAITING,
1094 CandidatePair.State.FAILED,
1095 ]:
1096 self.check_start_task(pair)
1098 # 7.2.1.5. Updating the Nominated Flag
1099 if "USE-CANDIDATE" in message.attributes and not self.ice_controlling:
1100 pair.remote_nominated = True
1102 if self.ice_lite:
1103 # 7.3.1.5: the peer's nomination alone puts the pair in the
1104 # valid list of an ICE-lite agent
1105 if pair.state != CandidatePair.State.SUCCEEDED:
1106 self.check_state(pair, CandidatePair.State.SUCCEEDED)
1107 if not pair.nominated:
1108 pair.nominated = True
1109 self.check_complete(pair)
1110 elif pair.state == CandidatePair.State.SUCCEEDED:
1111 pair.nominated = True
1112 self.check_complete(pair)
1114 def check_periodic(self) -> bool:
1115 # find the highest-priority pair that is in the waiting state
1116 for pair in self._check_list:
1117 if pair.state == CandidatePair.State.WAITING:
1118 self.check_start_task(pair)
1119 return True
1121 # find the highest-priority pair that is in the frozen state
1122 for pair in self._check_list:
1123 if pair.state == CandidatePair.State.FROZEN:
1124 self.check_start_task(pair)
1125 return True
1127 # if we expect more candidates, keep going
1128 if not self._remote_candidates_end:
1129 return not self._check_list_done
1131 return False
1133 async def check_start(self, pair: CandidatePair) -> None:
1134 """
1135 Starts a check.
1136 """
1137 self.check_state(pair, CandidatePair.State.IN_PROGRESS)
1139 nominate = self.ice_controlling and not self.remote_is_lite
1140 request = self.build_request(pair, nominate=nominate)
1141 try:
1142 response, addr = await pair.protocol.request(
1143 request,
1144 pair.remote_addr,
1145 integrity_key=self.remote_password.encode("utf8"),
1146 )
1147 except stun.TransactionError as exc:
1148 # 7.1.3.1. Failure Cases
1149 if (
1150 exc.response
1151 and exc.response.attributes.get("ERROR-CODE", (None, None))[0] == 487
1152 ):
1153 if "ICE-CONTROLLING" in request.attributes:
1154 self.switch_role(ice_controlling=False)
1155 elif "ICE-CONTROLLED" in request.attributes:
1156 self.switch_role(ice_controlling=True)
1157 return await self.check_start(pair)
1158 else:
1159 self.check_state(pair, CandidatePair.State.FAILED)
1160 self.check_complete(pair)
1161 return
1163 # check remote address matches
1164 if addr != pair.remote_addr:
1165 self.__log_info("Check %s failed : source address mismatch", pair)
1166 self.check_state(pair, CandidatePair.State.FAILED)
1167 self.check_complete(pair)
1168 return
1170 # success
1171 if nominate or pair.remote_nominated:
1172 # nominated by aggressive nomination or the remote party
1173 pair.nominated = True
1174 elif self.ice_controlling and pair.component not in self._nominating:
1175 # perform regular nomination
1176 self.__log_info("Check %s nominating pair", pair)
1177 self._nominating.add(pair.component)
1178 request = self.build_request(pair, nominate=True)
1179 try:
1180 await pair.protocol.request(
1181 request,
1182 pair.remote_addr,
1183 integrity_key=self.remote_password.encode("utf8"),
1184 )
1185 except stun.TransactionError:
1186 self.__log_info("Check %s failed : could not nominate pair", pair)
1187 self.check_state(pair, CandidatePair.State.FAILED)
1188 self.check_complete(pair)
1189 return
1190 pair.nominated = True
1191 self.check_state(pair, CandidatePair.State.SUCCEEDED)
1192 self.check_complete(pair)
1194 def check_start_task(self, pair: CandidatePair) -> None:
1195 """
1196 Starts a check in a task, unless already started.
1197 """
1198 if pair.task is None:
1199 pair.task = asyncio.create_task(self.check_start(pair))
1201 def check_state(self, pair: CandidatePair, state: CandidatePair.State) -> None:
1202 """
1203 Updates the state of a check.
1204 """
1205 self.__log_info("Check %s %s -> %s", pair, pair.state, state)
1206 pair.state = state
1208 def _emit_event(self, event: ConnectionEvent) -> None:
1209 if self._event_waiter is not None:
1210 waiter = self._event_waiter
1211 self._event_waiter = None
1212 waiter.set_result(event)
1214 def _find_pair(
1215 self, protocol: StunProtocol, remote_candidate: Candidate
1216 ) -> Optional[CandidatePair]:
1217 """
1218 Find a candidate pair in the check list.
1219 """
1220 for pair in self._check_list:
1221 if pair.protocol == protocol and pair.remote_candidate == remote_candidate:
1222 return pair
1223 return None
1225 def _selects(self, pair: CandidatePair) -> bool:
1226 """
1227 Whether a nominated `pair` becomes the selected pair of its component.
1229 The controlled agent follows the peer's latest nomination. The
1230 controlling agent nominates every pair that succeeds, and keeps the
1231 highest-priority one: a TCP pair that succeeds after a UDP one must
1232 not move the session onto TCP.
1233 """
1234 current = self._nominated.get(pair.component)
1235 if current is None or current is pair or not self.ice_controlling:
1236 return True
1237 return candidate_pair_priority(
1238 pair.local_candidate, pair.remote_candidate, self.ice_controlling
1239 ) > candidate_pair_priority(
1240 current.local_candidate, current.remote_candidate, self.ice_controlling
1241 )
1243 def _pairs_with(self, protocol: StunProtocol, remote_candidate: Candidate) -> bool:
1244 """
1245 Whether the local candidate behind `protocol` forms a check list pair
1246 with `remote_candidate` in advance.
1248 A passive TCP candidate (RFC 6544) opens no connection of its own, so
1249 its pairs are learned from the peer's checks in `check_incoming`.
1250 """
1251 local_candidate = protocol.local_candidate
1252 if not local_candidate.can_pair_with(remote_candidate):
1253 return False
1254 return local_candidate.tcptype != "passive"
1256 def _awaits_inbound_checks(self) -> bool:
1257 """
1258 Whether a path can still be opened only by the peer: an ICE-lite agent
1259 awaits its nomination, a passive TCP candidate its connection, as long
1260 as the peer offered (or may still offer) a TCP candidate.
1261 """
1262 if self.ice_lite:
1263 return True
1264 if not any(
1265 p.local_candidate is not None and p.local_candidate.tcptype == "passive"
1266 for p in self._protocols
1267 ):
1268 return False
1269 return not self._remote_candidates_end or any(
1270 c.transport.lower() == "tcp" for c in self._remote_candidates
1271 )
1273 def peer_stream_lost(self, addr: tuple[str, int]) -> None:
1274 """
1275 A stream to `addr` closed; when the nominated pair rode it the
1276 connection ends with it instead of waiting for consent to expire.
1277 """
1278 if any(pair.remote_addr == addr for pair in self._nominated.values()):
1279 self.__log_info("Stream to %s lost, connection over", addr)
1280 self.data_received(None, None)
1282 async def get_component_candidates(
1283 self, component: int, addresses: list[str], timeout: int = 5
1284 ) -> list[Candidate]:
1285 candidates = []
1286 loop = asyncio.get_running_loop()
1288 # gather host candidates
1289 #
1290 # With a configured port range, binding is the reservation: the ports
1291 # are tried in random order and the kernel rejects occupied ones, so
1292 # concurrent sessions sharing one range spread without coordination.
1293 # On a UDP mux the shared sockets are the reservation instead.
1294 if self._port_range is not None and self._udp_mux is None:
1295 bind_ports = list(range(self._port_range[0], self._port_range[1] + 1))
1296 random.shuffle(bind_ports)
1297 else:
1298 bind_ports = [0]
1299 host_protocols = []
1300 for index, address in enumerate(addresses):
1301 # create transport
1302 protocol = None
1303 if self._udp_mux is not None:
1304 try:
1305 protocol = await self._udp_mux.attach(
1306 address, self.local_username, lambda: MuxedStunProtocol(self)
1307 )
1308 except (OSError, ValueError) as exc:
1309 self.__log_info("Could not share the UDP mux on %s - %s", address, exc)
1310 continue
1311 else:
1312 last_error: Optional[OSError] = None
1313 for bind_port in bind_ports:
1314 try:
1315 transport, protocol = await loop.create_datagram_endpoint(
1316 lambda: StunProtocol(self), local_addr=(address, bind_port)
1317 )
1318 break
1319 except OSError as exc:
1320 last_error = exc
1321 # Only an occupied port can be cured by trying another
1322 # one; anything else (address gone, permissions) would
1323 # fail across the whole window identically.
1324 if bind_errno(exc) != errno.EADDRINUSE:
1325 break
1326 if protocol is None:
1327 self.__log_info("Could not bind to %s - %s", address, last_error)
1328 continue
1329 sock = transport.get_extra_info("socket")
1330 if sock is not None:
1331 turn.size_udp_socket(sock)
1332 host_protocols.append(protocol)
1334 # add host candidate
1335 candidate_address = protocol.transport.get_extra_info("sockname")
1336 protocol.local_candidate = Candidate(
1337 foundation=candidate_foundation("host", "udp", candidate_address[0]),
1338 component=component,
1339 transport="udp",
1340 priority=candidate_priority(
1341 component, "host", host_local_preference(index)
1342 ),
1343 host=self._advertised_host(candidate_address[0]),
1344 port=candidate_address[1],
1345 type="host",
1346 )
1347 if self._transport_policy == TransportPolicy.ALL:
1348 candidates.append(protocol.local_candidate)
1349 self._protocols += host_protocols
1351 # ICE-TCP: a passive host candidate per address on the shared TCP
1352 # port; the peer opens the stream, so it takes part in no outbound
1353 # check and never carries a server-reflexive query
1354 if self._tcp_mux is not None:
1355 for index, address in enumerate(addresses):
1356 try:
1357 protocol = await self._tcp_mux.attach(
1358 address, self.local_username, lambda: MuxedStunProtocol(self)
1359 )
1360 except (OSError, ValueError) as exc:
1361 self.__log_info("Could not share the TCP mux on %s - %s", address, exc)
1362 continue
1363 candidate_address = protocol.transport.get_extra_info("sockname")
1364 protocol.local_candidate = Candidate(
1365 foundation=candidate_foundation("host", "tcp", candidate_address[0]),
1366 component=component,
1367 transport="tcp",
1368 priority=candidate_priority(
1369 component, "host", host_local_preference(index, "passive")
1370 ),
1371 host=self._advertised_host(candidate_address[0]),
1372 port=candidate_address[1],
1373 type="host",
1374 tcptype="passive",
1375 )
1376 self._protocols.append(protocol)
1377 if self._transport_policy == TransportPolicy.ALL:
1378 candidates.append(protocol.local_candidate)
1380 tasks: list[asyncio.Task[tuple[Candidate, Optional[StunProtocol]]]] = []
1382 # Query STUN server for server-reflexive candidates (IPv4 only).
1383 if self.stun_server:
1384 for protocol in host_protocols:
1385 if ipaddress.ip_address(protocol.local_candidate.host).version == 4:
1386 tasks.append(
1387 asyncio.create_task(
1388 server_reflexive_candidate(protocol, self.stun_server)
1389 )
1390 )
1392 # Connect to TURN server.
1393 if self.turn_server:
1394 tasks.append(
1395 asyncio.create_task(
1396 relayed_candidate(
1397 component=component,
1398 protocol_factory=lambda: StunProtocol(self),
1399 turn_server=self.turn_server,
1400 turn_username=self.turn_username,
1401 turn_password=self.turn_password,
1402 turn_ssl=self.turn_ssl,
1403 turn_transport=self.turn_transport,
1404 )
1405 )
1406 )
1408 # Run tasks in parallel and handle exceptions.
1409 if len(tasks):
1410 done, pending = await asyncio.wait(tasks, timeout=timeout)
1411 for task in done:
1412 if task.exception() is None:
1413 candidate, protocol = task.result()
1414 candidates.append(candidate)
1415 if protocol is not None:
1416 self._protocols.append(protocol)
1417 for task in pending:
1418 task.cancel()
1420 return candidates
1422 def _advertised_host(self, bound_address: str) -> str:
1423 """
1424 The address a host candidate bound on `bound_address` advertises.
1426 NAT1TO1: the configured public address of the candidate's IP family
1427 replaces the private one while the socket stays bound privately, so a
1428 peer behind static 1:1 NAT connects to this host directly. Foundations
1429 stay keyed on the bound address (hashed, never signalled verbatim).
1430 """
1431 return self._nat1to1_ips.get(
1432 ipaddress.ip_address(bound_address).version, bound_address
1433 )
1435 def _prune_components(self) -> None:
1436 """
1437 Remove components for which the remote party did not provide any candidates.
1439 This can only be determined after end-of-candidates.
1440 """
1441 seen_components = set(map(lambda x: x.component, self._remote_candidates))
1442 missing_components = self._components - seen_components
1443 if missing_components:
1444 self.__log_info(
1445 "Components %s have no candidate pairs" % missing_components
1446 )
1447 self._components = seen_components
1449 async def query_consent(self) -> None:
1450 """
1451 Periodically check consent (RFC 7675).
1452 """
1453 consented_at = time.monotonic()
1454 while True:
1455 # randomize between 0.8 and 1.2 times CONSENT_INTERVAL
1456 await asyncio.sleep(CONSENT_INTERVAL * (0.8 + 0.4 * random.random()))
1458 for pair in self._nominated.values():
1459 request = self.build_request(pair, nominate=False)
1460 try:
1461 await pair.protocol.request(
1462 request,
1463 pair.remote_addr,
1464 integrity_key=self.remote_password.encode("utf8"),
1465 retransmissions=CONSENT_RETRANSMISSIONS,
1466 )
1467 consented_at = time.monotonic()
1468 except stun.TransactionError:
1469 pass
1470 if time.monotonic() - consented_at >= CONSENT_TIMEOUT:
1471 self.__log_info("Consent to send expired")
1472 self._query_consent_task = None
1473 return await self.close()
1475 def data_received(self, data: Optional[bytes], component: Optional[int]) -> None:
1476 self.last_received = time.monotonic()
1477 self._queue.put_nowait((data, component))
1479 def request_received(
1480 self,
1481 message: stun.Message,
1482 addr: tuple[str, int],
1483 protocol: StunProtocol,
1484 raw_data: bytes,
1485 ) -> None:
1486 if message.message_method != stun.Method.BINDING:
1487 self.respond_error(message, addr, protocol, (400, "Bad Request"))
1488 return
1490 # authenticate request: short-term credentials need both attributes (RFC 5389 10.1.2)
1491 try:
1492 username = message.attributes.get("USERNAME")
1493 if "MESSAGE-INTEGRITY" not in message.attributes or not isinstance(username, str):
1494 raise ValueError("No message integrity or username")
1495 if username.split(":", 1)[0] != self.local_username:
1496 raise ValueError("Wrong username")
1497 stun.parse_message(
1498 raw_data, integrity_key=self.local_password.encode("utf8")
1499 )
1500 if self.remote_username is not None:
1501 rx_username = "%s:%s" % (self.local_username, self.remote_username)
1502 if message.attributes.get("USERNAME") != rx_username:
1503 raise ValueError("Wrong username")
1504 except ValueError:
1505 self.respond_error(message, addr, protocol, (400, "Bad Request"))
1506 return
1507 self.last_received = time.monotonic()
1508 peer_authenticated = getattr(protocol, "peer_authenticated", None)
1509 if peer_authenticated is not None:
1510 peer_authenticated(addr)
1512 # 7.2.1.1. Detecting and Repairing Role Conflicts
1513 if self.ice_controlling and "ICE-CONTROLLING" in message.attributes:
1514 self.__log_info("Role conflict, expected to be controlling")
1515 if self._tie_breaker >= message.attributes["ICE-CONTROLLING"]:
1516 self.respond_error(message, addr, protocol, (487, "Role Conflict"))
1517 return
1518 self.switch_role(ice_controlling=False)
1519 elif not self.ice_controlling and "ICE-CONTROLLED" in message.attributes:
1520 if self.ice_lite:
1521 # only the full agent can take the controlling role
1522 self.respond_error(message, addr, protocol, (487, "Role Conflict"))
1523 return
1524 self.__log_info("Role conflict, expected to be controlled")
1525 if self._tie_breaker < message.attributes["ICE-CONTROLLED"]:
1526 self.respond_error(message, addr, protocol, (487, "Role Conflict"))
1527 return
1528 self.switch_role(ice_controlling=True)
1530 # send binding response
1531 response = stun.Message(
1532 message_method=stun.Method.BINDING,
1533 message_class=stun.Class.RESPONSE,
1534 transaction_id=message.transaction_id,
1535 )
1536 response.attributes["XOR-MAPPED-ADDRESS"] = addr
1537 response.add_message_integrity(self.local_password.encode("utf8"))
1538 protocol.send_stun(response, addr)
1540 if not self._check_list and not self._early_checks_done:
1541 self._early_checks.append((message, addr, protocol))
1542 else:
1543 self.check_incoming(message, addr, protocol)
1545 def respond_error(
1546 self,
1547 request: stun.Message,
1548 addr: tuple[str, int],
1549 protocol: StunProtocol,
1550 error_code: tuple[int, str],
1551 ) -> None:
1552 response = stun.Message(
1553 message_method=request.message_method,
1554 message_class=stun.Class.ERROR,
1555 transaction_id=request.transaction_id,
1556 )
1557 response.attributes["ERROR-CODE"] = error_code
1558 response.add_message_integrity(self.local_password.encode("utf8"))
1559 protocol.send_stun(response, addr)
1561 def sort_check_list(self) -> None:
1562 sort_candidate_pairs(self._check_list, self.ice_controlling)
1564 def switch_role(self, ice_controlling: bool) -> None:
1565 if ice_controlling and self.ice_lite:
1566 self.__log_info("Staying controlled: an ICE-lite agent never controls")
1567 return
1568 self.__log_info(
1569 "Switching to %s role", ice_controlling and "controlling" or "controlled"
1570 )
1571 self.ice_controlling = ice_controlling
1572 self.sort_check_list()
1574 def _unfreeze_initial(self) -> None:
1575 # unfreeze first pair for the first component
1576 first_pair = None
1577 for pair in self._check_list:
1578 if pair.component == min(self._components):
1579 first_pair = pair
1580 break
1581 if first_pair is None:
1582 return
1583 if first_pair.state == CandidatePair.State.FROZEN:
1584 self.check_state(first_pair, CandidatePair.State.WAITING)
1586 # unfreeze pairs with same component but different foundations
1587 seen_foundations = set(first_pair.local_candidate.foundation)
1588 for pair in self._check_list:
1589 if (
1590 pair.component == first_pair.component
1591 and pair.local_candidate.foundation not in seen_foundations
1592 and pair.state == CandidatePair.State.FROZEN
1593 ):
1594 self.check_state(pair, CandidatePair.State.WAITING)
1595 seen_foundations.add(pair.local_candidate.foundation)
1597 def __log_info(self, msg: str, *args: object) -> None:
1598 logger.info("%s " + msg, self, *args)
1600 def __repr__(self) -> str:
1601 return "Connection(%s)" % self._id