|
import hashlib |
|
import ipaddress |
|
from typing import Optional |
|
|
|
|
|
def candidate_foundation( |
|
candidate_type: str, candidate_transport: str, base_address: str |
|
) -> str: |
|
""" |
|
See RFC 5245 - 4.1.1.3. Computing Foundations |
|
""" |
|
key = "%s|%s|%s" % (candidate_type, candidate_transport, base_address) |
|
return hashlib.md5(key.encode("ascii")).hexdigest() |
|
|
|
|
|
def candidate_priority( |
|
candidate_component: int, candidate_type: str, local_pref: int = 65535 |
|
) -> int: |
|
""" |
|
See RFC 5245 - 4.1.2.1. Recommended Formula |
|
""" |
|
if candidate_type == "host": |
|
type_pref = 126 |
|
elif candidate_type == "prflx": |
|
type_pref = 110 |
|
elif candidate_type == "srflx": |
|
type_pref = 100 |
|
else: |
|
type_pref = 0 |
|
|
|
return (1 << 24) * type_pref + (1 << 8) * local_pref + (256 - candidate_component) |
|
|
|
|
|
class Candidate: |
|
""" |
|
An ICE candidate. |
|
""" |
|
|
|
def __init__( |
|
self, |
|
foundation: str, |
|
component: int, |
|
transport: str, |
|
priority: int, |
|
host: str, |
|
port: int, |
|
type: str, |
|
related_address: Optional[str] = None, |
|
related_port: Optional[int] = None, |
|
tcptype: Optional[str] = None, |
|
generation: Optional[int] = None, |
|
) -> None: |
|
self.foundation = foundation |
|
self.component = component |
|
self.transport = transport |
|
self.priority = priority |
|
self.host = host |
|
self.port = port |
|
self.type = type |
|
self.related_address = related_address |
|
self.related_port = related_port |
|
self.tcptype = tcptype |
|
self.generation = generation |
|
|
|
@classmethod |
|
def from_sdp(cls, sdp): |
|
""" |
|
Parse a :class:`Candidate` from SDP. |
|
|
|
.. code-block:: python |
|
|
|
Candidate.from_sdp( |
|
'6815297761 1 udp 659136 1.2.3.4 31102 typ host generation 0') |
|
""" |
|
bits = sdp.split() |
|
if len(bits) < 8: |
|
raise ValueError("SDP does not have enough properties") |
|
|
|
kwargs = { |
|
"foundation": bits[0], |
|
"component": int(bits[1]), |
|
"transport": bits[2], |
|
"priority": int(bits[3]), |
|
"host": bits[4], |
|
"port": int(bits[5]), |
|
"type": bits[7], |
|
} |
|
|
|
for i in range(8, len(bits) - 1, 2): |
|
if bits[i] == "raddr": |
|
kwargs["related_address"] = bits[i + 1] |
|
elif bits[i] == "rport": |
|
kwargs["related_port"] = int(bits[i + 1]) |
|
elif bits[i] == "tcptype": |
|
kwargs["tcptype"] = bits[i + 1] |
|
elif bits[i] == "generation": |
|
kwargs["generation"] = int(bits[i + 1]) |
|
|
|
return Candidate(**kwargs) |
|
|
|
def to_sdp(self) -> str: |
|
""" |
|
Return a string representation suitable for SDP. |
|
""" |
|
sdp = "%s %d %s %d %s %d typ %s" % ( |
|
self.foundation, |
|
self.component, |
|
self.transport, |
|
self.priority, |
|
self.host, |
|
self.port, |
|
self.type, |
|
) |
|
if self.related_address is not None: |
|
sdp += " raddr %s" % self.related_address |
|
if self.related_port is not None: |
|
sdp += " rport %s" % self.related_port |
|
if self.tcptype is not None: |
|
sdp += " tcptype %s" % self.tcptype |
|
if self.generation is not None: |
|
sdp += " generation %d" % self.generation |
|
return sdp |
|
|
|
def can_pair_with(self, other) -> bool: |
|
""" |
|
A local candidate is paired with a remote candidate if and only if |
|
the two candidates have the same component ID and have the same IP |
|
address version. |
|
""" |
|
a = ipaddress.ip_address(self.host) |
|
b = ipaddress.ip_address(other.host) |
|
return ( |
|
self.component == other.component |
|
and self.transport.lower() == other.transport.lower() |
|
and a.version == b.version |
|
) |
|
|
|
def __repr__(self) -> str: |
|
return "Candidate(%s)" % self.to_sdp() |
|
|
