Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/selkies/ice/turn.py: 25%

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. 

33 

34import asyncio 

35import hashlib 

36import logging 

37import socket 

38import ssl 

39import struct 

40import time 

41from collections.abc import Callable 

42from typing import Any, Optional, TypeVar, Union, cast 

43 

44from . import stun 

45from .utils import random_transaction_id 

46 

47logger = logging.getLogger(__name__) 

48 

49DEFAULT_CHANNEL_REFRESH_TIME = 500 

50DEFAULT_ALLOCATION_LIFETIME = 600 

51TCP_TRANSPORT = 0x06000000 

52UDP_TRANSPORT = 0x11000000 

53# Asked for both directions, the size libwebrtc gives its own UDP sockets; the 

54# kernel clamps it to its per-socket ceiling where that is lower. 

55UDP_SOCKET_BUFFER_SIZE = 262144 

56 

57 

58def size_udp_socket(sock: socket.socket) -> None: 

59 """Give an ICE or relay socket its receive and send buffers.""" 

60 for option in (socket.SO_RCVBUF, socket.SO_SNDBUF): 

61 sock.setsockopt(socket.SOL_SOCKET, option, UDP_SOCKET_BUFFER_SIZE) 

62 

63_ProtocolT = TypeVar("_ProtocolT", bound=asyncio.DatagramProtocol) 

64 

65 

66def is_channel_data(data: bytes) -> bool: 

67 return (data[0] & 0xC0) == 0x40 

68 

69 

70def make_integrity_key(username: str, realm: str, password: str) -> bytes: 

71 return hashlib.md5(":".join([username, realm, password]).encode("utf8")).digest() 

72 

73 

74class TurnStreamMixin: 

75 datagram_received: Callable[[bytes, Any], None] 

76 transport: asyncio.BaseTransport 

77 

78 def data_received(self, data: bytes) -> None: 

79 if not hasattr(self, "buffer"): 

80 self.buffer = b"" 

81 self.buffer += data 

82 

83 while len(self.buffer) >= 4: 

84 _, length = struct.unpack("!HH", self.buffer[0:4]) 

85 length += stun.padding_length(length) 

86 if is_channel_data(self.buffer): 

87 full_length = 4 + length 

88 else: 

89 full_length = 20 + length 

90 if len(self.buffer) < full_length: 

91 break 

92 

93 addr = self.transport.get_extra_info("peername") 

94 self.datagram_received(self.buffer[0:full_length], addr) 

95 self.buffer = self.buffer[full_length:] 

96 

97 def _padded(self, data: bytes) -> bytes: 

98 # TCP and TCP-over-TLS must pad messages to 4-byte boundaries. 

99 padding = stun.padding_length(len(data)) 

100 if padding: 

101 data += bytes(padding) 

102 return data 

103 

104 

105class TurnClientMixin: 

106 _send: Callable[[bytes], None] 

107 

108 def __init__( 

109 self, 

110 server: tuple[str, int], 

111 username: Optional[str], 

112 password: Optional[str], 

113 lifetime: int, 

114 channel_refresh_time: int, 

115 ) -> None: 

116 self.channel_refresh_at: dict[int, float] = {} 

117 self.channel_to_peer: dict[int, tuple[str, int]] = {} 

118 self.peer_connect_waiters: dict[ 

119 tuple[str, int], list[asyncio.Future[None]] 

120 ] = {} 

121 self.peer_to_channel: dict[tuple[str, int], int] = {} 

122 

123 self.channel_number = 0x4000 

124 self.channel_refresh_time = channel_refresh_time 

125 self.integrity_key: Optional[bytes] = None 

126 self.lifetime = lifetime 

127 self.nonce: Optional[bytes] = None 

128 self.password = password 

129 self.receiver: Optional[asyncio.DatagramProtocol] = None 

130 self.realm: Optional[str] = None 

131 self.refresh_task: Optional[asyncio.Task] = None 

132 self.relayed_address: Optional[tuple[str, int]] = None 

133 self.server = server 

134 self.transactions: dict[bytes, stun.Transaction] = {} 

135 self.username = username 

136 

137 async def channel_bind(self, channel_number: int, addr: tuple[str, int]) -> None: 

