Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/selkies/webrtc/sdp.py: 77%

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/.

5# This file incorporates work covered by the following copyright and

6# permission notice:

8# Copyright (c) Jeremy Lainé.

10#

11# Redistribution and use in source and binary forms, with or without

12# modification, are permitted provided that the following conditions are met:

13#

14# * Redistributions of source code must retain the above copyright notice,

15# this list of conditions and the following disclaimer.

16# * Redistributions in binary form must reproduce the above copyright notice,

17# this list of conditions and the following disclaimer in the documentation

18# and/or other materials provided with the distribution.

19# * Neither the name of aiortc nor the names of its contributors may

20# be used to endorse or promote products derived from this software without

21# specific prior written permission.

22#

23# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND

24# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

25# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

26# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE

27# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL

28# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR

29# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER

30# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,

31# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

32# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

34import enum

35import ipaddress

36import re

37from dataclasses import dataclass

38from typing import Any, Optional, Union

40from . import rtp

41from .rtcdtlstransport import RTCDtlsFingerprint, RTCDtlsParameters

42from .rtcicetransport import RTCIceCandidate, RTCIceParameters

43from .rtcrtpparameters import (

44 ParametersDict,

45 RTCRtcpFeedback,

46 RTCRtpCodecParameters,

47 RTCRtpHeaderExtensionParameters,

48 RTCRtpParameters,

49)

50from .rtcsctptransport import RTCSctpCapabilities

52# Most a media section may name of each, far past what a browser sends: parsing a

53# remote description, which arrives over signaling, stays linear in its length.

54MAX_CODECS = 256

55MAX_SSRCS = 512

56MAX_RTCP_FB = 512

58DIRECTIONS = ["inactive", "sendonly", "recvonly", "sendrecv"]

60DTLS_ROLE_SETUP = {"auto": "actpass", "client": "active", "server": "passive"}

61DTLS_SETUP_ROLE = dict([(v, k) for (k, v) in DTLS_ROLE_SETUP.items()])

63FMTP_INT_PARAMETERS = [

64 "apt",

65 "max-fr",

66 "max-fs",

67 "maxplaybackrate",

68 "minptime",

69 "stereo",

70 "useinbandfec",

71]

74class BitPattern:

75 def __init__(self, v: str) -> None:

76 self._mask = ~self._bytemaskstring("x", v)

77 self._masked_value = self._bytemaskstring("1", v)

79 def matches(self, v: int) -> bool:

80 return (v & self._mask) == self._masked_value

82 def _bytemaskstring(self, c: str, s: str) -> int:

83 return (

84 (s[0] == c) << 7

85 | (s[1] == c) << 6

86 | (s[2] == c) << 5

87 | (s[3] == c) << 4

88 | (s[4] == c) << 3

89 | (s[5] == c) << 2

90 | (s[6] == c) << 1

91 | (s[7] == c) << 0

92 )

95class H264Profile(enum.Enum):

96 PROFILE_CONSTRAINED_BASELINE = 0

97 PROFILE_BASELINE = 1

98 PROFILE_MAIN = 2

99 PROFILE_CONSTRAINED_HIGH = 3

100 PROFILE_HIGH = 4

101 PROFILE_PREDICTIVE_HIGH_444 = 5

102

103

104class H264Level(enum.IntEnum):

105 LEVEL1_B = -1

106 LEVEL1 = 10

107 LEVEL1_1 = 11

108 LEVEL1_2 = 12

109 LEVEL1_3 = 13

110 LEVEL2 = 20

111 LEVEL2_1 = 21

112 LEVEL2_2 = 22

113 LEVEL3 = 30

114 LEVEL3_1 = 31

115 LEVEL3_2 = 32

116 LEVEL4 = 40

117 LEVEL4_1 = 41

118 LEVEL4_2 = 42

119 LEVEL5 = 50

120 LEVEL5_1 = 51

121 LEVEL5_2 = 52

122

123

