Teredo: Parse teredo data from IPv6 packet

Author: tuxuser

tests/conftest.py

@@ -1,6 +1,7 @@
 from typing import Dict
 import os
 import pytest
+from binascii import unhexlify
 
 @pytest.fixture(scope='session')
 def test_data() -> Dict[str, bytes]:
@@ -10,4 +11,31 @@ def test_data() -> Dict[str, bytes]:
         with open(os.path.join(data_path, f), 'rb') as fh:
             data[f] = fh.read()
 
-    return data
+    return data
+
+@pytest.fixture(scope='session')
+def teredo_packet() -> bytes:
+    """
+    Teredo IPv6 over UDP tunneling
+    Internet Protocol Version 6, Src: 2001:0:338c:24f4:1c38:f3fd:d2f3:c93d, Dst: 2001:0:338c:24f4:43b:30e3:d2f3:c93d
+    0110 .... = Version: 6
+    .... 0000 0000 .... .... .... .... .... = Traffic Class: 0x00 (DSCP: CS0, ECN: Not-ECT)
+        .... 0000 00.. .... .... .... .... .... = Differentiated Services Codepoint: Default (0)
+        .... .... ..00 .... .... .... .... .... = Explicit Congestion Notification: Not ECN-Capable Transport (0)
+    .... .... .... 0000 0000 0000 0000 0000 = Flow Label: 0x00000
+    Payload Length: 0
+    Next Header: No Next Header for IPv6 (59)
+    Hop Limit: 21
+    Source Address: 2001:0:338c:24f4:1c38:f3fd:d2f3:c93d
+    Destination Address: 2001:0:338c:24f4:43b:30e3:d2f3:c93d
+    [Source Teredo Server IPv4: 51.140.36.244]
+    [Source Teredo Port: 3074]
+    [Source Teredo Client IPv4: 45.12.54.194]
+    [Destination Teredo Server IPv4: 51.140.36.244]
+    [Destination Teredo Port: 53020]
+    [Destination Teredo Client IPv4: 45.12.54.194]
+    """
+    return unhexlify(
+        '6000000000003b1520010000338c24f41c38f3fdd2f3c93d20010000'
+        '338c24f4043b30e3d2f3c93d01049eb8960803080000c0a889db0c02'
+    )

tests/test_teredo.py

@@ -0,0 +1,29 @@
+from binascii import unhexlify
+from xcloud.protocol import teredo
+
+def test_teredo_convert(teredo_packet: bytes):
+    parsed = teredo.TeredoPacket.parse(teredo_packet)
+    
+    # Ipv6
+    assert parsed.ipv6.version == 6
+    assert parsed.ipv6.traffic_cls == 0x00
+    assert parsed.ipv6.flow_label == 0x00
+    assert parsed.ipv6.payload_len == 0x00
+    assert parsed.ipv6.next_header == 59 # No next header
+    assert parsed.ipv6.hop_limit == 21
+    assert parsed.ipv6.src == unhexlify('20010000338c24f41c38f3fdd2f3c93d')
+    assert parsed.ipv6.dst == unhexlify('20010000338c24f4043b30e3d2f3c93d')
+
+    # Teredo source
+    assert parsed.src_teredo.udp_port == 3074
+    assert parsed.src_teredo.flags == 0x1c38
+    assert str(parsed.src_teredo.teredo_server_ipv4) == '51.140.36.244'
+    assert str(parsed.src_teredo.client_pub_ipv4) == '45.12.54.194'
+
+    # Teredo destination
+    assert parsed.dst_teredo.udp_port == 53020
+    assert parsed.dst_teredo.flags == 0x43b
+    assert str(parsed.dst_teredo.teredo_server_ipv4) == '51.140.36.244'
+    assert str(parsed.dst_teredo.client_pub_ipv4) == '45.12.54.194'
+
+

xcloud/protocol/ipv6.py

@@ -7,13 +7,45 @@ def __init__(
     ) -> None:
         self.ipv6_base = ipv6_base
 