138 request = stun.Message( 

139 message_method=stun.Method.CHANNEL_BIND, message_class=stun.Class.REQUEST 

140 ) 

141 request.attributes["CHANNEL-NUMBER"] = channel_number 

142 request.attributes["XOR-PEER-ADDRESS"] = addr 

143 await self.request_with_retry(request) 

144 logger.info("TURN channel bound %d %s", channel_number, addr) 

145 

146 async def connect(self) -> tuple[str, int]: 

147 """ 

148 Create a TURN allocation. 

149 """ 

150 request = stun.Message( 

151 message_method=stun.Method.ALLOCATE, message_class=stun.Class.REQUEST 

152 ) 

153 request.attributes["LIFETIME"] = self.lifetime 

154 request.attributes["REQUESTED-TRANSPORT"] = UDP_TRANSPORT 

155 response, _ = await self.request_with_retry(request) 

156 

157 time_to_expiry = response.attributes["LIFETIME"] 

158 self.relayed_address = response.attributes["XOR-RELAYED-ADDRESS"] 

159 logger.info( 

160 "TURN allocation created %s (expires in %d seconds)", 

161 self.relayed_address, 

162 time_to_expiry, 

163 ) 

164 

165 # periodically refresh allocation 

166 self.refresh_task = asyncio.create_task(self.refresh(time_to_expiry)) 

167 

168 return self.relayed_address 

169 

170 def connection_lost(self, exc: Exception) -> None: 

171 logger.debug("%s connection_lost(%s)", self, exc) 

172 if self.receiver is not None: 

173 self.receiver.connection_lost(exc) 

174 

175 def connection_made(self, transport: asyncio.BaseTransport) -> None: 

176 logger.debug("%s connection_made(%s)", self, transport) 

177 self.transport = transport 

178 

179 def datagram_received(self, data: Union[bytes, str], addr: tuple[str, int]) -> None: 

180 data = cast(bytes, data) 

181 

182 # demultiplex channel data 

183 if len(data) >= 4 and is_channel_data(data): 

184 channel, length = struct.unpack("!HH", data[0:4]) 

185 

186 if len(data) >= length + 4 and self.receiver is not None: 

187 peer_address = self.channel_to_peer.get(channel) 

188 if peer_address: 

189 payload = data[4 : 4 + length] 

190 self.receiver.datagram_received(payload, peer_address) 

191 

192 return 

193 

194 try: 

195 message = stun.parse_message(data) 

196 logger.debug("%s < %s %s", self, addr, message) 

197 except ValueError: 

198 return 

199 

200 if ( 

201 message.message_class == stun.Class.RESPONSE 

202 or message.message_class == stun.Class.ERROR 

203 ) and message.transaction_id in self.transactions: 

204 transaction = self.transactions[message.transaction_id] 

205 transaction.response_received(message, addr) 

206 

207 async def delete(self) -> None: 

208 """ 

209 Delete the TURN allocation. 

210 """ 

211 if self.refresh_task: 

212 self.refresh_task.cancel() 

213 self.refresh_task = None 

214 

215 request = stun.Message( 

216 message_method=stun.Method.REFRESH, message_class=stun.Class.REQUEST 

217 ) 

218 request.attributes["LIFETIME"] = 0 

219 try: 

220 await self.request_with_retry(request) 

221 except stun.TransactionError: 

222 # we do not care, we need to shutdown 

223 pass 

224 

225 logger.info("TURN allocation deleted %s", self.relayed_address) 

226 self.transport.close() 

227 

228 async def refresh(self, time_to_expiry: int) -> None: 

229 """ 

230 Periodically refresh the TURN allocation. 

231 """ 

232 while True: 

233 await asyncio.sleep(5 / 6 * time_to_expiry) 

234 

235 request = stun.Message( 

236 message_method=stun.Method.REFRESH, message_class=stun.Class.REQUEST 

237 ) 

238 request.attributes["LIFETIME"] = self.lifetime 

239 response, _ = await self.request_with_retry(request) 

240 

241 time_to_expiry = response.attributes["LIFETIME"] 

242 logger.info( 

243 "TURN allocation refreshed %s (expires in %d seconds)", 

244 self.relayed_address, 

245 time_to_expiry, 

246 ) 

247 

248 async def request( 

249 self, request: stun.Message 

250 ) -> tuple[stun.Message, tuple[str, int]]: 

