Implement basic PCAP parser

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

@@ -0,0 +1,62 @@
+import dpkt
+from typing import Any
+from enum import Enum
+
+NO_NEXT_HEADER = 59
+
+class IPv6Packet:
+    def __init__(
+        self,
+        ipv6_base: dpkt.ip6.IP6
+    ) -> 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) -> Any:
+        return self.ipv6_base.data
+
+    def __repr__(self):
+        return (
+            f"IPv6Packet(V={self.version}, SRC={self.src}, DST={self.dst}, PLEN={self.payload_len} NEXT={self.next_header} HLIM={self.hop_limit})"
+        )
+
+    @classmethod
+    def parse(cls, data: bytes):
+        ipv6_base = dpkt.ip6.IP6(data)
+        if ipv6_base.v != 6:
+            raise ValueError(
+                f'Invalid IP version: Not 6'
+            )
+        return cls(ipv6_base)

xcloud/protocol/teredo.py

@@ -0,0 +1,85 @@
+import os
+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')
+
+    _, 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,
+        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}, SRC={self.src_teredo}, DST={self.dst_teredo})"
+        )
+
+    @classmethod
+    def parse(cls, data: bytes):
+        if len(data) < TEREDO_HEADER_LENGTH:
+            raise ValueError(
+                f"Teredo packet length is less than {TEREDO_HEADER_LENGTH} bytes"
+            )
+        base = ipv6.IPv6Packet.parse(data)
+        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)
+

xcloud/scripts/pcap_reader.py

@@ -2,13 +2,63 @@
 PCAP Parser for XCloud network traffic
 """
 import argparse
+import logging
+from typing import Any, Optional
 
 import dpkt
+from aiortc import rtp
+from aioice import stun
 from construct.lib import containers
 
+from ..protocol import packets, teredo, ipv6
 
+
+logging.basicConfig(level=logging.DEBUG)
 containers.setGlobalPrintFullStrings(True)
+LOG = logging.getLogger(__name__)
+
+def get_info_stun(stun: stun.Message) -> None:
+    return f'STUN: {stun}'
+
+def get_info_rtp(rtp: rtp.RtpPacket) -> None:
+    try:
+        payload_name = packets.PayloadType(rtp.payload_type)
+    except:
+        payload_name = '<UNKNOWN>'
+
+    return f'RTP: {payload_name.name} {rtp} SSRC={rtp.ssrc}'
 
+def get_info_teredo(teredo: teredo.TeredoPacket) -> None:
+    info = f'TEREDO: {teredo}'
+    if teredo.ipv6.next_header != ipv6.NO_NEXT_HEADER:
+        data = teredo.ipv6.data
+        if type(data) == bytes:
+            raise ValueError(f'TEREDO contains unparsed-subpacket: {data}')
+        subpacket_info = get_info_general(data)
+        info += f'\n -> TEREDO-WRAPPED: {subpacket_info}'
+    return info
+
+
+PACKET_TYPES = [
+    (stun.parse_message, get_info_stun),
+    (rtp.RtpPacket.parse, get_info_rtp),
+    (teredo.TeredoPacket.parse, get_info_teredo)
+]
+
+def get_info_general(packet: Any) -> Optional[str]:
+    if isinstance(packet, dpkt.udp.UDP):
+        data = bytes(packet.data)
+        for cls, info_func in PACKET_TYPES:
+            try:
+                instance = cls(data)
+                info = info_func(instance)
+                return info
+            except:
+                pass
+    elif isinstance(packet, bytes):
+        return '<RAW BYTES>'
+    else:
+        return '<UNHANDLED>'
 
 def packet_filter(filepath):
     with open(filepath, 'rb') as fh:
@@ -20,16 +70,18 @@ def packet_filter(filepath):
                 continue
 
             ip = eth.data
-            if not isinstance(ip.data, dpkt.udp.UDP):
+            subpacket = ip.data
+            if not isinstance(subpacket, dpkt.udp.UDP):
                 continue
 
-            yield(ip.data, ts)
+            yield(subpacket, ts)
 
 
 def parse_file(pcap_filepath: str) -> None:
     for packet, timestamp in packet_filter(pcap_filepath):
-        print(packet, timestamp)
-
+        info = get_info_general(packet)
+        if info:
+            print(info)
 
 def main():
     parser = argparse.ArgumentParser(