124H264_PROFILE_PATTERNS = [

125 (0x42, BitPattern("x1xx0000"), H264Profile.PROFILE_CONSTRAINED_BASELINE),

126 (0x4D, BitPattern("1xxx0000"), H264Profile.PROFILE_CONSTRAINED_BASELINE),

127 (0x58, BitPattern("11xx0000"), H264Profile.PROFILE_CONSTRAINED_BASELINE),

128 (0x42, BitPattern("x0xx0000"), H264Profile.PROFILE_BASELINE),

129 (0x58, BitPattern("10xx0000"), H264Profile.PROFILE_BASELINE),

130 (0x4D, BitPattern("0x0x0000"), H264Profile.PROFILE_MAIN),

131 (0x64, BitPattern("00000000"), H264Profile.PROFILE_HIGH),

132 (0x64, BitPattern("00001100"), H264Profile.PROFILE_CONSTRAINED_HIGH),

133 (0xF4, BitPattern("00000000"), H264Profile.PROFILE_PREDICTIVE_HIGH_444),

134]

135

136

137def candidate_from_sdp(sdp: str) -> RTCIceCandidate:

138 bits = sdp.split()

139 if len(bits) < 8:

140 raise ValueError("SDP does not have enough properties")

141

142 candidate = RTCIceCandidate(

143 component=int(bits[1]),

144 foundation=bits[0],

145 ip=bits[4],

146 port=int(bits[5]),

147 priority=int(bits[3]),

148 protocol=bits[2],

149 type=bits[7],

150 )

151

152 for i in range(8, len(bits) - 1, 2):

153 if bits[i] == "raddr":

154 candidate.relatedAddress = bits[i + 1]

155 elif bits[i] == "rport":

156 candidate.relatedPort = int(bits[i + 1])

157 elif bits[i] == "tcptype":

158 candidate.tcpType = bits[i + 1]

159

160 return candidate

161

162

163def candidate_to_sdp(candidate: RTCIceCandidate) -> str:

164 sdp = (

165 f"{candidate.foundation} {candidate.component} {candidate.protocol} "

166 f"{candidate.priority} {candidate.ip} {candidate.port} typ {candidate.type}"

167 )

168

169 if candidate.relatedAddress is not None:

170 sdp += f" raddr {candidate.relatedAddress}"

171 if candidate.relatedPort is not None:

172 sdp += f" rport {candidate.relatedPort}"

173 if candidate.tcpType is not None:

174 sdp += f" tcptype {candidate.tcpType}"

175 return sdp

176

177

178def grouplines(sdp: str) -> tuple[list[str], list[list[str]]]:

179 session = []

180 media = []

181 for line in sdp.splitlines():

182 if line.startswith("m="):

183 media.append([line])

184 elif len(media):

185 media[-1].append(line)

186 else:

187 session.append(line)

188 return session, media

189

190

191def ipaddress_from_sdp(sdp: str) -> str:

192 m = re.match("^IN (IP4|IP6) ([^ ]+)$", sdp)

193 if not m:

194 raise ValueError("Could not parse address from `%s`" % sdp)

195 return m.group(2)

196

197

198def ipaddress_to_sdp(addr: str) -> str:

199 version = ipaddress.ip_address(addr).version

200 return f"IN IP{version} {addr}"

201

202

203def parameters_from_sdp(sdp: str) -> ParametersDict:

204 parameters: ParametersDict = {}

205 for param in sdp.split(";"):

206 if "=" in param:

207 k, v = param.split("=", 1)

208 if k in FMTP_INT_PARAMETERS:

209 try:

210 parameters[k] = int(v)

211 except ValueError:

212 parameters[k] = v

213 else:

214 parameters[k] = v

215 else:

216 parameters[param] = None

217 return parameters

218

219

220def parameters_to_sdp(parameters: ParametersDict) -> str:

221 params = []

222 for param_k, param_v in parameters.items():

223 if param_v is not None:

224 params.append(f"{param_k}={param_v}")

225 else:

226 params.append(param_k)

227 return ";".join(params)

228

229

230def parse_attr(line: str) -> tuple[str, str]:

231 if ":" in line:

232 bits = line[2:].split(":", 1)

233 return bits[0], bits[1]

234 else:

235 # Flag attribute (no colon): return "" not None so downstream str/int ops don't raise.

236 return line[2:], ""

237

238

239def parse_h264_profile_level_id(profile_str: str) -> tuple[H264Profile, H264Level]:

240 if not isinstance(profile_str, str) or not re.match(

241 "^[0-9a-f]{6}$", profile_str, re.I

242 ):

243 raise ValueError("Expected a 6 character hexadecimal string")

244

245 level_idc = int(profile_str[4:6], 16)

246 profile_iop = int(profile_str[2:4], 16)

247 profile_idc = int(profile_str[0:2], 16)

248

249 level: H264Level

250 if level_idc == H264Level.LEVEL1_1:

251 level = H264Level.LEVEL1_B if (profile_iop & 0x10) else H264Level.LEVEL1_1

252 else:

253 try:

254 level = H264Level(level_idc)

255 except ValueError:

256 raise ValueError(f"Unrecognized level_idc = {level_idc}")

257

258 for idc, pattern, profile in H264_PROFILE_PATTERNS:

259 if idc == profile_idc and pattern.matches(profile_iop):

260 return profile, level

261

262 raise ValueError(

263 f"Unrecognized profile_iop = {profile_iop}, profile_idc = {profile_idc}"

264 )

265

266

267@dataclass

268class GroupDescription:

269 semantic: str

270 items: list[Union[int, str]]

271

272 def __str__(self) -> str:

273 return f"{self.semantic} {' '.join(map(str, self.items))}"

274

275

276def parse_group(

277 dest: list[GroupDescription], value: str, type: Union[type[str], type[int]] = str

278) -> None:

279 bits = value.split()

280 if bits:

281 try:

282 items = list(map(type, bits[1:]))

283 except ValueError:

284 return

285 dest.append(GroupDescription(semantic=bits[0], items=items))

286

287

288@dataclass

289class SsrcDescription:

290 ssrc: int

291 cname: Optional[str] = None

292 msid: Optional[str] = None

293 mslabel: Optional[str] = None

294 label: Optional[str] = None

295

296

297SSRC_INFO_ATTRS = ["cname", "msid", "mslabel", "label"]

298

299

300class MediaDescription:

301 def __init__(self, kind: str, port: int, profile: str, fmt: list[Any]) -> None:

302 # rtp

303 self.kind = kind

304 self.port = port

305 self.host: Optional[str] = None

306 self.profile = profile

307 self.direction: Optional[str] = None

308 self.msid: Optional[str] = None

309

310 # rtcp

311 self.rtcp_port: Optional[int] = None

312 self.rtcp_host: Optional[str] = None

313 self.rtcp_mux = False

314

315 # ssrc

316 self.ssrc: list[SsrcDescription] = []

317 self.ssrc_group: list[GroupDescription] = []

318 # RFC 6236 resolution envelopes, emitted verbatim as a=imageattr lines.

319 self.imageattr: list[str] = []

320

321 # formats

322 self.fmt = fmt

323 self.rtp = RTCRtpParameters()

324

325 # SCTP

326 self.sctpCapabilities: Optional[RTCSctpCapabilities] = None

327 self.sctpmap: dict[int, str] = {}

328 self.sctp_port: Optional[int] = None

329

330 # DTLS

331 self.dtls: Optional[RTCDtlsParameters] = None

332

333 # ICE

334 self.ice: Optional[RTCIceParameters] = None

335 self.ice_candidates: list[RTCIceCandidate] = []

336 self.ice_candidates_complete = False

337 self.ice_options: Optional[str] = None

338

339 def __str__(self) -> str:

340 lines = []

341 lines.append(

342 f"m={self.kind} {self.port} {self.profile} {' '.join(map(str, self.fmt))}"

343 )

344 if self.host is not None:

345 lines.append(f"c={ipaddress_to_sdp(self.host)}")

346 if self.direction is not None:

347 lines.append(f"a={self.direction}")

348

349 for header in self.rtp.headerExtensions:

350 lines.append(f"a=extmap:{header.id} {header.uri}")

351

352 if self.rtp.muxId:

353 lines.append(f"a=mid:{self.rtp.muxId}")

354

355 if self.msid:

356 lines.append(f"a=msid:{self.msid}")

357

358 if self.rtcp_port is not None:

359 line = f"a=rtcp:{self.rtcp_port}"

360 if self.rtcp_host is not None:

361 line += f" {ipaddress_to_sdp(self.rtcp_host)}"

362 lines.append(line)

363 if self.rtcp_mux:

364 lines.append("a=rtcp-mux")

365

366 for group in self.ssrc_group:

367 lines.append(f"a=ssrc-group:{group}")

368 for imageattr in self.imageattr:

369 lines.append(f"a=imageattr:{imageattr}")

370 for ssrc_info in self.ssrc:

371 for ssrc_attr in SSRC_INFO_ATTRS:

372 ssrc_value = getattr(ssrc_info, ssrc_attr)

373 if ssrc_value is not None:

374 lines.append(f"a=ssrc:{ssrc_info.ssrc} {ssrc_attr}:{ssrc_value}")

375

376 for codec in self.rtp.codecs:

