import asyncio import functools import os import unittest from collections.abc import Callable, Coroutine from typing import Optional from unittest import mock import ifaddr from aioice import Candidate, TransportPolicy, ice, mdns, stun from .turnserver import run_turn_server from .utils import asynctest, invite_accept RUNNING_ON_CI = os.environ.get("GITHUB_ACTIONS") == "true" async def delay(coro: Callable[[], Coroutine[None, None, None]]) -> None: await asyncio.sleep(1) await coro() class ProtocolMock(ice.StunProtocol): def __init__(self, response_addr: tuple[str, int]) -> None: super().__init__(None) self.local_candidate = Candidate( foundation="some-foundation", component=1, transport="udp", priority=1234, host="1.2.3.4", port=1234, type="host", ) self.response_addr = response_addr self.sent_message: Optional[stun.Message] = None async def request( self, message: stun.Message, addr: tuple[str, int], integrity_key: Optional[bytes] = None, retransmissions: Optional[int] = None, ) -> tuple[stun.Message, tuple[str, int]]: response_message = stun.Message( message_method=message.message_method, message_class=stun.Class.RESPONSE, transaction_id=message.transaction_id, ) return (response_message, self.response_addr) def send_stun(self, message: stun.Message, addr: tuple[str, int]) -> None: self.sent_message = message class IceComponentTest(unittest.TestCase): @asynctest async def test_peer_reflexive(self) -> None: connection = ice.Connection(ice_controlling=True) connection.remote_password = "remote-password" connection.remote_username = "remote-username" protocol = ProtocolMock(response_addr=("2.3.4.5", 2345)) request = stun.Message( message_method=stun.Method.BINDING, message_class=stun.Class.REQUEST ) request.attributes["PRIORITY"] = 456789 connection.check_incoming(request, ("2.3.4.5", 2345), protocol) self.assertIsNone(protocol.sent_message) # check we have discovered a peer-reflexive candidate self.assertEqual(len(connection.remote_candidates), 1) candidate = connection.remote_candidates[0] self.assertEqual(candidate.component, 1) self.assertEqual(candidate.transport, "udp") self.assertEqual(candidate.priority, 456789) self.assertEqual(candidate.host, "2.3.4.5") self.assertEqual(candidate.port, 2345) self.assertEqual(candidate.type, "prflx") self.assertEqual(candidate.generation, None) # check a new pair was formed self.assertEqual(len(connection._check_list), 1) pair = connection._check_list[0] self.assertEqual(pair.protocol, protocol) self.assertEqual(pair.remote_candidate, candidate) # check a triggered check was scheduled self.assertIsNotNone(pair.task) await pair.task @asynctest async def test_request_with_invalid_method(self) -> None: connection = ice.Connection(ice_controlling=True) protocol = ProtocolMock(response_addr=("2.3.4.5", 2345)) request = stun.Message( message_method=stun.Method.ALLOCATE, message_class=stun.Class.REQUEST ) connection.request_received( request, ("2.3.4.5", 2345), protocol, bytes(request) ) self.assertIsNotNone(protocol.sent_message) self.assertEqual(protocol.sent_message.message_method, stun.Method.ALLOCATE) self.assertEqual(protocol.sent_message.message_class, stun.Class.ERROR) self.assertEqual( protocol.sent_message.attributes["ERROR-CODE"], (400, "Bad Request") ) @asynctest async def test_response_with_invalid_address(self) -> None: connection = ice.Connection(ice_controlling=True) connection.remote_password = "remote-password" connection.remote_username = "remote-username" protocol = ProtocolMock(response_addr=("3.4.5.6", 3456)) pair = ice.CandidatePair( protocol, Candidate( foundation="some-foundation", component=1, transport="udp", priority=2345, host="2.3.4.5", port=2345, type="host", ), ) self.assertEqual( repr(pair), "CandidatePair(('1.2.3.4', 1234) -> ('2.3.4.5', 2345))" ) await connection.check_start(pair) self.assertEqual(pair.state, ice.CandidatePair.State.FAILED) class IceConnectionTest(unittest.TestCase): def assertCandidateTypes(self, conn: ice.Connection, expected: set[str]) -> None: types = set([c.type for c in conn.local_candidates]) self.assertEqual(types, expected) def tearDown(self) -> None: ice.CONSENT_FAILURES = 6 ice.CONSENT_INTERVAL = 5 stun.RETRY_MAX = 6 async def connect_and_exchange_data( self, conn_a: ice.Connection, conn_b: ice.Connection ) -> None: try: # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) # send data a -> b await conn_a.send(b"howdee") data = await conn_b.recv() self.assertEqual(data, b"howdee") # send data b -> a await conn_b.send(b"gotcha") data = await conn_a.recv() self.assertEqual(data, b"gotcha") finally: # close await conn_a.close() await conn_b.close() @asynctest async def test_local_username_and_password(self) -> None: # No username or password. connection = ice.Connection(ice_controlling=True) self.assertEqual(len(connection.local_username), 4) self.assertEqual(len(connection.local_password), 22) # Valid username and password. connection = ice.Connection( ice_controlling=True, local_username="test+user", local_password="some+password/that+is/long+enough", ) self.assertEqual(connection.local_username, "test+user") self.assertEqual(connection.local_password, "some+password/that+is/long+enough") # Invalid username. with self.assertRaises(ValueError) as cm: ice.Connection(ice_controlling=True, local_username="a") self.assertEqual(str(cm.exception), "Username must satisfy 4*256ice-char") # Invalid password. with self.assertRaises(ValueError) as cm: ice.Connection(ice_controlling=True, local_password="aaaaaa") self.assertEqual(str(cm.exception), "Password must satisfy 22*256ice-char") @mock.patch("ifaddr.get_adapters") def test_get_host_addresses(self, mock_get_adapters: mock.MagicMock) -> None: mock_get_adapters.return_value = [ ifaddr.Adapter( ips=[ ifaddr.IP(ip="127.0.0.1", network_prefix=8, nice_name="lo"), ifaddr.IP(ip=("::1", 0, 0), network_prefix=128, nice_name="lo"), ], name="lo", nice_name="lo", ), ifaddr.Adapter( ips=[ ifaddr.IP( ip="1.2.3.4", network_prefix=24, nice_name="eth0", ), ifaddr.IP( ip=("2a02:0db8:85a3:0000:0000:8a2e:0370:7334", 0, 0), network_prefix=64, nice_name="eth0", ), ifaddr.IP( ip=("fe80::1234:5678:9abc:def0", 0, 2), network_prefix=64, nice_name="eth0", ), ], name="eth0", nice_name="eth0", ), ] # IPv4 only addresses = ice.get_host_addresses(use_ipv4=True, use_ipv6=False) self.assertEqual(addresses, ["1.2.3.4"]) # IPv6 only addresses = ice.get_host_addresses(use_ipv4=False, use_ipv6=True) self.assertEqual(addresses, ["2a02:0db8:85a3:0000:0000:8a2e:0370:7334"]) # both addresses = ice.get_host_addresses(use_ipv4=True, use_ipv6=True) self.assertEqual( addresses, ["1.2.3.4", "2a02:0db8:85a3:0000:0000:8a2e:0370:7334"] ) @asynctest async def test_close(self) -> None: conn_a = ice.Connection(ice_controlling=True) # close event, _ = await asyncio.gather(conn_a.get_event(), conn_a.close()) self.assertTrue(isinstance(event, ice.ConnectionClosed)) # no more events event = await conn_a.get_event() self.assertIsNone(event) # close again await conn_a.close() @asynctest async def test_connect(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we should only have host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) # there should be a default candidate for component 1 candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "host") # there should not be a default candidate for component 2 candidate = conn_a.get_default_candidate(2) self.assertIsNone(candidate) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_close(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # close while connecting await conn_b.close() done, pending = await asyncio.wait( [ asyncio.create_task(conn_a.connect()), asyncio.create_task(delay(conn_a.close)), ] ) for task in pending: task.cancel() self.assertEqual(len(done), 2) exceptions = [x.exception() for x in done if x.exception()] self.assertEqual(len(exceptions), 1) self.assertTrue(isinstance(exceptions[0], ConnectionError)) @asynctest async def test_connect_early_checks(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # connect await conn_a.connect() await asyncio.sleep(1) await conn_b.connect() # send data a -> b await conn_a.send(b"howdee") data = await conn_b.recv() self.assertEqual(data, b"howdee") # send data b -> a await conn_b.send(b"gotcha") data = await conn_a.recv() self.assertEqual(data, b"gotcha") # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_early_checks_2(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # both sides gather local candidates and exchange credentials await conn_a.gather_candidates() await conn_b.gather_candidates() conn_a.remote_username = conn_b.local_username conn_a.remote_password = conn_b.local_password conn_b.remote_username = conn_a.local_username conn_b.remote_password = conn_a.local_password async def connect_b() -> None: # side B receives offer and connects for candidate in conn_a.local_candidates: await conn_b.add_remote_candidate(candidate) await conn_b.add_remote_candidate(None) await conn_b.connect() # side A receives candidates for candidate in conn_b.local_candidates: await conn_a.add_remote_candidate(candidate) await conn_a.add_remote_candidate(None) # The sequence is: # - side A starts connecting immediately, but has no candidates # - side B receives candidates and connects # - side A receives candidates, and connection completes await asyncio.gather(conn_a.connect(), connect_b()) # send data a -> b await conn_a.send(b"howdee") data = await conn_b.recv() self.assertEqual(data, b"howdee") # send data b -> a await conn_b.send(b"gotcha") data = await conn_a.recv() self.assertEqual(data, b"gotcha") # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_two_components(self) -> None: conn_a = ice.Connection(ice_controlling=True, components=2) conn_b = ice.Connection(ice_controlling=False, components=2) # invite / accept await invite_accept(conn_a, conn_b) # we should only have host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) # there should be a default candidate for component 1 candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "host") # there should be a default candidate for component 2 candidate = conn_a.get_default_candidate(2) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "host") # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) self.assertEqual(conn_a._components, set([1, 2])) self.assertEqual(conn_b._components, set([1, 2])) # send data a -> b (component 1) await conn_a.sendto(b"howdee", 1) data, component = await conn_b.recvfrom() self.assertEqual(data, b"howdee") self.assertEqual(component, 1) # send data b -> a (component 1) await conn_b.sendto(b"gotcha", 1) data, component = await conn_a.recvfrom() self.assertEqual(data, b"gotcha") self.assertEqual(component, 1) # send data a -> b (component 2) await conn_a.sendto(b"howdee 2", 2) data, component = await conn_b.recvfrom() self.assertEqual(data, b"howdee 2") self.assertEqual(component, 2) # send data b -> a (component 2) await conn_b.sendto(b"gotcha 2", 2) data, component = await conn_a.recvfrom() self.assertEqual(data, b"gotcha 2") self.assertEqual(component, 2) # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_two_components_vs_one_component(self) -> None: """ It is possible that some of the local candidates won't get paired with remote candidates, and some of the remote candidates won't get paired with local candidates. This can happen if one agent doesn't include candidates for the all of the components for a media stream. If this happens, the number of components for that media stream is effectively reduced, and considered to be equal to the minimum across both agents of the maximum component ID provided by each agent across all components for the media stream. """ conn_a = ice.Connection(ice_controlling=True, components=2) conn_b = ice.Connection(ice_controlling=False, components=1) # invite / accept await invite_accept(conn_a, conn_b) self.assertTrue(len(conn_a.local_candidates) > 0) for candidate in conn_a.local_candidates: self.assertEqual(candidate.type, "host") # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) self.assertEqual(conn_a._components, set([1])) self.assertEqual(conn_b._components, set([1])) # send data a -> b (component 1) await conn_a.sendto(b"howdee", 1) data, component = await conn_b.recvfrom() self.assertEqual(data, b"howdee") self.assertEqual(component, 1) # send data b -> a (component 1) await conn_b.sendto(b"gotcha", 1) data, component = await conn_a.recvfrom() self.assertEqual(data, b"gotcha") self.assertEqual(component, 1) # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_to_ice_lite(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_a.remote_is_lite = True conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we should only have host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) # there should be a default candidate for component 1 candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "host") # there should not be a default candidate for component 2 candidate = conn_a.get_default_candidate(2) self.assertIsNone(candidate) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_to_ice_lite_nomination_fails(self) -> None: def mock_request_received( self: ice.Connection, message: stun.Message, addr: tuple[str, int], protocol: ice.StunProtocol, raw_data: bytes, ) -> None: if "USE-CANDIDATE" in message.attributes: self.respond_error(message, addr, protocol, (500, "Internal Error")) else: self.real_request_received(message, addr, protocol, raw_data) # type: ignore conn_a = ice.Connection(ice_controlling=True) conn_a.remote_is_lite = True conn_b = ice.Connection(ice_controlling=False) conn_b.real_request_received = conn_b.request_received # type: ignore conn_b.request_received = functools.partial(mock_request_received, conn_b) # type: ignore # invite / accept await invite_accept(conn_a, conn_b) # connect with self.assertRaises(ConnectionError) as cm: await asyncio.gather(conn_a.connect(), conn_b.connect()) self.assertEqual(str(cm.exception), "ICE negotiation failed") # close await conn_a.close() await conn_b.close() @unittest.skipIf(RUNNING_ON_CI, "CI lacks ipv6") @asynctest async def test_connect_ipv6(self) -> None: conn_a = ice.Connection(ice_controlling=True, use_ipv4=False, use_ipv6=True) conn_b = ice.Connection(ice_controlling=False, use_ipv4=False, use_ipv6=True) # invite / accept await invite_accept(conn_a, conn_b) self.assertTrue(len(conn_a.local_candidates) > 0) for candidate in conn_a.local_candidates: self.assertEqual(candidate.type, "host") await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_reverse_order(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # introduce a delay so that B's checks complete before A's await asyncio.gather(delay(conn_a.connect), conn_b.connect()) # send data a -> b await conn_a.send(b"howdee") data = await conn_b.recv() self.assertEqual(data, b"howdee") # send data b -> a await conn_b.send(b"gotcha") data = await conn_a.recv() self.assertEqual(data, b"gotcha") # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_invalid_password(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite await conn_a.gather_candidates() for candidate in conn_a.local_candidates: await conn_b.add_remote_candidate(candidate) await conn_b.add_remote_candidate(None) conn_b.remote_username = conn_a.local_username conn_b.remote_password = conn_a.local_password # accept await conn_b.gather_candidates() for candidate in conn_b.local_candidates: await conn_a.add_remote_candidate(candidate) await conn_a.add_remote_candidate(None) conn_a.remote_username = conn_b.local_username conn_a.remote_password = "wrong-password" # connect done, pending = await asyncio.wait( [ asyncio.create_task(conn_a.connect()), asyncio.create_task(conn_b.connect()), ], return_when=asyncio.FIRST_EXCEPTION, ) for task in pending: task.cancel() self.assertEqual(len(done), 1) self.assertTrue(isinstance(done.pop().exception(), ConnectionError)) # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_invalid_username(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite await conn_a.gather_candidates() for candidate in conn_a.local_candidates: await conn_b.add_remote_candidate(candidate) await conn_b.add_remote_candidate(None) conn_b.remote_username = conn_a.local_username conn_b.remote_password = conn_a.local_password # accept await conn_b.gather_candidates() for candidate in conn_b.local_candidates: await conn_a.add_remote_candidate(candidate) await conn_a.add_remote_candidate(None) conn_a.remote_username = "wrong-username" conn_a.remote_password = conn_b.local_password # connect done, pending = await asyncio.wait( [ asyncio.create_task(conn_a.connect()), asyncio.create_task(conn_b.connect()), ] ) for task in pending: task.cancel() self.assertEqual(len(done), 2) self.assertTrue(isinstance(done.pop().exception(), ConnectionError)) self.assertTrue(isinstance(done.pop().exception(), ConnectionError)) # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_no_gather(self) -> None: """ If local candidates gathering was not performed, connect fails. """ conn = ice.Connection(ice_controlling=True) await conn.add_remote_candidate( Candidate.from_sdp( "6815297761 1 udp 659136 1.2.3.4 31102 typ host generation 0" ) ) await conn.add_remote_candidate(None) conn.remote_username = "foo" conn.remote_password = "bar" with self.assertRaises(ConnectionError) as cm: await conn.connect() self.assertEqual( str(cm.exception), "Local candidates gathering was not performed" ) await conn.close() @asynctest async def test_connect_no_local_candidates(self) -> None: """ If local candidates gathering yielded no candidates, connect fails. """ conn = ice.Connection(ice_controlling=True) conn._local_candidates_end = True await conn.add_remote_candidate( Candidate.from_sdp( "6815297761 1 udp 659136 1.2.3.4 31102 typ host generation 0" ) ) await conn.add_remote_candidate(None) conn.remote_username = "foo" conn.remote_password = "bar" with self.assertRaises(ConnectionError) as cm: await conn.connect() self.assertEqual(str(cm.exception), "ICE negotiation failed") await conn.close() @asynctest async def test_connect_no_remote_candidates(self) -> None: """ If no remote candidates were provided, connect fails. """ conn = ice.Connection(ice_controlling=True) await conn.gather_candidates() await conn.add_remote_candidate(None) conn.remote_username = "foo" conn.remote_password = "bar" with self.assertRaises(ConnectionError) as cm: await conn.connect() self.assertEqual(str(cm.exception), "ICE negotiation failed") await conn.close() @asynctest async def test_connect_no_remote_credentials(self) -> None: """ If remote credentials have not been provided, connect fails. """ conn = ice.Connection(ice_controlling=True) await conn.gather_candidates() await conn.add_remote_candidate( Candidate.from_sdp( "6815297761 1 udp 659136 1.2.3.4 31102 typ host generation 0" ) ) await conn.add_remote_candidate(None) with self.assertRaises(ConnectionError) as cm: await conn.connect() self.assertEqual(str(cm.exception), "Remote username or password is missing") await conn.close() @asynctest async def test_connect_role_conflict_both_controlling(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=True) # set tie breaker for a deterministic outcome conn_a._tie_breaker = 1 conn_b._tie_breaker = 2 # invite / accept await invite_accept(conn_a, conn_b) # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) self.assertFalse(conn_a.ice_controlling) self.assertTrue(conn_b.ice_controlling) # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_role_conflict_both_controlled(self) -> None: conn_a = ice.Connection(ice_controlling=False) conn_b = ice.Connection(ice_controlling=False) # set tie breaker for a deterministic outcome conn_a._tie_breaker = 1 conn_b._tie_breaker = 2 # invite / accept await invite_accept(conn_a, conn_b) # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) self.assertFalse(conn_a.ice_controlling) self.assertTrue(conn_b.ice_controlling) # close await conn_a.close() await conn_b.close() @asynctest async def test_connect_timeout(self) -> None: # lower STUN retries stun.RETRY_MAX = 1 conn = ice.Connection(ice_controlling=True) await conn.gather_candidates() await conn.add_remote_candidate( Candidate.from_sdp( "6815297761 1 udp 659136 1.2.3.4 31102 typ host generation 0" ) ) await conn.add_remote_candidate(None) conn.remote_username = "foo" conn.remote_password = "bar" with self.assertRaises(ConnectionError) as cm: await conn.connect() self.assertEqual(str(cm.exception), "ICE negotiation failed") await conn.close() @asynctest async def test_connect_with_stun_server(self) -> None: async with run_turn_server() as stun_server: conn_a = ice.Connection( ice_controlling=True, stun_server=stun_server.udp_address ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we whould have both host and server-reflexive candidates self.assertCandidateTypes(conn_a, set(["host", "srflx"])) self.assertCandidateTypes(conn_b, set(["host"])) # the default candidate should be server-reflexive candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "srflx") self.assertIsNotNone(candidate.related_address) self.assertIsNotNone(candidate.related_port) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_with_stun_server_dns_lookup_error(self) -> None: conn_a = ice.Connection(ice_controlling=True, stun_server=("invalid.", 1234)) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we whould have only host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_with_stun_server_timeout(self) -> None: async with run_turn_server() as stun_server: # immediately stop turn server await stun_server.close() conn_a = ice.Connection( ice_controlling=True, stun_server=stun_server.udp_address ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we whould have only host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) await self.connect_and_exchange_data(conn_a, conn_b) @unittest.skipIf(RUNNING_ON_CI, "CI lacks ipv6") @asynctest async def test_connect_with_stun_server_ipv6(self) -> None: async with run_turn_server() as stun_server: conn_a = ice.Connection( ice_controlling=True, stun_server=stun_server.udp_address, use_ipv4=False, use_ipv6=True, ) conn_b = ice.Connection( ice_controlling=False, use_ipv4=False, use_ipv6=True ) # invite / accept await invite_accept(conn_a, conn_b) # we only want host candidates : no STUN for IPv6 self.assertTrue(len(conn_a.local_candidates) > 0) for candidate in conn_a.local_candidates: self.assertEqual(candidate.type, "host") await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_with_turn_server_tcp(self) -> None: async with run_turn_server(users={"foo": "bar"}) as turn_server: # create connections conn_a = ice.Connection( ice_controlling=True, turn_server=turn_server.tcp_address, turn_username="foo", turn_password="bar", turn_transport="tcp", ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we should have both host and relayed candidates self.assertCandidateTypes(conn_a, set(["host", "relay"])) self.assertCandidateTypes(conn_b, set(["host"])) # the default candidate should be relayed candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "relay") self.assertIsNotNone(candidate.related_address) self.assertIsNotNone(candidate.related_port) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_with_turn_server_udp(self) -> None: async with run_turn_server(users={"foo": "bar"}) as turn_server: # create connections conn_a = ice.Connection( ice_controlling=True, turn_server=turn_server.udp_address, turn_username="foo", turn_password="bar", ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we should have both host and relayed candidates self.assertCandidateTypes(conn_a, set(["host", "relay"])) self.assertCandidateTypes(conn_b, set(["host"])) # the default candidate should be relayed candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "relay") self.assertIsNotNone(candidate.related_address) self.assertIsNotNone(candidate.related_port) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_with_turn_server_udp_auth_failed(self) -> None: async with run_turn_server(users={"foo": "bar"}) as turn_server: # create connections conn_a = ice.Connection( ice_controlling=True, turn_server=turn_server.udp_address, turn_username="foo", turn_password="incorrect", ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we should only have host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) # the default candidate should be host candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "host") self.assertIsNone(candidate.related_address) self.assertIsNone(candidate.related_port) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_connect_with_turn_server_udp_timeout(self) -> None: async with run_turn_server(users={"foo": "bar"}) as turn_server: # immediately stop turn server await turn_server.close() # create connections conn_a = ice.Connection( ice_controlling=True, turn_server=turn_server.udp_address, turn_username="foo", turn_password="bar", ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we should only have host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) # the default candidate should be host candidate = conn_a.get_default_candidate(1) self.assertIsNotNone(candidate) self.assertEqual(candidate.type, "host") self.assertIsNone(candidate.related_address) self.assertIsNone(candidate.related_port) await self.connect_and_exchange_data(conn_a, conn_b) @asynctest async def test_consent_expired(self) -> None: # lower consent timer ice.CONSENT_FAILURES = 1 ice.CONSENT_INTERVAL = 1 conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) self.assertEqual(len(conn_a._nominated), 1) # let consent expire await conn_b.close() await asyncio.sleep(2) self.assertEqual(len(conn_a._nominated), 0) # close await conn_a.close() @asynctest async def test_consent_valid(self) -> None: # lower consent timer ice.CONSENT_FAILURES = 1 ice.CONSENT_INTERVAL = 1 conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) self.assertEqual(len(conn_a._nominated), 1) # check consent await asyncio.sleep(2) self.assertEqual(len(conn_a._nominated), 1) # close await conn_a.close() await conn_b.close() @asynctest async def test_set_selected_pair(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we should only have host candidates self.assertCandidateTypes(conn_a, set(["host"])) self.assertCandidateTypes(conn_b, set(["host"])) # force selected pair default_a = conn_a.get_default_candidate(1) default_b = conn_a.get_default_candidate(1) conn_a.set_selected_pair(1, default_a.foundation, default_b.foundation) conn_b.set_selected_pair(1, default_b.foundation, default_a.foundation) # send data a -> b await conn_a.send(b"howdee") data = await conn_b.recv() self.assertEqual(data, b"howdee") # send data b -> a await conn_b.send(b"gotcha") data = await conn_a.recv() self.assertEqual(data, b"gotcha") # close await conn_a.close() await conn_b.close() @asynctest async def test_recv_not_connected(self) -> None: conn_a = ice.Connection(ice_controlling=True) with self.assertRaises(ConnectionError) as cm: await conn_a.recv() self.assertEqual(str(cm.exception), "Cannot receive data, not connected") @asynctest async def test_recv_connection_lost(self) -> None: conn_a = ice.Connection(ice_controlling=True) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # connect await asyncio.gather(conn_a.connect(), conn_b.connect()) # disconnect while receiving with self.assertRaises(ConnectionError) as cm: await asyncio.gather(conn_a.recv(), delay(conn_a.close)) self.assertEqual(str(cm.exception), "Connection lost while receiving data") # close await conn_b.close() @asynctest async def test_send_not_connected(self) -> None: conn_a = ice.Connection(ice_controlling=True) with self.assertRaises(ConnectionError) as cm: await conn_a.send(b"howdee") self.assertEqual(str(cm.exception), "Cannot send data, not connected") @asynctest async def test_add_remote_candidate(self) -> None: conn_a = ice.Connection(ice_controlling=True) remote_candidate = Candidate( foundation="some-foundation", component=1, transport="udp", priority=1234, host="1.2.3.4", port=1234, type="host", ) # add candidate await conn_a.add_remote_candidate(remote_candidate) self.assertEqual(len(conn_a.remote_candidates), 1) self.assertEqual(conn_a.remote_candidates[0].host, "1.2.3.4") self.assertEqual(conn_a._remote_candidates_end, False) # end-of-candidates await conn_a.add_remote_candidate(None) self.assertEqual(len(conn_a.remote_candidates), 1) self.assertEqual(conn_a._remote_candidates_end, True) # try adding another candidate with self.assertRaises(ValueError) as cm: await conn_a.add_remote_candidate(remote_candidate) self.assertEqual( str(cm.exception), "Cannot add remote candidate after end-of-candidates." ) self.assertEqual(len(conn_a.remote_candidates), 1) self.assertEqual(conn_a._remote_candidates_end, True) @asynctest async def test_add_remote_candidate_mdns_bad(self) -> None: """ Add an mDNS candidate which cannot be resolved. """ conn_a = ice.Connection(ice_controlling=True) await conn_a.add_remote_candidate( Candidate( foundation="some-foundation", component=1, transport="udp", priority=1234, host=mdns.create_mdns_hostname(), port=1234, type="host", ) ) self.assertEqual(len(conn_a.remote_candidates), 0) self.assertEqual(conn_a._remote_candidates_end, False) # close await conn_a.close() @asynctest async def test_add_remote_candidate_mdns_good(self) -> None: """ Add an mDNS candidate which can be resolved. """ hostname = mdns.create_mdns_hostname() publisher = await mdns.create_mdns_protocol() await publisher.publish(hostname, "1.2.3.4") conn_a = ice.Connection(ice_controlling=True) await conn_a.add_remote_candidate( Candidate( foundation="some-foundation", component=1, transport="udp", priority=1234, host=hostname, port=1234, type="host", ) ) self.assertEqual(len(conn_a.remote_candidates), 1) self.assertEqual(conn_a.remote_candidates[0].host, "1.2.3.4") self.assertEqual(conn_a._remote_candidates_end, False) # close await conn_a.close() await publisher.close() @asynctest async def test_add_remote_candidate_unknown_type(self) -> None: conn_a = ice.Connection(ice_controlling=True) await conn_a.add_remote_candidate( Candidate( foundation="some-foundation", component=1, transport="udp", priority=1234, host="1.2.3.4", port=1234, type="bogus", ) ) self.assertEqual(len(conn_a.remote_candidates), 0) self.assertEqual(conn_a._remote_candidates_end, False) @mock.patch("asyncio.base_events.BaseEventLoop.create_datagram_endpoint") @asynctest async def test_gather_candidates_oserror(self, mock_create: mock.MagicMock) -> None: exc = OSError() exc.errno = 99 exc.strerror = "Cannot assign requested address" mock_create.side_effect = exc conn = ice.Connection(ice_controlling=True) await conn.gather_candidates() self.assertEqual(conn.local_candidates, []) @asynctest async def test_gather_candidates_relay_only_no_servers(self) -> None: with self.assertRaises(ValueError) as cm: ice.Connection(ice_controlling=True, transport_policy=TransportPolicy.RELAY) self.assertEqual( str(cm.exception), "Relay transport policy requires a STUN and/or TURN server.", ) @asynctest async def test_gather_candidates_relay_only_with_stun_server(self) -> None: async with run_turn_server() as stun_server: conn_a = ice.Connection( ice_controlling=True, stun_server=stun_server.udp_address, transport_policy=TransportPolicy.RELAY, ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we whould only have a server-reflexive candidate in connection a self.assertCandidateTypes(conn_a, set(["srflx"])) # close await conn_a.close() await conn_b.close() @asynctest async def test_gather_candidates_relay_only_with_turn_server(self) -> None: async with run_turn_server(users={"foo": "bar"}) as turn_server: conn_a = ice.Connection( ice_controlling=True, turn_server=turn_server.udp_address, turn_username="foo", turn_password="bar", transport_policy=TransportPolicy.RELAY, ) conn_b = ice.Connection(ice_controlling=False) # invite / accept await invite_accept(conn_a, conn_b) # we whould only have a server-reflexive candidate in connection a self.assertCandidateTypes(conn_a, set(["relay"])) # close await conn_a.close() await conn_b.close() @asynctest async def test_repr(self) -> None: conn = ice.Connection(ice_controlling=True) conn._id = 1 self.assertEqual(repr(conn), "Connection(1)") class StunProtocolTest(unittest.TestCase): @asynctest async def test_error_received(self) -> None: protocol = ice.StunProtocol(None) protocol.error_received(OSError("foo")) @asynctest async def test_repr(self) -> None: protocol = ice.StunProtocol(None) protocol.id = 1 self.assertEqual(repr(protocol), "protocol(1)")