PCAP: Parse teredo-wrapped payloads

Author: tuxuser

xcloud/protocol/ipv6.py

@@ -1,4 +1,8 @@
 import dpkt
+from typing import Any
+from enum import Enum
+
+NO_NEXT_HEADER = 59
 
 class IPv6Packet:
     def __init__(
@@ -40,12 +44,12 @@ def dst(self) -> bytes:
         return self.ipv6_base.dst
 
     @property
-    def data(self) -> bytes:
+    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}) 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})"
         )
 
     @classmethod

xcloud/protocol/teredo.py

@@ -35,7 +35,7 @@ def convert_teredo_addr_to_endpoint(
     if not teredo_tuple:
         raise ValueError('Not a teredo address')
 
-    prefix, server_ipv4, flags, udp_port, client_ipv4 = \
+    _, server_ipv4, flags, udp_port, client_ipv4 = \
         unpack('!IIHHI', teredo_addr)
 
     # Deobfuscate/invert client address and port

xcloud/scripts/pcap_reader.py

@@ -3,39 +3,63 @@
 """
 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
+from ..protocol import packets, teredo, ipv6
 
 
 logging.basicConfig(level=logging.DEBUG)
 containers.setGlobalPrintFullStrings(True)
 LOG = logging.getLogger(__name__)
 
-def print_stun(stun: stun.Message) -> None:
-    print(f'STUN: {stun}')
+def get_info_stun(stun: stun.Message) -> None:
+    return f'STUN: {stun}'
 
-def print_rtp(rtp: rtp.RtpPacket) -> None:
+def get_info_rtp(rtp: rtp.RtpPacket) -> None:
     try:
         payload_name = packets.PayloadType(rtp.payload_type)
     except:
         payload_name = '<UNKNOWN>'
 
-    print(f'RTP: {payload_name.name} {rtp}')
+    return f'RTP: {payload_name.name} {rtp}'
+
+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
 
-def print_teredo(teredo: teredo.TeredoPacket) -> None:
-    print(f'TEREDO: {teredo}')
 
 PACKET_TYPES = [
-    (stun.parse_message, print_stun),
-    (rtp.RtpPacket.parse, print_rtp),
-    (teredo.TeredoPacket.parse, print_teredo)
+    (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:
         for ts, buf in dpkt.pcap.Reader(fh):
@@ -46,23 +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):
-        packet = bytes(packet.data)
-        for cls, print_func in PACKET_TYPES:
-            try:
-                instance = cls(packet)
-                print_func(instance)
-            except:
-                #LOG.exception('Failed to parse')
-                pass
-        
+        info = get_info_general(packet)
+        if info:
+            print(info)
 
 def main():
     parser = argparse.ArgumentParser(