377 lines.append(f"a=rtpmap:{codec.payloadType} {codec}")

378

379 # RTCP feedback

380 for feedback in codec.rtcpFeedback:

381 value = feedback.type

382 if feedback.parameter:

383 value += f" {feedback.parameter}"

384 lines.append(f"a=rtcp-fb:{codec.payloadType} {value}")

385

386 # parameters

387 params = parameters_to_sdp(codec.parameters)

388 if params:

389 lines.append(f"a=fmtp:{codec.payloadType} {params}")

390

391 for k, v in self.sctpmap.items():

392 lines.append(f"a=sctpmap:{k} {v}")

393 if self.sctp_port is not None:

394 lines.append(f"a=sctp-port:{self.sctp_port}")

395 if self.sctpCapabilities is not None:

396 lines.append(f"a=max-message-size:{self.sctpCapabilities.maxMessageSize}")

397

398 # ice

399 for candidate in self.ice_candidates:

400 lines.append("a=candidate:" + candidate_to_sdp(candidate))

401 if self.ice_candidates_complete:

402 lines.append("a=end-of-candidates")

403 if self.ice.usernameFragment is not None:

404 lines.append(f"a=ice-ufrag:{self.ice.usernameFragment}")

405 if self.ice.password is not None:

406 lines.append(f"a=ice-pwd:{self.ice.password}")

407 if self.ice_options is not None:

408 lines.append(f"a=ice-options:{self.ice_options}")

409

410 # dtls

411 if self.dtls:

412 for fingerprint in self.dtls.fingerprints:

413 lines.append(

414 f"a=fingerprint:{fingerprint.algorithm} {fingerprint.value}"

415 )

416 lines.append(f"a=setup:{DTLS_ROLE_SETUP[self.dtls.role]}")

417

418 return "\r\n".join(lines) + "\r\n"

419

420

421class SessionDescription:

422 def __init__(self) -> None:

423 self.version = 0

424 self.origin: Optional[str] = None

425 self.name = "-"

426 self.time = "0 0"

427 self.host: Optional[str] = None

428 self.group: list[GroupDescription] = []

429 self.msid_semantic: list[GroupDescription] = []

430 self.media: list[MediaDescription] = []

431 self.type: Optional[str] = None

432

433 @classmethod

434 def parse(cls, sdp: str) -> "SessionDescription":

435 current_media: Optional[MediaDescription] = None

436 dtls_fingerprints = []

437 dtls_role = None

438 ice_lite = False

439 ice_options = None

440 ice_password = None

441 ice_usernameFragment = None

442

443 session_lines, media_groups = grouplines(sdp)

444

445 # parse session

446 session = cls()

447 for line in session_lines:

448 if line.startswith("v="):

449 session.version = int(line.strip()[2:])

450 elif line.startswith("o="):

451 session.origin = line.strip()[2:]

452 elif line.startswith("s="):

453 session.name = line.strip()[2:]

454 elif line.startswith("c="):

455 session.host = ipaddress_from_sdp(line[2:])

456 elif line.startswith("t="):

457 session.time = line.strip()[2:]

458 elif line.startswith("a="):

459 attr, value = parse_attr(line)

460 if attr == "fingerprint":

461 fp_bits = value.split()

462 if len(fp_bits) != 2:

463 continue

464 algorithm, fingerprint = fp_bits

465 dtls_fingerprints.append(

466 RTCDtlsFingerprint(algorithm=algorithm, value=fingerprint)

467 )

468 elif attr == "ice-lite":

469 ice_lite = True

470 elif attr == "ice-options":

471 ice_options = value

472 elif attr == "ice-pwd":

473 ice_password = value

474 elif attr == "ice-ufrag":

475 ice_usernameFragment = value

476 elif attr == "group":

477 parse_group(session.group, value)

478 elif attr == "msid-semantic":

479 parse_group(session.msid_semantic, value)

480 elif attr == "setup":

481 # Unknown setup value: ignore rather than reject the whole offer.

482 role = DTLS_SETUP_ROLE.get(value)

483 if role is not None:

484 dtls_role = role

485

486 # parse media

487 for media_lines in media_groups:

488 m = re.match("^m=([^ ]+) ([0-9]+) ([A-Z/]+) (.+)$", media_lines[0])

489 if not m:

490 raise ValueError("Could not parse media line `%s`" % media_lines[0])

491

492 # check payload types are valid

493 kind = m.group(1)

494 fmt = m.group(4).split()

