Stun improvements (#4712)

* STUN: Fix building binding response

Building a binding success response with an XOR-MAPPED-ADDRESS attribute
filed with:
      File "/home/simon/src/scapy/scapy/contrib/stun.py", line 152, in i2m
        return struct.pack(">i", struct.unpack(">i", inet_aton(x)) ^ MAGIC_COOKIE)
                                 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^~~~~~~~~~~~~~
    TypeError: While dissecting field 'attributes': While dissecting field 'xip': unsupported operand type(s) for ^: 'tuple' and 'int'

Fixed by properly handling the return value of struct.unpack (a tuple
with one int).

* STUN: Support IPv6 in binding response

The XOR-MAPPED-ADDRESS attribute can contain either IPv4 or IPv6
addresses in the xip field. The address_family field selects the type of
the xip field.

* STUN: Support MAPPED-ADDRESS attribute

Even RFC 8489 compliant STUN servers respond with the non-XORed
MAPPED-ADDRESS attribute, if the magic cookie in the request doesn't
have the expected value.

* STUN: Fix UDP/TCP layer binding

STUN responses weren't properly dissected, because the protocol was only
bound to the destination port. Fixed by binding both source and
destination port number 3478 to the STUN protocol. Also set both port
numbers when building messages, otherwise the UDP/TCP default source
port (DNS / FTP) would be used.

Author: holesch

scapy/contrib/stun.py

@@ -20,7 +20,7 @@
 
 from scapy.layers.inet import UDP, TCP
 from scapy.config import conf
-from scapy.packet import Packet, bind_layers
+from scapy.packet import Packet, bind_bottom_up, bind_top_down
 from scapy.utils import inet_ntoa, inet_aton
 from scapy.fields import (
     BitField,
@@ -39,7 +39,9 @@
     XLongField,
     XIntField,
     XBitField,
-    IPField
+    IPField,
+    IP6Field,
+    MultipleTypeField,
 )
 
 MAGIC_COOKIE = 0x2112A442
@@ -149,19 +151,64 @@ def m2i(self, pkt, x):
     def i2m(self, pkt, x):
         if x is None:
             return b"\x00\x00\x00\x00"
-        return struct.pack(">i", struct.unpack(">i", inet_aton(x)) ^ MAGIC_COOKIE)
+        return struct.pack(">i", struct.unpack(">i", inet_aton(x))[0] ^ MAGIC_COOKIE)
+
+
+class XorIp6(IP6Field):
+
+    def m2i(self, pkt, x):
+        addr = self._xor_address(pkt, x)
+        return super().m2i(pkt, addr)
+
+    def i2m(self, pkt, x):
+        addr = super().i2m(pkt, x)
+        return self._xor_address(pkt, addr)
+
+    def _xor_address(self, pkt, addr):
+        xor_words = [pkt.parent.magic_cookie]
+        xor_words += struct.unpack(
+            ">III", pkt.parent.transaction_id.to_bytes(12, "big")
+        )
+        addr_words = struct.unpack(">IIII", addr)
+        xor_addr = [a ^ b for a, b in zip(addr_words, xor_words)]
+        return struct.pack(">IIII", *xor_addr)
 
 
 class STUNXorMappedAddress(STUNGenericTlv):
     name = "STUN XOR Mapped Address"
 
     fields_desc = [
         XShortField("type", 0x0020),
-        ShortField("length", 8),
+        FieldLenField("length", None, length_of="xip", adjust=lambda pkt, x: x + 4),
         ByteField("RESERVED", 0),
         ByteEnumField("address_family", 1, _xor_mapped_address_family),
         XorPort("xport", 0),
-        XorIp("xip", 0)     # FIXME <- only IPv4 addresses will work
+        MultipleTypeField(
+            [
+                (XorIp("xip", "127.0.0.1"), lambda pkt: pkt.address_family == 1),
+                (XorIp6("xip", "::1"), lambda pkt: pkt.address_family == 2),
+            ],
+            XorIp("xip", "127.0.0.1"),
+        ),
+    ]
+
+
+class STUNMappedAddress(STUNGenericTlv):
+    name = "STUN Mapped Address"
+
+    fields_desc = [
+        XShortField("type", 0x0001),
+        FieldLenField("length", None, length_of="ip", adjust=lambda pkt, x: x + 4),
+        ByteField("RESERVED", 0),
+        ByteEnumField("address_family", 1, _xor_mapped_address_family),
+        ShortField("port", 0),
+        MultipleTypeField(
+            [
+                (IPField("ip", "127.0.0.1"), lambda pkt: pkt.address_family == 1),
+                (IP6Field("ip", "::1"), lambda pkt: pkt.address_family == 2),
+            ],
+            IPField("ip", "127.0.0.1"),
+        ),
     ]
 
 
@@ -207,6 +254,7 @@ class STUNGoogNetworkInfo(STUNGenericTlv):
 
 
 _stun_tlv_class = {
+    0x0001: STUNMappedAddress,
     0x0006: STUNUsername,
     0x0008: STUNMessageIntegrity,
     0x0020: STUNXorMappedAddress,
@@ -259,5 +307,10 @@ def post_build(self, pkt, pay):
         return pkt
 
 
-bind_layers(UDP, STUN, dport=3478)
-bind_layers(TCP, STUN, dport=3478)
+bind_bottom_up(UDP, STUN, sport=3478)
+bind_bottom_up(UDP, STUN, dport=3478)
+bind_top_down(UDP, STUN, sport=3478, dport=3478)
+
+bind_bottom_up(TCP, STUN, sport=3478)
+bind_bottom_up(TCP, STUN, dport=3478)
+bind_top_down(TCP, STUN, sport=3478, dport=3478)

test/contrib/stun.uts

@@ -78,7 +78,7 @@ assert parsed.length == 44
 assert parsed.magic_cookie == 0x2112A442
 assert parsed.transaction_id == 0xcfacb2a43aa2de5a9d56d85a, parsed.transaction_id
 assert parsed.attributes == [
-      STUNXorMappedAddress(xport=40480, xip="172.20.0.42"),
+      STUNXorMappedAddress(length=8, xport=40480, xip="172.20.0.42"),
       STUNMessageIntegrity(hmac_sha1=0xb71fc9235897c802e3fff8e3d889fa41428d967d),
       STUNFingerprint(crc_32=0xea9b6559)
 ]
@@ -99,14 +99,51 @@ assert STUN.stun_message_type.i2repr(None, parsed.stun_message_type) == "Binding
 assert parsed.length == 88
 assert parsed.magic_cookie == 0x2112A442
 assert parsed.transaction_id == 0x3479476534635936316a796a, parsed.transaction_id
-assert parsed.attributes[0] == STUNXorMappedAddress(xport=25000, xip="172.20.0.200"), parsed.attributes
+assert parsed.attributes[0] == STUNXorMappedAddress(length=8, xport=25000, xip="172.20.0.200"), parsed.attributes
 assert parsed.attributes == [
-      STUNXorMappedAddress(xport=25000, xip="172.20.0.200"),
+      STUNXorMappedAddress(length=8, xport=25000, xip="172.20.0.200"),
       STUNUsername(length=37, username="Ht11MaRZHc4GOLJUsbu1R3YCs72HYN25:oNph"),
       STUNMessageIntegrity(hmac_sha1=0x4b67036dfb65ca84d63bcac86c8d5981df657031),
       STUNFingerprint(crc_32=0x4041e9c3)
 ]
 
+= test STUN binding success response IPv6
+
+raw = b"\x01\x01\x00\x18\x21\x12\xa4\x42\x91\x1b\x25\x32\x99\x8d\xa0\x1c" \
+      b"\xf9\xd0\x53\xd9\x00\x20\x00\x14\x00\x02\x3c\xd7\x21\x12\xa4\x42" \
+      b"\x91\x1b\x25\x32\x99\x8d\xa0\x1c\xf9\xd0\x53\xd8"
+
+parsed = STUN(raw)
+assert parsed.RESERVED == 0x00, parsed.RESERVED
+assert STUN.stun_message_type.i2repr(None, parsed.stun_message_type) == "Binding success response"
+assert parsed.length == 24
+assert parsed.magic_cookie == 0x2112A442
+assert parsed.transaction_id == 0x911b2532998da01cf9d053d9, parsed.transaction_id
+assert len(parsed.attributes) == 1, len(parsed.attributes)
+assert parsed.attributes[0].type == 0x0020, parsed.attributes[0].type
+assert parsed.attributes[0].length == 20, parsed.attributes[0].length
+assert parsed.attributes[0].address_family == 0x02, parsed.attributes[0].address_family
+assert parsed.attributes[0].xport == 7621, parsed.attributes[0].xport
+assert parsed.attributes[0].xip == "::1", parsed.attributes[0].xip
+
+= test STUN classic binding success response
+
+raw = b"\x01\x01\x00\x0c\x37\x06\xd1\x4d\x38\x3a\xd6\xc8\x40\x5e\x17\x9a" \
+      b"\x93\x92\xea\xa8\x00\x01\x00\x08\x00\x01\x0d\x14\xc0\xa8\x00\x05"
+
+parsed = STUN(raw)
+assert parsed.RESERVED == 0x00, parsed.RESERVED
+assert STUN.stun_message_type.i2repr(None, parsed.stun_message_type) == "Binding success response"
+assert parsed.length == 12
+assert parsed.magic_cookie == 0x3706d14d
+assert parsed.transaction_id == 0x383ad6c8405e179a9392eaa8, parsed.transaction_id
+assert len(parsed.attributes) == 1, len(parsed.attributes)
+assert parsed.attributes[0].type == 0x0001, parsed.attributes[0].type
+assert parsed.attributes[0].length == 8, parsed.attributes[0].length
+assert parsed.attributes[0].address_family == 0x01, parsed.attributes[0].address_family
+assert parsed.attributes[0].port == 3348, parsed.attributes[0].port
+assert parsed.attributes[0].ip == "192.168.0.5", parsed.attributes[0].ip
+
 = test STUN binding indication 1
 
 raw = b"\x00\x11\x00\x08\x21\x12\xa4\x42\x29\x3d\x68\x7b\x0f\xbc\x44\x7c" \
@@ -145,4 +182,61 @@ stun = STUN(
 built = stun.build()
 parsed = STUN(built)
 
-assert parsed.build() == built
+assert parsed.build() == built
+
+= test STUN packet build with attributes
+stun = STUN(
+    stun_message_type="Binding success response",
+    transaction_id=0x3479476534635936316a796a,
+    attributes=[
+        STUNXorMappedAddress(xport=25000, xip="172.20.0.200"),
+        STUNUsername(length=37, username="Ht11MaRZHc4GOLJUsbu1R3YCs72HYN25:oNph"),
+        STUNMessageIntegrity(hmac_sha1=0x4b67036dfb65ca84d63bcac86c8d5981df657031),
+        STUNFingerprint(crc_32=0x4041e9c3)
+    ]
+)
+
+built = stun.build()
+parsed = STUN(built)
+
+assert parsed.build() == built
+
+= test STUN packet build IPv6
+
+stun = STUN(
+    stun_message_type="Binding success response",
+    transaction_id=0x911b2532998da01cf9d053d9,
+    attributes=[
+        STUNXorMappedAddress(xport=7621, address_family="IPv6", xip="::1")
+    ]
+)
+built = stun.build()
+parsed = STUN(built)
+
+assert parsed.build() == built
+assert parsed.attributes[0].length == 20
+
+= test STUN bottom up binding 1
+
+udp = UDP(sport=62049, dport=3478) / STUN()
+built = udp.build()
+parsed = UDP(built)
+
+assert type(parsed.payload) == STUN, parsed.show(dump=True)
+
+= test STUN bottom up binding 2
+
+udp = UDP(sport=3478, dport=62049) / STUN(stun_message_type="Binding error response")
+built = udp.build()
+parsed = UDP(built)
+
+assert type(parsed.payload) == STUN, parsed.show(dump=True)
+
+= test STUN top down binding
+
+udp = UDP() / STUN()
+built = udp.build()
+parsed = UDP(built)
+
+assert parsed.sport == 3478, parsed.sport
+assert parsed.dport == 3478, parsed.dport