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

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 binascii 

36import enum 

37import hmac 

38import ipaddress 

39from collections.abc import Callable 

40from struct import pack, unpack, error as StructError 

41from typing import Optional, Protocol 

42 

43from .utils import random_transaction_id 

44 

45COOKIE = 0x2112A442 

46FINGERPRINT_LENGTH = 8 

47FINGERPRINT_XOR = 0x5354554E 

48HEADER_LENGTH = 20 

49INTEGRITY_LENGTH = 24 

50IPV4_PROTOCOL = 1 

51IPV6_PROTOCOL = 2 

52 

53RETRY_MAX = 6 

54RETRY_RTO = 0.5 

55 

56 

57def set_body_length(data: bytes, length: int) -> bytes: 

58 return data[0:2] + pack("!H", length) + data[4:] 

59 

60 

61def message_fingerprint(data: bytes) -> int: 

62 check_data = set_body_length(data, len(data) - HEADER_LENGTH + FINGERPRINT_LENGTH) 

63 return binascii.crc32(check_data) ^ FINGERPRINT_XOR 

64 

65 

66def message_integrity(data: bytes, key: bytes) -> bytes: 

67 check_data = set_body_length(data, len(data) - HEADER_LENGTH + INTEGRITY_LENGTH) 

68 return hmac.new(key, check_data, "sha1").digest() 

69 

70 

71def xor_data(data: bytes, transaction_id: bytes) -> bytes: 

72 xpad = pack("!HI", COOKIE >> 16, COOKIE) + transaction_id 

73 xdata = b"" 

74 for i in range(len(data)): 

75 xdata += int.to_bytes(data[i] ^ xpad[i], 1, "big", signed=False) 

76 return xdata 

77 

78 

79def pack_address( 

80 value: tuple[str, int], 

81 xor: Callable[[bytes], bytes] = lambda x: x, 

82) -> bytes: 

83 ip_address = ipaddress.ip_address(value[0]) 

84 if isinstance(ip_address, ipaddress.IPv4Address): 

85 protocol = IPV4_PROTOCOL 

86 else: 

87 protocol = IPV6_PROTOCOL 

88 return pack("!BB", 0, protocol) + xor(pack("!H", value[1]) + ip_address.packed) 

89 

90 

91def pack_bytes(value: bytes) -> bytes: 

92 return value 

93 

94 

95def pack_error_code(value: tuple[int, str]) -> bytes: 