495 fmt_int: Optional[list[int]] = None

496 if kind in ["audio", "video"]:

497 fmt_int = [int(x) for x in fmt]

498 for pt in fmt_int:

499 if not (pt >= 0 and pt < 256):

500 raise ValueError("Payload type must be in range [0, 256[")

501 if pt in rtp.FORBIDDEN_PAYLOAD_TYPES:

502 raise ValueError(

503 "Payload type `%d` is forbidden" % pt

504 )

505

506 current_media = MediaDescription(

507 kind=kind, port=int(m.group(2)), profile=m.group(3), fmt=fmt_int or fmt

508 )

509 current_media.dtls = RTCDtlsParameters(

510 fingerprints=dtls_fingerprints[:], role=dtls_role

511 )

512 current_media.ice = RTCIceParameters(

513 iceLite=ice_lite,

514 usernameFragment=ice_usernameFragment,

515 password=ice_password,

516 )

517 current_media.ice_options = ice_options

518 session.media.append(current_media)

519 ssrcs: dict[int, SsrcDescription] = {}

520

521 for line in media_lines[1:]:

522 if line.startswith("c="):

523 current_media.host = ipaddress_from_sdp(line[2:])

524 elif line.startswith("a="):

525 attr, value = parse_attr(line)

526 if attr == "candidate":

527 try:

528 candidate = candidate_from_sdp(value)

529 except ValueError:

530 # Malformed candidate: skip this attr, not the whole offer.

531 continue

532 current_media.ice_candidates.append(candidate)

533 elif attr == "end-of-candidates":

534 current_media.ice_candidates_complete = True

535 elif attr == "extmap":

536 if not value:

537 continue

538 ext_bits = value.split()

539 if len(ext_bits) < 2:

540 continue

541 ext_id, ext_uri = ext_bits[0], ext_bits[1]

542 if "/" in ext_id:

543 # Bounded split: tolerate an id with extra '/' (e.g. "2/recvonly/foo").

544 ext_id, ext_direction = ext_id.split("/", 1)

545 try:

546 ext_id_int = int(ext_id)

547 except ValueError:

548 # Non-numeric extmap id: skip this extension, not the whole offer.

549 continue

550 extension = RTCRtpHeaderExtensionParameters(

551 id=ext_id_int, uri=ext_uri

552 )

553 current_media.rtp.headerExtensions.append(extension)

554 elif attr == "fingerprint":

555 fp_bits = value.split()

556 if len(fp_bits) != 2:

557 continue

558 algorithm, fingerprint = fp_bits

559 current_media.dtls.fingerprints.append(

560 RTCDtlsFingerprint(algorithm=algorithm, value=fingerprint)

561 )

562 elif attr == "ice-options":

563 current_media.ice_options = value

564 elif attr == "ice-pwd":

565 current_media.ice.password = value

566 elif attr == "ice-ufrag":

567 current_media.ice.usernameFragment = value

568 elif attr == "max-message-size":

569 try:

570 current_media.sctpCapabilities = RTCSctpCapabilities(

571 maxMessageSize=int(value)

572 )

573 except ValueError:

574 continue

575 elif attr == "mid":

576 current_media.rtp.muxId = value

577 elif attr == "msid":

578 current_media.msid = value

579 elif attr == "rtcp":

580 if not value:

581 continue

582 bits = value.split(" ", 1)

583 try:

584 current_media.rtcp_port = int(bits[0])

585 if len(bits) > 1:

586 # Malformed address: skip attr, not the whole offer.

587 current_media.rtcp_host = ipaddress_from_sdp(bits[1])

588 except ValueError:

589 continue

590 elif attr == "rtcp-mux":

591 current_media.rtcp_mux = True

592 elif attr == "setup":

593 # Unknown setup value: dtls.role stays None; below it is set

594 # to None (the media section itself is retained, not dropped).

595 role = DTLS_SETUP_ROLE.get(value)

596 if role is not None:

597 current_media.dtls.role = role

598 elif attr in DIRECTIONS:

599 current_media.direction = attr

600 elif attr == "rtpmap":

601 if not value or " " not in value:

602 continue

603 format_id, format_desc = value.split(" ", 1)

604 bits = format_desc.split("/")

605 if len(bits) < 2:

606 continue

607 try:

608 if current_media.kind == "audio":

609 if len(bits) > 2:

610 channels = int(bits[2])

611 else:

612 channels = 1

613 else:

614 channels = None