251 """ 

252 Execute a STUN transaction and return the response. 

253 """ 

254 assert request.transaction_id not in self.transactions 

255 

256 if self.integrity_key: 

257 self.__add_authentication(request) 

258 

259 transaction = stun.Transaction(request, self.server, self) 

260 self.transactions[request.transaction_id] = transaction 

261 try: 

262 return await transaction.run() 

263 finally: 

264 del self.transactions[request.transaction_id] 

265 

266 async def request_with_retry( 

267 self, request: stun.Message 

268 ) -> tuple[stun.Message, tuple[str, int]]: 

269 """ 

270 Execute a STUN transaction and return the response. 

271 

272 On recoverable errors it will retry the request. 

273 """ 

274 try: 

275 response, addr = await self.request(request) 

276 except stun.TransactionFailed as e: 

277 error_code = e.response.attributes["ERROR-CODE"][0] 

278 if ( 

279 "NONCE" in e.response.attributes 

280 and self.username is not None 

281 and self.password is not None 

282 and ( 

283 (error_code == 401 and "REALM" in e.response.attributes) # type: ignore 

284 or (error_code == 438 and self.realm is not None) 

285 ) 

286 ): 

287 # update long-term credentials 

288 self.nonce = e.response.attributes["NONCE"] 

289 if error_code == 401: 

290 self.realm = e.response.attributes["REALM"] 

291 self.integrity_key = make_integrity_key( 

292 self.username, self.realm, self.password 

293 ) 

294 

295 # retry request with authentication 

296 request.transaction_id = random_transaction_id() 

297 response, addr = await self.request(request) 

298 else: 

299 raise 

300 

301 return response, addr 

302 

303 async def send_data(self, data: bytes, addr: tuple[str, int]) -> None: 

304 """ 

305 Send data to a remote host via the TURN server. 

306 """ 

307 # if a channel is being bound for the peer, wait 

308 if addr in self.peer_connect_waiters: 

309 loop = asyncio.get_running_loop() 

310 waiter = loop.create_future() 

311 self.peer_connect_waiters[addr].append(waiter) 

312 await waiter 

313 

314 channel = self.peer_to_channel.get(addr) 

315 now = time.time() 

316 if channel is None: 

317 self.peer_connect_waiters[addr] = [] 

318 channel = self.channel_number 

319 self.channel_number += 1 

320 

321 # bind channel 

322 await self.channel_bind(channel, addr) 

323 

324 # update state 

325 self.channel_refresh_at[channel] = now + self.channel_refresh_time 

326 self.channel_to_peer[channel] = addr 

327 self.peer_to_channel[addr] = channel 

328 

329 # notify waiters 

330 for waiter in self.peer_connect_waiters.pop(addr): 

331 waiter.set_result(None) 

332 elif now > self.channel_refresh_at[channel]: 

333 # refresh channel 

334 await self.channel_bind(channel, addr) 

335 

336 # update state 

337 self.channel_refresh_at[channel] = now + self.channel_refresh_time 

338 

339 header = struct.pack("!HH", channel, len(data)) 

340 self._send(header + data) 

341 

342 def send_stun(self, message: stun.Message, addr: tuple[str, int]) -> None: 

343 """ 

344 Send a STUN message to the TURN server. 

345 """ 

346 logger.debug("%s > %s %s", self, addr, message) 

347 self._send(bytes(message)) 

348 

349 def __add_authentication(self, request: stun.Message) -> None: 

350 request.attributes["USERNAME"] = self.username 

351 request.attributes["NONCE"] = self.nonce 

352 request.attributes["REALM"] = self.realm 

353 request.add_message_integrity(self.integrity_key) 

354 

355 

356class TurnClientTcpProtocol(TurnClientMixin, TurnStreamMixin, asyncio.Protocol): 

357 """ 

358 Protocol for handling TURN over TCP. 

359 """ 

360 

361 transport: asyncio.Transport # type: ignore 

362 

363 def _send(self, data: bytes) -> None: # type: ignore 

364 self.transport.write(self._padded(data)) 

365 

366 def __repr__(self) -> str: 

367 return "turn/tcp" 

368 

369 

370class TurnClientUdpProtocol(TurnClientMixin, asyncio.DatagramProtocol): 

