Fix GSMTAP linktype + type/sub-type values so Wireshark decodes PCAP
pcap.py: LINKTYPE_GSMTAP (155) was wrong — 155 is DLT_USER8, so Wireshark showed 'USER 8 / DLT=155'. Now write Ethernet-encapsulated PCAP (LINKTYPE 1): each GSMTAP message wrapped in Ethernet + IPv4 + UDP (port 4729), with correct checksums — matching a real tcpdump capture of GSMTAP-over-UDP and how Wireshark's gsmtap dissector is keyed. gsmtap.py: GSMTAP_TYPE_SIM was 0x07 (WiMAX burst) — corrected to 0x04; GSMTAP_SIM_APDU was 0x02 (PPS req) — corrected to 0x00; added PPS_REQ/ PPS_RSP. These match libosmocore/Wireshark packet-gsmtap.h. Verified with tshark: PCAP now decodes as GSM SIM (SELECT/STATUS/ATR) instead of 'USER 8'. tests: pcap tests assert Ethernet linktype + framing + GSMTAP payload.
This commit is contained in:
1 parent
e09aac4cb7
commit
cb4c07f3c2
3 files changed
+76
-20
No files matched your search
@@ -8,10 +8,12 @@ import socket
|
|||||||
|
|
||||||
GSMTAP_VERSION = 0x02
|
GSMTAP_VERSION = 0x02
|
||||||
GSMTAP_HDR_LEN = 4 # in 32-bit words (16 bytes)
|
GSMTAP_HDR_LEN = 4 # in 32-bit words (16 bytes)
|
||||||
GSMTAP_TYPE_SIM = 0x07
|
GSMTAP_TYPE_SIM = 0x04
|
||||||
|
|
||||||
|
GSMTAP_SIM_APDU = 0x00
|
||||||
GSMTAP_SIM_ATR = 0x01
|
GSMTAP_SIM_ATR = 0x01
|
||||||
GSMTAP_SIM_APDU = 0x02
|
GSMTAP_SIM_PPS_REQ = 0x02
|
||||||
|
GSMTAP_SIM_PPS_RSP = 0x03
|
||||||
|
|
||||||
GSMTAP_UDP_PORT = 4729
|
GSMTAP_UDP_PORT = 4729
|
||||||
|
|
||||||
|
|||||||
@@ -1,7 +1,9 @@
|
|||||||
"""PCAP file writer for Wireshark analysis.
|
"""PCAP file writer for Wireshark analysis.
|
||||||
|
|
||||||
Writes a standard PCAP file with LINKTYPE_GSMTAP (155, 0x009b).
|
Writes a standard PCAP file with Ethernet (LINKTYPE 1, DLT_EN10MB)
|
||||||
Each packet is a GSMTAP header + raw APDU/ATR data.
|
encapsulation. Each packet is Ethernet + IPv4 + UDP (port 4729) + GSMTAP,
|
||||||
|
matching a real ``tcpdump -i lo udp port 4729`` capture of GSMTAP-over-UDP
|
||||||
|
(simtrace2-sniff / simtrace2-pysniff output).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import io
|
import io
|
||||||
@@ -9,7 +11,11 @@ import struct
|
|||||||
import time
|
import time
|
||||||
|
|
||||||
PCAP_MAGIC = 0xa1b2c3d4
|
PCAP_MAGIC = 0xa1b2c3d4
|
||||||
LINKTYPE_GSMTAP = 155 # 0x009b
|
LINKTYPE_ETHERNET = 1 # DLT_EN10MB
|
||||||
|
|
||||||
|
GSMTAP_UDP_PORT = 4729
|
||||||
|
|
||||||
|
_LOOPBACK = b'\x7f\x00\x00\x01'
|
||||||
|
|
||||||
_PCAP_GLOBAL_HDR_FMT = '<IHHiIII'
|
_PCAP_GLOBAL_HDR_FMT = '<IHHiIII'
|
||||||
_PCAP_GLOBAL_HDR_SIZE = struct.calcsize(_PCAP_GLOBAL_HDR_FMT)
|
_PCAP_GLOBAL_HDR_SIZE = struct.calcsize(_PCAP_GLOBAL_HDR_FMT)
|
||||||
@@ -18,6 +24,35 @@ _PCAP_PKT_HDR_FMT = '<IIII'
|
|||||||
_PCAP_PKT_HDR_SIZE = struct.calcsize(_PCAP_PKT_HDR_FMT)
|
_PCAP_PKT_HDR_SIZE = struct.calcsize(_PCAP_PKT_HDR_FMT)
|
||||||
|
|
||||||
|
|
||||||
|
def _checksum(data):
|
||||||
|
if len(data) % 2:
|
||||||
|
data += b'\x00'
|
||||||
|
s = sum(struct.unpack('!%dH' % (len(data) // 2), data))
|
||||||
|
s = (s >> 16) + (s & 0xffff)
|
||||||
|
s = (s >> 16) + (s & 0xffff)
|
||||||
|
return ~s & 0xffff
|
||||||
|
|
||||||
|
|
||||||
|
def wrap_gsmtap(gsmtap_payload):
|
||||||
|
"""Wrap a GSMTAP message in Ethernet + IPv4 + UDP (port 4729) headers."""
|
||||||
|
udp_len = 8 + len(gsmtap_payload)
|
||||||
|
|
||||||
|
udp_nocs = struct.pack('!HHHH', GSMTAP_UDP_PORT, GSMTAP_UDP_PORT, udp_len, 0)
|
||||||
|
pseudo = _LOOPBACK + _LOOPBACK + struct.pack('!BBH', 0, 17, udp_len)
|
||||||
|
udp_csum = _checksum(pseudo + udp_nocs + gsmtap_payload)
|
||||||
|
udp = struct.pack('!HHHH', GSMTAP_UDP_PORT, GSMTAP_UDP_PORT, udp_len, udp_csum) + gsmtap_payload
|
||||||
|
|
||||||
|
ip_len = 20 + len(udp)
|
||||||
|
ip_nocs = struct.pack('!BBHHHBBH4s4s', 0x45, 0x00, ip_len, 0, 0x4000, 64, 17, 0,
|
||||||
|
_LOOPBACK, _LOOPBACK)
|
||||||
|
ip_csum = _checksum(ip_nocs)
|
||||||
|
ip = struct.pack('!BBHHHBBH4s4s', 0x45, 0x00, ip_len, 0, 0x4000, 64, 17, ip_csum,
|
||||||
|
_LOOPBACK, _LOOPBACK) + udp
|
||||||
|
|
||||||
|
eth = struct.pack('!6s6sH', b'\x00' * 6, b'\x00' * 6, 0x0800)
|
||||||
|
return eth + ip
|
||||||
|
|
||||||
|
|
||||||
def build_pcap(packets):
|
def build_pcap(packets):
|
||||||
"""Build a PCAP file in memory.
|
"""Build a PCAP file in memory.
|
||||||
|
|
||||||
@@ -34,10 +69,10 @@ def build_pcap(packets):
|
|||||||
0,
|
0,
|
||||||
0,
|
0,
|
||||||
65535,
|
65535,
|
||||||
LINKTYPE_GSMTAP,
|
LINKTYPE_ETHERNET,
|
||||||
))
|
))
|
||||||
for gsmtap_hdr, data, ts in packets:
|
for gsmtap_hdr, data, ts in packets:
|
||||||
packet = gsmtap_hdr + data
|
packet = wrap_gsmtap(gsmtap_hdr + data)
|
||||||
ts_sec = int(ts)
|
ts_sec = int(ts)
|
||||||
ts_usec = int((ts - ts_sec) * 1_000_000)
|
ts_usec = int((ts - ts_sec) * 1_000_000)
|
||||||
buf.write(struct.pack(
|
buf.write(struct.pack(
|
||||||
@@ -52,7 +87,7 @@ def build_pcap(packets):
|
|||||||
|
|
||||||
|
|
||||||
class PcapWriter:
|
class PcapWriter:
|
||||||
"""Write sniffed messages as GSMTAP-encapsulated PCAP packets."""
|
"""Write sniffed messages as Ethernet/IP/UDP/GSMTAP PCAP packets."""
|
||||||
|
|
||||||
def __init__(self, path):
|
def __init__(self, path):
|
||||||
self._f = open(path, 'wb')
|
self._f = open(path, 'wb')
|
||||||
@@ -67,12 +102,12 @@ class PcapWriter:
|
|||||||
0, # thiszone (GMT)
|
0, # thiszone (GMT)
|
||||||
0, # sigfigs
|
0, # sigfigs
|
||||||
65535, # snaplen
|
65535, # snaplen
|
||||||
LINKTYPE_GSMTAP, # network (LINKTYPE_GSMTAP)
|
LINKTYPE_ETHERNET, # network (Ethernet)
|
||||||
)
|
)
|
||||||
self._f.write(hdr)
|
self._f.write(hdr)
|
||||||
|
|
||||||
def write_packet(self, gsmtap_hdr, data, ts=None):
|
def write_packet(self, gsmtap_hdr, data, ts=None):
|
||||||
"""Write a single GSMTAP-encapsulated packet.
|
"""Write a single GSMTAP-encapsulated packet (wrapped in IP/UDP/Eth).
|
||||||
|
|
||||||
*gsmtap_hdr* is the 16-byte GSMTAP header (already packed).
|
*gsmtap_hdr* is the 16-byte GSMTAP header (already packed).
|
||||||
*data* is the raw APDU/ATR bytes.
|
*data* is the raw APDU/ATR bytes.
|
||||||
@@ -80,7 +115,7 @@ class PcapWriter:
|
|||||||
if ts is None:
|
if ts is None:
|
||||||
ts = time.time()
|
ts = time.time()
|
||||||
|
|
||||||
packet = gsmtap_hdr + data
|
packet = wrap_gsmtap(gsmtap_hdr + data)
|
||||||
ts_sec = int(ts)
|
ts_sec = int(ts)
|
||||||
ts_usec = int((ts - ts_sec) * 1_000_000)
|
ts_usec = int((ts - ts_sec) * 1_000_000)
|
||||||
|
|
||||||
|
|||||||
+28
-9
@@ -3,13 +3,14 @@
|
|||||||
import struct
|
import struct
|
||||||
import unittest
|
import unittest
|
||||||
|
|
||||||
from simtrace2_pysniff.pcap import build_pcap, PCAP_MAGIC, LINKTYPE_GSMTAP
|
from simtrace2_pysniff.pcap import build_pcap, PCAP_MAGIC, LINKTYPE_ETHERNET
|
||||||
from simtrace2_pysniff.server.server import _content_disposition
|
from simtrace2_pysniff.server.server import _content_disposition
|
||||||
|
|
||||||
|
FRAMING = 14 + 20 + 8 # Ethernet + IPv4 + UDP
|
||||||
|
|
||||||
|
|
||||||
class TestBuildPcap(unittest.TestCase):
|
class TestBuildPcap(unittest.TestCase):
|
||||||
def _gsmtap_hdr(self, sub_type):
|
def _gsmtap_hdr(self, sub_type):
|
||||||
import struct
|
|
||||||
return struct.pack('!BBBBHBBIBBBB', 2, 4, 7, 0, 0, 0, 0, 0, sub_type, 0, 0, 0)
|
return struct.pack('!BBBBHBBIBBBB', 2, 4, 7, 0, 0, 0, 0, 0, sub_type, 0, 0, 0)
|
||||||
|
|
||||||
def test_empty(self):
|
def test_empty(self):
|
||||||
@@ -19,16 +20,34 @@ class TestBuildPcap(unittest.TestCase):
|
|||||||
self.assertEqual(magic, PCAP_MAGIC)
|
self.assertEqual(magic, PCAP_MAGIC)
|
||||||
self.assertEqual((vmaj, vmin), (2, 4))
|
self.assertEqual((vmaj, vmin), (2, 4))
|
||||||
self.assertEqual(snaplen, 65535)
|
self.assertEqual(snaplen, 65535)
|
||||||
self.assertEqual(network, LINKTYPE_GSMTAP)
|
self.assertEqual(network, LINKTYPE_ETHERNET)
|
||||||
|
|
||||||
def test_one_packet(self):
|
def test_one_packet(self):
|
||||||
pkt = build_pcap([(self._gsmtap_hdr(2), b'\xa0\xa4\x00\x00\x02\x3f\x00', 1234567890.5)])
|
pkt = build_pcap([(self._gsmtap_hdr(2), b'\xa0\xa4\x00\x00\x02\x3f\x00', 1234567890.5)])
|
||||||
self.assertEqual(len(pkt), 24 + 16 + (16 + 7)) # global + pkt hdr + gsmtap hdr + data
|
# global + pkt hdr + framing (Eth+IP+UDP) + gsmtap hdr + data
|
||||||
ts_sec, ts_usec, incl_len, orig_len = struct.unpack('<IIII', data_bytes := pkt[24:40])
|
self.assertEqual(len(pkt), 24 + 16 + FRAMING + 16 + 7)
|
||||||
|
ts_sec, ts_usec, incl_len, orig_len = struct.unpack('<IIII', pkt[24:40])
|
||||||
self.assertEqual(ts_sec, 1234567890)
|
self.assertEqual(ts_sec, 1234567890)
|
||||||
self.assertEqual(ts_usec, 500000)
|
self.assertEqual(ts_usec, 500000)
|
||||||
self.assertEqual(incl_len, 23)
|
self.assertEqual(incl_len, FRAMING + 16 + 7)
|
||||||
self.assertEqual(orig_len, 23)
|
self.assertEqual(orig_len, FRAMING + 16 + 7)
|
||||||
|
|
||||||
|
def test_ethernet_and_udp_headers(self):
|
||||||
|
gsmtap_hdr = self._gsmtap_hdr(2)
|
||||||
|
data = b'\xa0\xa4\x00\x00\x02\x3f\x00'
|
||||||
|
pkt = build_pcap([(gsmtap_hdr, data, 1.0)])
|
||||||
|
payload = pkt[24 + 16:] # skip global + pkt header
|
||||||
|
|
||||||
|
# Ethernet: dst MAC, src MAC, ethertype 0x0800 (IPv4)
|
||||||
|
self.assertEqual(payload[12:14], b'\x08\x00')
|
||||||
|
# IPv4: protocol UDP (17), src/dst 127.0.0.1
|
||||||
|
self.assertEqual(payload[23], 17)
|
||||||
|
self.assertEqual(payload[26:30], b'\x7f\x00\x00\x01')
|
||||||
|
self.assertEqual(payload[30:34], b'\x7f\x00\x00\x01')
|
||||||
|
# UDP: dst port 4729
|
||||||
|
self.assertEqual(struct.unpack('!H', payload[36:38])[0], 4729)
|
||||||
|
# GSMTAP payload follows framing
|
||||||
|
self.assertEqual(payload[FRAMING:], gsmtap_hdr + data)
|
||||||
|
|
||||||
def test_multiple_packets(self):
|
def test_multiple_packets(self):
|
||||||
packets = [
|
packets = [
|
||||||
@@ -36,8 +55,8 @@ class TestBuildPcap(unittest.TestCase):
|
|||||||
(self._gsmtap_hdr(2), b'\x80\xf2\x00\x00\x00', 2.0),
|
(self._gsmtap_hdr(2), b'\x80\xf2\x00\x00\x00', 2.0),
|
||||||
]
|
]
|
||||||
data = build_pcap(packets)
|
data = build_pcap(packets)
|
||||||
# global header + 2 packets (each: 16 pkt hdr + 16 gsmtap hdr + data)
|
# global header + 2 packets (each: 16 pkt hdr + framing + 16 gsmtap hdr + data)
|
||||||
self.assertEqual(len(data), 24 + (16 + 16 + 2) + (16 + 16 + 5))
|
self.assertEqual(len(data), 24 + (16 + FRAMING + 16 + 2) + (16 + FRAMING + 16 + 5))
|
||||||
|
|
||||||
|
|
||||||
class TestContentDisposition(unittest.TestCase):
|
class TestContentDisposition(unittest.TestCase):
|
||||||
|
|||||||
Reference in new issue
Block a user