Add session export to PCAP

pcap.py: build_pcap(packets) — build PCAP bytes in memory (same format
as PcapWriter)
database.py: get_messages_raw() — raw elapsed/type/data without decode
server.py: GET /api/sessions/<id>/pcap generates a GSMTAP-encapsulated
PCAP (ATR/TPDU) with real timestamps and downloads it as
<session-name>.pcap (ASCII filename fallback + RFC 5987 UTF-8 for
spaces/unicode via _content_disposition)

tests: build_pcap + _content_disposition (67 total)

simtrace2_pysniff/pcap.py

@@ -4,6 +4,7 @@ Writes a standard PCAP file with LINKTYPE_GSMTAP (155, 0x009b).
 Each packet is a GSMTAP header + raw APDU/ATR data.
 """
 
+import io
 import struct
 import time
 
@@ -17,6 +18,39 @@ _PCAP_PKT_HDR_FMT = '<IIII'
 _PCAP_PKT_HDR_SIZE = struct.calcsize(_PCAP_PKT_HDR_FMT)
 
 
+def build_pcap(packets):
+    """Build a PCAP file in memory.
+
+    *packets* is a list of ``(gsmtap_hdr, data, ts)`` tuples, where
+    gsmtap_hdr is the packed 16-byte GSMTAP header and ts is a
+    seconds-since-epoch float.  Returns the PCAP file contents as bytes.
+    """
+    buf = io.BytesIO()
+    buf.write(struct.pack(
+        _PCAP_GLOBAL_HDR_FMT,
+        PCAP_MAGIC,
+        2,
+        4,
+        0,
+        0,
+        65535,
+        LINKTYPE_GSMTAP,
+    ))
+    for gsmtap_hdr, data, ts in packets:
+        packet = gsmtap_hdr + data
+        ts_sec = int(ts)
+        ts_usec = int((ts - ts_sec) * 1_000_000)
+        buf.write(struct.pack(
+            _PCAP_PKT_HDR_FMT,
+            ts_sec,
+            ts_usec,
+            len(packet),
+            len(packet),
+        ))
+        buf.write(packet)
+    return buf.getvalue()
+
+
 class PcapWriter:
     """Write sniffed messages as GSMTAP-encapsulated PCAP packets."""
 

simtrace2_pysniff/server/database.py

@@ -160,6 +160,12 @@ class Database:
         initial_prev = self._last_tpdu_context(session_id, after_id + 1)
         return self._decode_rows(rows, initial_prev=initial_prev)
 
+    def get_messages_raw(self, session_id):
+        rows = self._conn.execute(
+            'SELECT elapsed, type, data FROM messages WHERE session_id=? ORDER BY elapsed, id',
+            (session_id,)).fetchall()
+        return [{'elapsed': r[0], 'type': r[1], 'data': r[2]} for r in rows]
+
     def count_messages(self, session_id):
         row = self._conn.execute(
             'SELECT COUNT(*) FROM messages WHERE session_id=?',

simtrace2_pysniff/server/server.py

@@ -4,11 +4,21 @@ import json
 import os
 import sys
 import threading
+import time
+from datetime import datetime
 from http.server import HTTPServer, BaseHTTPRequestHandler
-from urllib.parse import urlparse, parse_qs
+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
+
+
+def _content_disposition(filename):
+    """RFC 6266/5987 Content-Disposition filename (ASCII fallback + UTF-8)."""
+    fallback = filename.encode('ascii', 'replace').decode('ascii').replace('"', '_')
+    return f"attachment; filename=\"{fallback}\"; filename*=UTF-8''{quote(filename, safe='')}"
 
 
 class RequestHandler(BaseHTTPRequestHandler):
@@ -29,6 +39,15 @@ class RequestHandler(BaseHTTPRequestHandler):
         self.end_headers()
         self.wfile.write(body)
 
+    def _send_binary(self, data, content_type, filename):
+        self.send_response(200)
+        self.send_header('Content-Type', content_type)
+        self.send_header('Content-Length', len(data))
+        self.send_header('Content-Disposition', _content_disposition(filename))
+        self.send_header('Access-Control-Allow-Origin', '*')
+        self.end_headers()
+        self.wfile.write(data)
+
     def _send_error(self, status, message):
         self._send_json({'error': message}, status)
 
@@ -59,6 +78,9 @@ class RequestHandler(BaseHTTPRequestHandler):
             self._handle_status()
         elif path == '/api/sessions':
             self._handle_list_sessions()
+        elif path.startswith('/api/sessions/') and path.endswith('/pcap'):
+            session_id = int(path.split('/')[-2])
+            self._handle_get_session_pcap(session_id)
         elif path.startswith('/api/sessions/'):
             session_id = int(path.split('/')[-1])
             self._handle_get_session(session_id, params)
@@ -149,6 +171,32 @@ class RequestHandler(BaseHTTPRequestHandler):
             'type_counts': type_counts,
         })
 
+    def _handle_get_session_pcap(self, session_id):
+        session = self.db.get_session(session_id)
+        if session is None:
+            self._send_error(404, 'Session not found')
+            return
+
+        try:
+            start_ts = datetime.fromisoformat(session['started']).timestamp()
+        except (ValueError, TypeError):
+            start_ts = time.time()
+
+        packets = []
+        for m in self.db.get_messages_raw(session_id):
+            if m['type'] == 'atr':
+                sub_type = GSMTAP_SIM_ATR
+            elif m['type'] == 'tpdu':
+                sub_type = GSMTAP_SIM_APDU
+            else:
+                continue
+            gsmtap_hdr = build_gsmtap_packet(sub_type, b'')[:16]
+            packets.append((gsmtap_hdr, m['data'], start_ts + m['elapsed']))
+
+        pcap = build_pcap(packets)
+        name = (session.get('name') or f'session-{session_id}') + '.pcap'
+        self._send_binary(pcap, 'application/vnd.tcpdump.pcap', name)
+
     def _handle_capture_latest(self, params):
         after_id = int(params.get('after', [0])[0])
         if not self.capture.active:

tests/test_pcap.py

@@ -0,0 +1,60 @@
+"""Tests for PCAP generation and Content-Disposition filename."""
+
+import struct
+import unittest
+
+from simtrace2_pysniff.pcap import build_pcap, PCAP_MAGIC, LINKTYPE_GSMTAP
+from simtrace2_pysniff.server.server import _content_disposition
+
+
+class TestBuildPcap(unittest.TestCase):
+    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)
+
+    def test_empty(self):
+        data = build_pcap([])
+        self.assertEqual(len(data), 24)  # just the global header
+        magic, vmaj, vmin, _tz, _sig, snaplen, network = struct.unpack('<IHHiIII', data[:24])
+        self.assertEqual(magic, PCAP_MAGIC)
+        self.assertEqual((vmaj, vmin), (2, 4))
+        self.assertEqual(snaplen, 65535)
+        self.assertEqual(network, LINKTYPE_GSMTAP)
+
+    def test_one_packet(self):
+        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
+        ts_sec, ts_usec, incl_len, orig_len = struct.unpack('<IIII', data_bytes := pkt[24:40])
+        self.assertEqual(ts_sec, 1234567890)
+        self.assertEqual(ts_usec, 500000)
+        self.assertEqual(incl_len, 23)
+        self.assertEqual(orig_len, 23)
+
+    def test_multiple_packets(self):
+        packets = [
+            (self._gsmtap_hdr(1), b'\x3b\x00', 1.0),
+            (self._gsmtap_hdr(2), b'\x80\xf2\x00\x00\x00', 2.0),
+        ]
+        data = build_pcap(packets)
+        # global header + 2 packets (each: 16 pkt hdr + 16 gsmtap hdr + data)
+        self.assertEqual(len(data), 24 + (16 + 16 + 2) + (16 + 16 + 5))
+
+
+class TestContentDisposition(unittest.TestCase):
+    def test_ascii(self):
+        h = _content_disposition('My Capture.pcap')
+        self.assertIn('filename="My Capture.pcap"', h)
+
+    def test_unicode(self):
+        h = _content_disposition('Моя сессия.pcap')
+        # ASCII fallback uses '?' replacements; UTF-8 filename* carries the real name
+        self.assertIn("filename*=UTF-8''", h)
+        self.assertIn('%D0%9C', h)  # 'М' = %D0%9C
+
+    def test_special_chars(self):
+        h = _content_disposition('a"b.pcap')
+        self.assertIn('filename="a_b.pcap"', h)
+
+
+if __name__ == '__main__':
+    unittest.main()