371 """ 

372 Protocol for handling TURN over UDP. 

373 """ 

374 

375 transport: asyncio.DatagramTransport # type: ignore 

376 

377 def _send(self, data: bytes) -> None: # type: ignore 

378 self.transport.sendto(data) 

379 

380 def __repr__(self) -> str: 

381 return "turn/udp" 

382 

383 

384TurnClientProtocol = Union[TurnClientTcpProtocol, TurnClientUdpProtocol] 

385 

386 

387class TurnTransport: 

388 """ 

389 Behaves like a Datagram transport, but uses a TURN allocation. 

390 """ 

391 

392 def __init__(self, inner_protocol: TurnClientProtocol) -> None: 

393 self.__inner_protocol = inner_protocol 

394 self.__relayed_address: Optional[tuple[str, int]] = None 

395 

396 def close(self) -> None: 

397 """ 

398 Close the transport. 

399 

400 After the TURN allocation has been deleted, the protocol's 

401 `connection_lost()` method will be called with None as its argument. 

402 """ 

403 asyncio.create_task(self.__inner_protocol.delete()) 

404 

405 def get_extra_info(self, name: str, default: Any = None) -> Any: 

406 """ 

407 Return optional transport information. 

408 

409 - `'related_address'`: the related address 

410 - `'sockname'`: the relayed address 

411 """ 

412 if name == "related_address": 

413 return self.__inner_protocol.transport.get_extra_info("sockname") 

414 elif name == "sockname": 

415 return self.__relayed_address 

416 return default 

417 

418 def sendto(self, data: bytes, addr: tuple[str, int]) -> None: 

419 """ 

420 Sends the `data` bytes to the remote peer given `addr`. 

421 

422 This will bind a TURN channel as necessary. 

423 """ 

424 asyncio.create_task(self.__inner_protocol.send_data(data, addr)) 

425 

426 async def _connect(self, protocol: asyncio.DatagramProtocol) -> None: 

427 self.__relayed_address = await self.__inner_protocol.connect() 

428 

429 # Once the allocation has succeeded, notify the protocol 

430 # and start relaying received data. 

431 self.__inner_protocol.receiver = protocol 

432 protocol.connection_made(cast(asyncio.DatagramTransport, self)) 

433 

434 

435async def create_turn_endpoint( 

436 protocol_factory: Callable[[], _ProtocolT], 

437 server_addr: tuple[str, int], 

438 username: Optional[str], 

439 password: Optional[str], 

440 lifetime: int = DEFAULT_ALLOCATION_LIFETIME, 

441 channel_refresh_time: int = DEFAULT_CHANNEL_REFRESH_TIME, 

442 ssl: Optional[Union[bool, ssl.SSLContext]] = None, 

443 transport: str = "udp", 

444) -> tuple[TurnTransport, _ProtocolT]: 

445 """ 

446 Create datagram connection relayed over TURN. 

447 """ 

448 loop = asyncio.get_running_loop() 

449 inner_protocol: TurnClientProtocol 

450 inner_transport: asyncio.BaseTransport 

451 if transport == "tcp": 

452 inner_transport, inner_protocol = await loop.create_connection( 

453 lambda: TurnClientTcpProtocol( 

454 server_addr, 

455 username=username, 

456 password=password, 

457 lifetime=lifetime, 

458 channel_refresh_time=channel_refresh_time, 

459 ), 

460 host=server_addr[0], 

461 port=server_addr[1], 

462 ssl=ssl, 

463 ) 

464 else: 

465 inner_transport, inner_protocol = await loop.create_datagram_endpoint( 

466 lambda: TurnClientUdpProtocol( 

467 server_addr, 

468 username=username, 

469 password=password, 

470 lifetime=lifetime, 

471 channel_refresh_time=channel_refresh_time, 

472 ), 

473 remote_addr=server_addr, 

474 ) 

475 sock = inner_transport.get_extra_info("socket") 

476 if sock is not None: 

477 size_udp_socket(sock) 

478 

479 try: 

480 protocol = protocol_factory() 

481 turn_transport = TurnTransport(inner_protocol) 

482 await turn_transport._connect(protocol) 

483 except Exception: 

484 inner_transport.close() 

485 raise 

486 

487 return turn_transport, protocol