+    @property
+    def version(self) -> int:
+        return self.ipv6_base.v
+    
+    @property
+    def traffic_cls(self) -> int:
+        return self.ipv6_base.fc
+    
+    @property
+    def flow_label(self) -> int:
+        return self.ipv6_base.flow
+    
+    @property
+    def payload_len(self) -> int:
+        return self.ipv6_base.plen
+
+    @property
+    def next_header(self) -> int:
+        return self.ipv6_base.nxt
+
+    @property
+    def hop_limit(self) -> int:
+        return self.ipv6_base.hlim
+
+    @property
+    def src(self) -> bytes:
+        return self.ipv6_base.src
+    
+    @property
+    def dst(self) -> bytes:
+        return self.ipv6_base.dst
+
     @property
     def data(self) -> bytes:
         return self.ipv6_base.data
 
     def __repr__(self):
         return (
-            f"IPv6Packet(V={self.ipv6_base.v}, SRC={self.ipv6_base.src}, DST={self.ipv6_base.dst}, PLEN={self.ipv6_base.plen} NEXT={self.ipv6_base.nxt} HLIM={self.ipv6_base.hlim}) DATA={self.data}"
+            f"IPv6Packet(V={self.version}, SRC={self.src}, DST={self.dst}, PLEN={self.payload_len} NEXT={self.next_header} HLIM={self.hop_limit}) DATA={self.data}"
         )
 
     @classmethod

xcloud/protocol/teredo.py

@@ -2,22 +2,74 @@
 from dataclasses import dataclass, field
 from struct import pack, unpack, unpack_from
 from typing import Any, List, Optional, Tuple, Union
+from ipaddress import IPv4Address, IPv6Address
 
 from . import ipv6
 
 TEREDO_PORT = 3544
 TEREDO_HEADER_LENGTH = 22
 
+
+@dataclass
+class TeredoEndpoint:
+    teredo_server_ipv4: IPv4Address
+    flags: int
+    udp_port: int
+    client_pub_ipv4: IPv4Address
+
+
+def convert_teredo_addr_to_endpoint(
+    teredo_addr: bytes
+) -> TeredoEndpoint:
+    """
+    0x00-0x04 Prefix                // 32 bits
+    0x04-0x08 Teredo server IPv4    // 32 bits
+    0x08-0x0A Flags                 // 16 bits
+    0x0A-0x0C UDP Port              // 16 bits
+    0x0C-0x10 Client public IPv4    // 32 bits
+
+    Client IP address and UDP port are obfuscated / inverted 
+    """
+    addr = IPv6Address(teredo_addr)
+    teredo_tuple = addr.teredo
+    if not teredo_tuple:
+        raise ValueError('Not a teredo address')
+
+    prefix, server_ipv4, flags, udp_port, client_ipv4 = \
+        unpack('!IIHHI', teredo_addr)
+    
+    # Deobfuscate/invert client address and port
+    client_ipv4 ^= 0xFFFFFFFF
+    udp_port ^= 0xFFFF
+
+    # Convert IP addresses to object
+    server_ipv4 = IPv4Address(server_ipv4)
+    client_ipv4 = IPv4Address(client_ipv4)
+    
+    assert server_ipv4 == teredo_tuple[0]
+    assert client_ipv4 == teredo_tuple[1]
+
+    return TeredoEndpoint(
+        teredo_server_ipv4=server_ipv4,
+        flags=flags,
+        udp_port=udp_port,
+        client_pub_ipv4=client_ipv4
+    )
+
 class TeredoPacket:
     def __init__(
         self,
-        ipv6_base: ipv6.IPv6Packet
+        ipv6_base: ipv6.IPv6Packet,
+        src_teredo: TeredoEndpoint,
+        dst_teredo: TeredoEndpoint
     ) -> None:
         self.ipv6 = ipv6_base
+        self.src_teredo = src_teredo
+        self.dst_teredo = dst_teredo
 
     def __repr__(self) -> str:
         return (
-            f"TeredoPacket(IPv6={self.ipv6})"
+            f"TeredoPacket(IPv6={self.ipv6}, SRC={self.src_teredo}, DST={self.dst_teredo})"
         )
 
     @classmethod
@@ -27,5 +79,7 @@ def parse(cls, data: bytes):
                 f"Teredo packet length is less than {TEREDO_HEADER_LENGTH} bytes"
             )
         base = ipv6.IPv6Packet.parse(data)
-        return cls(base)
+        src_teredo = convert_teredo_addr_to_endpoint(base.src)
+        dst_teredo = convert_teredo_addr_to_endpoint(base.dst)
+        return cls(base, src_teredo, dst_teredo)