From dd34c258994c238e67e4bb2e9f519458ebc91d75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=D0=90=D0=BD=D1=82=D0=BE=D0=BD=20=D0=A2=D1=80=D0=BE=D1=88?= =?UTF-8?q?=D0=B8=D0=BD?= Date: Thu, 13 Aug 2026 10:15:22 +0300 Subject: [PATCH] Add PCAP/PCAPNG import via POST /api/sessions/import - pcap.py: parse_pcap (micro/nano, both byte orders) and parse_pcapng (SHB byte-order, IDB linktype + if_tsresol, EPB/SPB) readers that unwrap Ethernet/RAW/IPv4/NULL/SLL/SLL2 to UDP and extract GSMTAP-SIM packets (APDU->tpdu, ATR->atr, PPS->pps), with a heuristic fallback - database: insert_messages bulk-insert helper - server: import endpoint reads raw body, auto-detects format, skips non-GSMTAP packets; 400 'No GSMTAP data found in trace' (no session) when empty, else creates a 'pcap' session named after the file Tests: 102 (pcap round-trip, pcapng, no-GSMTAP, endpoint 200/400). --- simtrace2_pysniff/pcap.py | 260 ++++++++++++++++++++++++++- simtrace2_pysniff/server/database.py | 8 + simtrace2_pysniff/server/server.py | 47 ++++- tests/test_pcap_import.py | 141 +++++++++++++++ 4 files changed, 453 insertions(+), 3 deletions(-) create mode 100644 tests/test_pcap_import.py diff --git a/simtrace2_pysniff/pcap.py b/simtrace2_pysniff/pcap.py index e3b82a9..8578555 100644 --- a/simtrace2_pysniff/pcap.py +++ b/simtrace2_pysniff/pcap.py @@ -1,9 +1,12 @@ -"""PCAP file writer for Wireshark analysis. +"""PCAP file writer and reader for Wireshark analysis. Writes a standard PCAP file with Ethernet (LINKTYPE 1, DLT_EN10MB) 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). + +Reads classic PCAP (micro/nano, both byte orders) and PCAPNG files, +extracting GSMTAP-SIM packets (APDU/ATR/PPS) for import into a session. """ import io @@ -23,6 +26,38 @@ _PCAP_GLOBAL_HDR_SIZE = struct.calcsize(_PCAP_GLOBAL_HDR_FMT) _PCAP_PKT_HDR_FMT = '> 4) != 4: + return None + ihl = (ip[0] & 0x0F) * 4 + if len(ip) < ihl + 8 or ip[9] != 17: # protocol UDP + return None + udp = ip[ihl:] + udp_len = struct.unpack('!H', udp[4:6])[0] + if udp_len < 8 or len(udp) < 8: + return None + return udp[8:udp_len] + + +def _frame_to_udp_payload(frame, linktype): + """Unwrap a link-layer frame down to the UDP payload (bytes or None).""" + if linktype == LINKTYPE_ETHERNET: + if len(frame) < 14: + return None + ethertype = frame[12:14] + if ethertype == b'\x08\x00': # IPv4 + return _ipv4_udp_payload(frame[14:]) + # IPv6 and non-IP ethertypes are not GSMTAP-over-UDP for our use + return None + if linktype in (LINKTYPE_RAW, LINKTYPE_IPV4): + return _ipv4_udp_payload(frame) + if linktype == LINKTYPE_NULL: + if len(frame) < 4: + return None + family = struct.unpack(' n - off: + break + frame = data[off:off + incl_len] + off += incl_len + ts = ts_sec + ts_frac / (1_000_000_000 if ns else 1_000_000) + parsed = extract_gsmtap(frame, linktype) + if parsed is None: + continue + sub_type, payload = parsed + msg_type = _gsmtap_to_msg(sub_type, payload) + if msg_type is not None: + yield ts, msg_type, payload + + +def parse_pcapng(data): + """Parse a PCAPNG file, yielding (ts, msg_type, payload) tuples. + + Only GSMTAP-SIM packets are yielded. Timestamps are raw + (ts_high << 32 | ts_low) scaled by each interface's resolution. + """ + if len(data) < 12 or struct.unpack(' linktype + tsresols = {} # interface_id -> seconds per tick + spb_ts = 0 # file-order timestamp for SPB blocks (no timestamp) + off = 0 + n = len(data) + while off + 8 <= n: + block_type = struct.unpack(endian + 'I', data[off:off + 4])[0] + total_len = struct.unpack(endian + 'I', data[off + 4:off + 8])[0] + if total_len < 12 or off + total_len > n: + break + body = data[off + 8:off + total_len - 4] + off += total_len + + if block_type == _PNG_IDB and len(body) >= 8: + linktype, _res, _snaplen = struct.unpack( + endian + 'HHI', body[:8]) + iface = len(linktypes) # IDBs are numbered by appearance + linktypes[iface] = linktype + tsresols[iface] = _parse_tsresol(body[8:], endian) + elif block_type == _PNG_EPB and len(body) >= 20: + iface, ts_hi, ts_lo, cap_len, _orig = struct.unpack( + endian + 'IIIII', body[:20]) + if cap_len <= len(body) - 20: + frame = body[20:20 + cap_len] + ts = _scale_ts(ts_hi, ts_lo, tsresols.get(iface)) + parsed = extract_gsmtap(frame, linktypes.get(iface)) + if parsed is not None: + sub_type, payload = parsed + msg_type = _gsmtap_to_msg(sub_type, payload) + if msg_type is not None: + yield ts, msg_type, payload + elif block_type == _PNG_SPB and len(body) >= 4: + cap_len, = struct.unpack(endian + 'I', body[:4]) + if cap_len <= len(body) - 4: + frame = body[4:4 + cap_len] + # SPB has no timestamp and no interface; use file-order ts. + parsed = _scan_gsmtap(frame) + if parsed is not None: + sub_type, payload = parsed + msg_type = _gsmtap_to_msg(sub_type, payload) + if msg_type is not None: + yield float(spb_ts), msg_type, payload + spb_ts += 1 + + +def _parse_tsresol(options, endian): + """Parse if_tsresol from IDB options; return seconds per timestamp tick.""" + i = 0 + n = len(options) + while i + 4 <= n: + code, length = struct.unpack(endian + 'HH', options[i:i + 4]) + i += 4 + if code == 9 and length == 1: + b = options[i] + if b & 0x80: + return 10 ** (-(b & 0x7F)) + return 10 ** (-b) + i += length + (length & 1) # options are padded to 32-bit + return 1e-6 # default: microseconds + + +def _scale_ts(ts_hi, ts_lo, tsresol): + """Combine 64-bit pcapng timestamp into seconds since epoch (approx).""" + raw = (ts_hi << 32) | ts_lo + return raw * (tsresol if tsresol is not None else 1e-6) diff --git a/simtrace2_pysniff/server/database.py b/simtrace2_pysniff/server/database.py index c4f9699..03d3ebd 100644 --- a/simtrace2_pysniff/server/database.py +++ b/simtrace2_pysniff/server/database.py @@ -134,6 +134,14 @@ class Database: self._conn.commit() return cur.lastrowid + def insert_messages(self, session_id, rows): + """Bulk-insert messages; *rows* is an iterable of + (elapsed, msg_type, data, flags) tuples. Returns the last rowid.""" + self._conn.executemany( + 'INSERT INTO messages (session_id, elapsed, type, data, flags) VALUES (?,?,?,?,?)', + [(session_id, elapsed, msg_type, data, flags) for elapsed, msg_type, data, flags in rows]) + self._conn.commit() + def get_message(self, msg_id): row = self._conn.execute( 'SELECT id, session_id, elapsed, type, data, flags FROM messages WHERE id=?', diff --git a/simtrace2_pysniff/server/server.py b/simtrace2_pysniff/server/server.py index 4ffb53b..f782204 100644 --- a/simtrace2_pysniff/server/server.py +++ b/simtrace2_pysniff/server/server.py @@ -12,7 +12,7 @@ from urllib.parse import urlparse, parse_qs, quote from .database import Database from .capture import CaptureManager, GsmtapListener, DirectSniffer from ..gsmtap import build_gsmtap_packet, GSMTAP_SIM_ATR, GSMTAP_SIM_APDU -from ..pcap import build_pcap +from ..pcap import build_pcap, parse_pcap, parse_pcapng def _content_disposition(filename): @@ -57,6 +57,12 @@ class RequestHandler(BaseHTTPRequestHandler): return {} return json.loads(self.rfile.read(length)) + def _read_body(self): + length = int(self.headers.get('Content-Length', 0)) + if length == 0: + return b'' + return self.rfile.read(length) + def _parse_path(self): parsed = urlparse(self.path) path = parsed.path.rstrip('/') @@ -104,12 +110,14 @@ class RequestHandler(BaseHTTPRequestHandler): # --- POST --- def do_POST(self): - path, _ = self._parse_path() + path, params = self._parse_path() if path == '/api/capture/start': self._handle_capture_start() elif path == '/api/capture/stop': self._handle_capture_stop() + elif path == '/api/sessions/import': + self._handle_import_pcap(params) else: self._send_error(404, 'Not found') @@ -236,6 +244,41 @@ class RequestHandler(BaseHTTPRequestHandler): 'messages_count': self.db.count_messages(session_id), }) + def _handle_import_pcap(self, params): + raw = self._read_body() + name = (params.get('name', [''])[0]).strip() + + if not raw: + self._send_error(400, 'Empty file') + return + + if raw[:4] == b'\x0a\x0d\x0d\x0a': + packets = list(parse_pcapng(raw)) + else: + packets = list(parse_pcap(raw)) + + if not packets: + self._send_error(400, 'No GSMTAP data found in trace') + return + + session_id = self.db.create_session('pcap') + if name: + self.db.rename_session(session_id, name) + + first_ts = packets[0][0] + rows = [] + for ts, msg_type, payload in packets: + elapsed = round(ts - first_ts, 6) if ts is not None else 0.0 + rows.append((max(0.0, elapsed), msg_type, payload, 0)) + self.db.insert_messages(session_id, rows) + self.db.close_session(session_id) + + self._send_json({ + 'session_id': session_id, + 'name': name, + 'message_count': len(packets), + }) + def _handle_rename_session(self, session_id, name): self.db.rename_session(session_id, name) self._send_json({'ok': True}) diff --git a/tests/test_pcap_import.py b/tests/test_pcap_import.py new file mode 100644 index 0000000..04d7d46 --- /dev/null +++ b/tests/test_pcap_import.py @@ -0,0 +1,141 @@ +"""Tests for PCAP/PCAPNG import parsing and the import endpoint.""" + +import os +import struct +import tempfile +import unittest + +from simtrace2_pysniff.pcap import ( + build_pcap, parse_pcap, parse_pcapng, wrap_gsmtap, extract_gsmtap, + LINKTYPE_ETHERNET, +) +from simtrace2_pysniff.gsmtap import build_gsmtap_packet, GSMTAP_SIM_ATR, GSMTAP_SIM_APDU + + +def _gsmtap_hdr(sub_type): + return build_gsmtap_packet(sub_type, b'')[:16] + + +def _pcapng_block(block_type, body): + body = body + b'\x00' * ((-len(body)) % 4) + total = 8 + len(body) + 4 + return struct.pack('>II', block_type, total) + body + struct.pack('>I', total) + + +def _build_pcapng(packets): + """Build a minimal big-endian PCAPNG with one Ethernet interface. + + *packets* is a list of (gsmtap_hdr, data, ts_hi, ts_lo) tuples. + """ + shb = _pcapng_block(0x0A0D0D0A, struct.pack('>IHHq', 0x1A2B3C4D, 1, 0, -1)) + idb = _pcapng_block(0x00000001, + struct.pack('>HHI', LINKTYPE_ETHERNET, 0, 65535) + + struct.pack('>HH', 0, 0)) + out = shb + idb + for gsmtap_hdr, data, ts_hi, ts_lo in packets: + frame = wrap_gsmtap(gsmtap_hdr + data) + epb = _pcapng_block( + 0x00000006, + struct.pack('>IIIII', 0, ts_hi, ts_lo, len(frame), len(frame)) + frame) + out += epb + return out + + +class TestParsePcap(unittest.TestCase): + def test_roundtrip(self): + packets = [ + (_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x00', 1.0), + (_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 2.5), + ] + data = build_pcap(packets) + parsed = [(round(ts, 1), t, payload) for ts, t, payload in parse_pcap(data)] + self.assertEqual(parsed, [ + (1.0, 'atr', b'\x3b\x00'), + (2.5, 'tpdu', b'\x80\xf2\x00\x00\x00'), + ]) + + def test_no_gsmtap(self): + # A packet that is not GSMTAP-SIM yields nothing. + frame = wrap_gsmtap(b'\x03' + b'\x00' * 15) # type=3, not SIM + # build a fake pcap with an Ethernet frame that has no GSMTAP + packets = [(b'', b'\x00' * 64, 1.0)] + data = build_pcap(packets) + self.assertEqual(list(parse_pcap(data)), []) + + +class TestParsePcapng(unittest.TestCase): + def test_parse(self): + data = _build_pcapng([ + (_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x04', 0, 0), + (_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 0, 1), + ]) + parsed = [(t, payload) for _ts, t, payload in parse_pcapng(data)] + self.assertEqual(parsed, [ + ('atr', b'\x3b\x04'), + ('tpdu', b'\x80\xf2\x00\x00\x00'), + ]) + + def test_no_gsmtap(self): + # IDB + EPB carrying a GSMTAP header whose type != SIM → no messages. + shb = _pcapng_block(0x0A0D0D0A, struct.pack('>IHHq', 0x1A2B3C4D, 1, 0, -1)) + idb = _pcapng_block(0x00000001, + struct.pack('>HHI', LINKTYPE_ETHERNET, 0, 65535) + + struct.pack('>HH', 0, 0)) + non_sim = struct.pack('!BBBBHBBIBBBB', 2, 4, 7, 0, 0, 0, 0, 0, 0, 0, 0, 0) + b'\x00' + epb = _pcapng_block(0x00000006, + struct.pack('>IIIII', 0, 0, 0, len(non_sim), len(non_sim)) + non_sim) + self.assertEqual(list(parse_pcapng(shb + idb + epb)), []) + + +class TestImportEndpoint(unittest.TestCase): + def _handler(self, body): + from simtrace2_pysniff.server.server import RequestHandler + from simtrace2_pysniff.server.database import Database + + class Fake(RequestHandler): + def __init__(self, body, db): + self.db = db + self._body = body + self._status = None + self._json = None + + def _read_body(self): + return self._body + + def _send_json(self, data, status=200): + self._status = status + self._json = data + + def _send_error(self, status, message): + self._status = status + self._json = {'error': message} + + tmp = tempfile.NamedTemporaryFile(suffix='.db', delete=False) + tmp.close() + db = Database(tmp.name) + self.addCleanup(lambda: (db._conn.close(), os.remove(tmp.name))) + return Fake(body, db), tmp.name + + def test_import_pcap_creates_session(self): + packets = [ + (_gsmtap_hdr(GSMTAP_SIM_ATR), b'\x3b\x00', 1.0), + (_gsmtap_hdr(GSMTAP_SIM_APDU), b'\x80\xf2\x00\x00\x00', 2.0), + ] + h, _ = self._handler(build_pcap(packets)) + h._handle_import_pcap({'name': ['mytrace']}) + self.assertEqual(h._status, 200) + self.assertEqual(h._json['name'], 'mytrace') + self.assertEqual(h._json['message_count'], 2) + self.assertEqual(h.db.count_messages(h._json['session_id']), 2) + + def test_import_no_gsmtap_400(self): + # Ethernet frame with no GSMTAP → no usable data. + h, _ = self._handler(build_pcap([(b'', b'\x00' * 64, 1.0)])) + h._handle_import_pcap({'name': ['empty']}) + self.assertEqual(h._status, 400) + self.assertIn('No GSMTAP data', h._json['error']) + self.assertEqual(h.db.count_sessions(), 0) + + +if __name__ == '__main__': + unittest.main()