Repository: aiortc/aioice

Issue:
IndexError in `xor_address`
Encountered this in production.

<img width="1143" alt="image" src="https://github.com/aiortc/aioice/assets/812086/6d350d08-3f0a-47e4-b21b-33eb8162c062">

Hints from issue comments:
@jlaine: That is really odd, would it be possible to have a wireshark capture with one such packet? Im wondering if its actually the address parsing that is failing or whether the error is happening earlier@pencil: We no longer use WebRTC for our service so unfortunately I won't be able to help you with a packet capture. At the time, we were using aiortc in conjunction with Twilio's STUN and TURN servers. We saw this error fairly regularly.@jlaine: Looking at the data you provided, it looks like the data is far too long to be a real XOR-MAPPED-ADDRESS. I don't know what's causing this - but it's probably not an actual STUN packet. In any case I have a pull request which fixes the `IndexError`.

Solution patch:
diff --git a/src/aioice/stun.py b/src/aioice/stun.py
index e5971fe..1d1681e 100644
--- a/src/aioice/stun.py
+++ b/src/aioice/stun.py
@@ -35,21 +35,24 @@ def message_integrity(data: bytes, key: bytes) -> bytes:
     return hmac.new(key, check_data, "sha1").digest()
 
 
-def xor_address(data: bytes, transaction_id: bytes) -> bytes:
+def xor_data(data: bytes, transaction_id: bytes) -> bytes:
     xpad = pack("!HI", COOKIE >> 16, COOKIE) + transaction_id
-    xdata = data[0:2]
-    for i in range(2, len(data)):
-        xdata += int.to_bytes(data[i] ^ xpad[i - 2], 1, "big", signed=False)
+    xdata = b""
+    for i in range(len(data)):
+        xdata += int.to_bytes(data[i] ^ xpad[i], 1, "big", signed=False)
     return xdata
 
 
-def pack_address(value: tuple[str, int]) -> bytes:
+def pack_address(
+    value: tuple[str, int],
+    xor: Callable[[bytes], bytes] = lambda x: x,
+) -> bytes:
     ip_address = ipaddress.ip_address(value[0])
     if isinstance(ip_address, ipaddress.IPv4Address):
         protocol = IPV4_PROTOCOL
     else:
         protocol = IPV6_PROTOCOL
-    return pack("!BBH", 0, protocol, value[1]) + ip_address.packed
+    return pack("!BB", 0, protocol) + xor(pack("!H", value[1]) + ip_address.packed)
 
 
 def pack_bytes(value: bytes) -> bytes:
@@ -81,28 +84,46 @@ def pack_unsigned_64(value: int) -> bytes:
 
 
 def pack_xor_address(value: tuple[str, int], transaction_id: bytes) -> bytes:
-    return xor_address(pack_address(value), transaction_id)
+    return pack_address(value, lambda x: xor_data(x, transaction_id))
 
 
-def unpack_address(data: bytes) -> tuple[str, int]:
+def unpack_address(
+    data: bytes,
+    xor: Callable[[bytes], bytes] = lambda x: x,
+) -> tuple[str, int]:
     if len(data) < 4:
         raise ValueError("STUN address length is less than 4 bytes")
-    reserved, protocol, port = unpack("!BBH", data[0:4])
-    address = data[4:]
+
+    # The first two bytes are a reserved byte and the protocol family.
+    reserved, protocol = unpack("!BB", data[0:2])
+
+    # The remaining data represents a port and an address.
+    port_address = data[2:]
+    print(len(port_address))
     if protocol == IPV4_PROTOCOL:
-        if len(address) != 4:
+        # For IPv4 we expect 2 bytes for the port, 4 bytes for the address.
+        if len(port_address) != 6:
             raise ValueError("STUN address has invalid length for IPv4")
-        return (str(ipaddress.IPv4Address(address)), port)
+        port_address = xor(port_address)
+        return (
+            str(ipaddress.IPv4Address(port_address[2:])),
+            unpack("!H", port_address[0:2])[0],
+        )
     elif protocol == IPV6_PROTOCOL:
-        if len(address) != 16:
+        # For IPv6 we expect 2 bytes for the port, 16 bytes for the address.
+        if len(port_address) != 18:
             raise ValueError("STUN address has invalid length for IPv6")
-        return (str(ipaddress.IPv6Address(address)), port)
+        port_address = xor(port_address)
+        return (
+            str(ipaddress.IPv6Address(port_address[2:])),
+            unpack("!H", port_address[0:2])[0],
+        )
     else:
         raise ValueError("STUN address has unknown protocol")
 
 
 def unpack_xor_address(data: bytes, transaction_id: bytes) -> tuple[str, int]:
-    return unpack_address(xor_address(data, transaction_id))
+    return unpack_address(data, lambda x: xor_data(x, transaction_id))
 
 
 def unpack_bytes(data: bytes) -> bytes:


Test patch:
diff --git a/tests/test_stun.py b/tests/test_stun.py
index caed6cd..85c566b 100644
--- a/tests/test_stun.py
+++ b/tests/test_stun.py
@@ -28,7 +28,18 @@ class AttributeTest(unittest.TestCase):
         self.assertEqual(address, "192.0.2.1")
         self.assertEqual(port, 32853)
 
-    def test_unpack_xor_address_ipv4_truncated(self) -> None:
+    def test_unpack_xor_address_ipv4_too_long(self) -> None:
+        transaction_id = unhexlify("b7e7a701bc34d686fa87dfae")
+        with self.assertRaises(ValueError) as cm:
+            stun.unpack_xor_address(
+                # Use 21 bytes for port + address, to check we don't try
+                # XOR'ing this data against the 20 bytes key.
+                unhexlify("0001a147e112a643000000000000000000000000000000"),
+                transaction_id,
+            )
+        self.assertEqual(str(cm.exception), "STUN address has invalid length for IPv4")
+
+    def test_unpack_xor_address_ipv4_too_short(self) -> None:
         transaction_id = unhexlify("b7e7a701bc34d686fa87dfae")
         with self.assertRaises(ValueError) as cm:
             stun.unpack_xor_address(unhexlify("0001a147e112a6"), transaction_id)
@@ -42,7 +53,18 @@ class AttributeTest(unittest.TestCase):
         self.assertEqual(address, "2001:db8:1234:5678:11:2233:4455:6677")
         self.assertEqual(port, 32853)
 
-    def test_unpack_xor_address_ipv6_truncated(self) -> None:
+    def test_unpack_xor_address_ipv6_too_long(self) -> None:
+        transaction_id = unhexlify("b7e7a701bc34d686fa87dfae")
+        with self.assertRaises(ValueError) as cm:
+            stun.unpack_xor_address(
+                # Use 21 bytes for port + address, to check we don't try
+                # XOR'ing this data against the 20 bytes key.
+                unhexlify("0002a1470113a9faa5d3f179bc25f4b5bed2b900000000"),
+                transaction_id,
+            )
+        self.assertEqual(str(cm.exception), "STUN address has invalid length for IPv6")
+
+    def test_unpack_xor_address_ipv6_too_short(self) -> None:
         transaction_id = unhexlify("b7e7a701bc34d686fa87dfae")
         with self.assertRaises(ValueError) as cm:
             stun.unpack_xor_address(