96 return pack("!HBB", 0, value[0] // 100, value[0] % 100) + value[1].encode("utf8") 

97 

98 

99def pack_none(value: None) -> bytes: 

100 return b"" 

101 

102 

103def pack_string(value: str) -> bytes: 

104 return value.encode("utf8") 

105 

106 

107def pack_unsigned(value: int) -> bytes: 

108 return pack("!I", value) 

109 

110 

111def pack_unsigned_short(value: int) -> bytes: 

112 return pack("!H", value) + b"\x00\x00" 

113 

114 

115def pack_unsigned_64(value: int) -> bytes: 

116 return pack("!Q", value) 

117 

118 

119def pack_xor_address(value: tuple[str, int], transaction_id: bytes) -> bytes: 

120 return pack_address(value, lambda x: xor_data(x, transaction_id)) 

121 

122 

123def unpack_address( 

124 data: bytes, 

125 xor: Callable[[bytes], bytes] = lambda x: x, 

126) -> tuple[str, int]: 

127 if len(data) < 4: 

128 raise ValueError("STUN address length is less than 4 bytes") 

129 

130 # The first two bytes are a reserved byte and the protocol family. 

131 reserved, protocol = unpack("!BB", data[0:2]) 

132 

133 # The remaining data represents a port and an address. 

134 port_address = data[2:] 

135 if protocol == IPV4_PROTOCOL: 

136 # For IPv4 we expect 2 bytes for the port, 4 bytes for the address. 

137 if len(port_address) != 6: 

138 raise ValueError("STUN address has invalid length for IPv4") 

139 port_address = xor(port_address) 

140 return ( 

141 str(ipaddress.IPv4Address(port_address[2:])), 

142 unpack("!H", port_address[0:2])[0], 

143 ) 

144 elif protocol == IPV6_PROTOCOL: 

145 # For IPv6 we expect 2 bytes for the port, 16 bytes for the address. 

146 if len(port_address) != 18: 

147 raise ValueError("STUN address has invalid length for IPv6") 

148 port_address = xor(port_address) 

149 return ( 

150 str(ipaddress.IPv6Address(port_address[2:])), 

151 unpack("!H", port_address[0:2])[0], 

152 ) 

153 else: 

154 raise ValueError("STUN address has unknown protocol") 

155 

156 

157def unpack_xor_address(data: bytes, transaction_id: bytes) -> tuple[str, int]: 

158 return unpack_address(data, lambda x: xor_data(x, transaction_id)) 

159 

160 

161def unpack_bytes(data: bytes) -> bytes: 

162 return data 

163 

164 

165def unpack_error_code(data: bytes) -> tuple[int, str]: 

166 if len(data) < 4: 

167 raise ValueError("STUN error code is less than 4 bytes") 

168 reserved, code_high, code_low = unpack("!HBB", data[0:4]) 

169 reason = data[4:].decode("utf8") 

170 return (code_high * 100 + code_low, reason) 

171 

172 

173def unpack_none(data: bytes) -> None: 

174 return None 

175 

176 

177def unpack_string(data: bytes) -> str: 

178 return data.decode("utf8") 

179 

180 

181def unpack_unsigned(data: bytes) -> int: 

182 return unpack("!I", data)[0] 

183 

184 

185def unpack_unsigned_short(data: bytes) -> int: 

186 return unpack("!H", data[0:2])[0] 

187 

188 

189def unpack_unsigned_64(data: bytes) -> int: 

190 return unpack("!Q", data)[0] 

191 

192 

193AttributeEntry = tuple[int, str, Callable, Callable] 

194 

195ATTRIBUTES: list[AttributeEntry] = [ 

196 (0x0001, "MAPPED-ADDRESS", pack_address, unpack_address), 

197 (0x0003, "CHANGE-REQUEST", pack_unsigned, unpack_unsigned), 

198 (0x0004, "SOURCE-ADDRESS", pack_address, unpack_address), 

199 (0x0005, "CHANGED-ADDRESS", pack_address, unpack_address), 

200 (0x0006, "USERNAME", pack_string, unpack_string), 

201 (0x0008, "MESSAGE-INTEGRITY", pack_bytes, unpack_bytes), 

202 (0x0009, "ERROR-CODE", pack_error_code, unpack_error_code), 

203 (0x000C, "CHANNEL-NUMBER", pack_unsigned_short, unpack_unsigned_short), 

204 (0x000D, "LIFETIME", pack_unsigned, unpack_unsigned), 

205 (0x0012, "XOR-PEER-ADDRESS", pack_xor_address, unpack_xor_address), 

206 (0x0014, "REALM", pack_string, unpack_string), 

207 (0x0015, "NONCE", pack_bytes, unpack_bytes), 

208 (0x0016, "XOR-RELAYED-ADDRESS", pack_xor_address, unpack_xor_address), 

209 (0x0019, "REQUESTED-TRANSPORT", pack_unsigned, unpack_unsigned), 

210 (0x0020, "XOR-MAPPED-ADDRESS", pack_xor_address, unpack_xor_address), 

211 (0x0024, "PRIORITY", pack_unsigned, unpack_unsigned), 

212 (0x0025, "USE-CANDIDATE", pack_none, unpack_none), 

213 (0x8022, "SOFTWARE", pack_string, unpack_string), 

214 (0x8028, "FINGERPRINT", pack_unsigned, unpack_unsigned), 

215 (0x8029, "ICE-CONTROLLED", pack_unsigned_64, unpack_unsigned_64), 

216 (0x802A, "ICE-CONTROLLING", pack_unsigned_64, unpack_unsigned_64), 

217 (0x802B, "RESPONSE-ORIGIN", pack_address, unpack_address), 

218 (0x802C, "OTHER-ADDRESS", pack_address, unpack_address), 

219] 

220 

221ATTRIBUTES_BY_TYPE: dict[int, AttributeEntry] = {} 

222ATTRIBUTES_BY_NAME: dict[str, AttributeEntry] = {} 

223for attr in ATTRIBUTES: 

224 ATTRIBUTES_BY_TYPE[attr[0]] = attr 

225 ATTRIBUTES_BY_NAME[attr[1]] = attr 

226 

227 

228class Class(enum.IntEnum): 

229 REQUEST = 0x000 

230 INDICATION = 0x010 

231 RESPONSE = 0x100 

232 ERROR = 0x110 

233 

234 

235class Method(enum.IntEnum): 

236 BINDING = 0x1 

237 SHARED_SECRET = 0x2 

238 ALLOCATE = 0x3 

239 REFRESH = 0x4 

240 SEND = 0x6 

241 DATA = 0x7 

242 CREATE_PERMISSION = 0x8 

243 CHANNEL_BIND = 0x9 

244 

245 

246class Message: 

247 def __init__( 

248 self, 

249 message_method: Method, 

250 message_class: Class, 

251 transaction_id: Optional[bytes] = None, 

252 attributes: Optional[dict] = None, 

253 ) -> None: 

254 self.message_method = message_method 

255 self.message_class = message_class 

256 self.transaction_id = transaction_id or random_transaction_id() 

257 self.attributes = attributes or {} 

258 

259 def add_message_integrity(self, key: bytes) -> None: 

260 """ 

261 Add MESSAGE-INTEGRITY and FINGERPRINT attributes to the message. 

262 

263 This must be the last step before sending out the message. 

264 """ 

265 self.attributes.pop("MESSAGE-INTEGRITY", None) 

266 self.attributes.pop("FINGERPRINT", None) 

267 self.attributes["MESSAGE-INTEGRITY"] = message_integrity(bytes(self), key) 

268 self.attributes["FINGERPRINT"] = message_fingerprint(bytes(self)) 

269 

270 def __bytes__(self) -> bytes: 

271 data = b"" 

272 for attr_name, attr_value in self.attributes.items(): 

273 attr_type, _, attr_pack, attr_unpack = ATTRIBUTES_BY_NAME[attr_name] 

274 if attr_pack == pack_xor_address: 

275 v = attr_pack(attr_value, self.transaction_id) 

276 else: 

277 v = attr_pack(attr_value) 

278 

279 attr_len = len(v) 

280 pad_len = padding_length(attr_len) 

281 data += pack("!HH", attr_type, attr_len) + v + bytes(pad_len) 

282 return ( 

283 pack( 

284 "!HHI12s", 

285 self.message_method | self.message_class, 

286 len(data), 

287 COOKIE, 

288 self.transaction_id, 

289 ) 

290 + data 

291 ) 

292 

293 def __repr__(self) -> str: 

294 return ( 

295 f"Message(message_method=Method.{self.message_method.name}, " 

296 f"message_class=Class.{self.message_class.name}, " 

297 f"transaction_id={self.transaction_id.hex()}, " 

298 f"attributes={self.attributes})" 

299 ) 

300 

301 

302class TransactionError(Exception): 

303 response: Optional[Message] = None 

304 

305 

306class TransactionFailed(TransactionError): 

307 def __init__(self, response: Message) -> None: 

308 self.response = response 

309 

310 def __str__(self) -> str: 

311 out = "STUN transaction failed" 

312 if "ERROR-CODE" in self.response.attributes: 

313 out += " (%s - %s)" % self.response.attributes["ERROR-CODE"] 

314 return out 

315 

316 

317class TransactionTimeout(TransactionError): 

318 def __str__(self) -> str: 

319 return "STUN transaction timed out" 

320 

321 

322class TransactionSender(Protocol): 

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

324 

325 

326class Transaction: 

327 def __init__( 

328 self, 

329 request: Message, 

330 addr: tuple[str, int], 

331 protocol: TransactionSender, 

332 retransmissions: Optional[int] = None, 

333 ) -> None: 

334 self.__addr = addr 

335 self.__future: asyncio.Future[tuple[Message, tuple[str, int]]] = ( 

336 asyncio.Future() 

337 ) 

338 self.__request = request 

339 self.__timeout_delay = RETRY_RTO 

340 self.__timeout_handle: Optional[asyncio.TimerHandle] = None 

341 self.__protocol = protocol 

342 self.__tries = 0 

343 self.__tries_max = 1 + ( 

344 retransmissions if retransmissions is not None else RETRY_MAX 

345 ) 

346 

347 def response_received(self, message: Message, addr: tuple[str, int]) -> None: 

348 if not self.__future.done(): 

349 if message.message_class == Class.RESPONSE: 

350 self.__future.set_result((message, addr)) 

351 else: 

352 self.__future.set_exception(TransactionFailed(message)) 

353 

354 async def run(self) -> tuple[Message, tuple[str, int]]: 

355 try: 

356 self.__retry() 

357 return await self.__future 

358 finally: 

359 if self.__timeout_handle: 

360 self.__timeout_handle.cancel() 

361 

362 def __retry(self) -> None: 

363 if self.__tries >= self.__tries_max: 

364 self.__future.set_exception(TransactionTimeout()) 

365 return 

366 

367 self.__protocol.send_stun(self.__request, self.__addr) 

368 

369 loop = asyncio.get_running_loop() 

370 self.__timeout_handle = loop.call_later(self.__timeout_delay, self.__retry) 

371 self.__timeout_delay *= 2 

372 self.__tries += 1 

373 

374 

375def padding_length(length: int) -> int: 

376 """ 

377 STUN message attributes are padded to a 4-byte boundary. 

378 """ 

379 rest = length % 4 

380 if rest == 0: 

381 return 0 

382 else: 

383 return 4 - rest 

384 

385 

386def parse_message(data: bytes, integrity_key: Optional[bytes] = None) -> Message: 

387 """ 

388 Parses a STUN message. 

389 

390 If the ``integrity_key`` parameter is given, the message's HMAC will be verified. 

391 """ 

392 if len(data) < HEADER_LENGTH: 

393 raise ValueError("STUN message length is less than 20 bytes") 

394 message_type, length, cookie, transaction_id = unpack( 

395 "!HHI12s", data[0:HEADER_LENGTH] 

396 ) 

397 if len(data) != HEADER_LENGTH + length: 

398 raise ValueError("STUN message length does not match") 

399 

400 attributes = {} 

401 pos = HEADER_LENGTH 

402 while pos <= len(data) - 4: 

403 attr_type, attr_len = unpack("!HH", data[pos : pos + 4]) 

404 v = data[pos + 4 : pos + 4 + attr_len] 

405 pad_len = padding_length(attr_len) 

406 if attr_type in ATTRIBUTES_BY_TYPE: 

407 _, attr_name, attr_pack, attr_unpack = ATTRIBUTES_BY_TYPE[attr_type] 

408 try: 

409 if attr_unpack == unpack_xor_address: 

410 attributes[attr_name] = attr_unpack(v, transaction_id=transaction_id) 

411 else: 

412 attributes[attr_name] = attr_unpack(v) 

413 except StructError as e: 

414 # A fixed-width attribute shorter than its type is a malformed message like 

415 # any other, which every receive path rejects as a ValueError. 

416 raise ValueError(f"STUN attribute {attr_name} is truncated") from e 

417 

418 if attr_name == "FINGERPRINT": 

419 if attributes[attr_name] != message_fingerprint(data[0:pos]): 

420 raise ValueError("STUN message fingerprint does not match") 

421 elif attr_name == "MESSAGE-INTEGRITY": 

422 if integrity_key is not None and attributes[ 

423 attr_name 

424 ] != message_integrity(data[0:pos], integrity_key): 

425 raise ValueError("STUN message integrity does not match") 

426 

427 pos += 4 + attr_len + pad_len 

428 

429 return Message( 

430 # An unknown method raises a `ValueError`. 

431 message_method=Method(message_type & 0x3EEF), 

432 # This cast cannot fail, as all 4 possible classes are defined. 

433 message_class=Class(message_type & 0x0110), 

434 transaction_id=transaction_id, 

435 attributes=attributes, 

436 )