615 codec = RTCRtpCodecParameters(

616 mimeType=current_media.kind + "/" + bits[0],

617 channels=channels,

618 clockRate=int(bits[1]),

619 payloadType=int(format_id),

620 )

621 except ValueError:

622 continue

623 if len(current_media.rtp.codecs) >= MAX_CODECS:

624 raise ValueError(f"A media section names more than {MAX_CODECS} codecs")

625 current_media.rtp.codecs.append(codec)

626 elif attr == "sctpmap":

627 if not value or " " not in value:

628 continue

629 format_id, format_desc = value.split(" ", 1)

630 try:

631 getattr(current_media, attr)[int(format_id)] = format_desc

632 except ValueError:

633 continue

634 elif attr == "sctp-port":

635 try:

636 current_media.sctp_port = int(value)

637 except ValueError:

638 continue

639 elif attr == "ssrc-group":

640 parse_group(current_media.ssrc_group, value, type=int)

641 elif attr == "imageattr":

642 current_media.imageattr.append(value)

643 elif attr == "ssrc":

644 if not value or " " not in value:

645 continue

646 ssrc_str, ssrc_desc = value.split(" ", 1)

647 try:

648 ssrc = int(ssrc_str)

649 except ValueError:

650 continue

651 if ":" not in ssrc_desc:

652 continue

653 ssrc_attr, ssrc_value = ssrc_desc.split(":", 1)

654

655 ssrc_info = ssrcs.get(ssrc)

656 if ssrc_info is None:

657 if len(ssrcs) >= MAX_SSRCS:

658 raise ValueError(f"A media section names more than {MAX_SSRCS} SSRCs")

659 ssrc_info = ssrcs[ssrc] = SsrcDescription(ssrc=ssrc)

660 current_media.ssrc.append(ssrc_info)

661 if ssrc_attr in SSRC_INFO_ATTRS:

662 setattr(ssrc_info, ssrc_attr, ssrc_value)

663

664 if current_media.dtls.role is None:

665 current_media.dtls = None

666

667 # requires codecs to have been parsed

668 by_pt: dict[int, list[RTCRtpCodecParameters]] = {}

669 for codec in current_media.rtp.codecs:

670 by_pt.setdefault(codec.payloadType, []).append(codec)

671 feedback = 0

672 for line in media_lines[1:]:

673 if line.startswith("a="):

674 attr, value = parse_attr(line)

675 if attr == "fmtp":

676 if not value or " " not in value:

677 continue

678 format_id, format_desc = value.split(" ", 1)

679 try:

680 named = by_pt.get(int(format_id))

681 except ValueError:

682 continue

683 if not named:

684 continue

685 named[0].parameters = parameters_from_sdp(format_desc)

686 elif attr == "rtcp-fb":

687 if not value:

688 continue

689 bits = value.split(" ", 2)

690 if len(bits) < 2:

691 continue

692 feedback += 1

693 if feedback > MAX_RTCP_FB:

694 raise ValueError(f"A media section names more than {MAX_RTCP_FB} feedback lines")

695 if bits[0] == "*":

696 named = current_media.rtp.codecs

697 else:

698 try:

699 named = by_pt.get(int(bits[0]), [])

700 except ValueError:

701 continue

702 for codec in named:

703 codec.rtcpFeedback.append(

704 RTCRtcpFeedback(

705 type=bits[1],

706 parameter=bits[2] if len(bits) > 2 else None,

707 )

708 )

709

710 return session

711

712 def webrtc_track_id(self, media: MediaDescription) -> Optional[str]:

713 assert media in self.media

714 if media.msid is not None and " " in media.msid:

715 bits = media.msid.split()

716 for group in self.msid_semantic:

717 if group.semantic == "WMS" and (

718 bits[0] in group.items or "*" in group.items

719 ):

720 return bits[1]

721 return None

722

723 def __str__(self) -> str:

724 lines = [f"v={self.version}", f"o={self.origin}", f"s={self.name}"]

725 if self.host is not None:

726 lines += [f"c={ipaddress_to_sdp(self.host)}"]

727 lines += [f"t={self.time}"]

728 if any(m.ice.iceLite for m in self.media):

729 lines += ["a=ice-lite"]

730 for group in self.group:

731 lines += [f"a=group:{group}"]

732 for group in self.msid_semantic:

733 lines += [f"a=msid-semantic:{group}"]

734 return "\r\n".join(lines) + "\r\n" + "".join([str(m) for m in self.media])