384 lines
14 KiB
Python
384 lines
14 KiB
Python
"""HTTP API + static PWA server for simtrace2-pysniff."""
|
|
|
|
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, 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, parse_pcap, parse_pcapng
|
|
from .. import __version__
|
|
|
|
|
|
# Static file serving (the PWA lives in <repo>/frontend, served by this server
|
|
# so the UI and the API share an origin and no CORS/PNA is involved).
|
|
_STATIC_MIME = {
|
|
'.html': 'text/html; charset=utf-8',
|
|
'.css': 'text/css',
|
|
'.js': 'application/javascript',
|
|
'.json': 'application/json',
|
|
'.png': 'image/png',
|
|
'.svg': 'image/svg+xml',
|
|
'.ico': 'image/x-icon',
|
|
'.webmanifest': 'application/manifest+json',
|
|
'.map': 'application/json',
|
|
'.wasm': 'application/wasm',
|
|
}
|
|
|
|
|
|
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):
|
|
db: Database = None
|
|
capture: CaptureManager = None
|
|
capture_mode: str = 'gsmtap'
|
|
|
|
def log_message(self, fmt, *args):
|
|
print(f'[{self.log_date_time_string()}] {args[0]}', file=sys.stderr)
|
|
|
|
def _send_json(self, data, status=200):
|
|
body = json.dumps(data, ensure_ascii=False).encode('utf-8')
|
|
self.send_response(status)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.send_header('Content-Length', len(body))
|
|
self.send_header('Access-Control-Allow-Origin', '*')
|
|
self.send_header('Access-Control-Allow-Methods', 'GET, POST, PATCH, DELETE, OPTIONS')
|
|
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
|
self.send_header('Access-Control-Allow-Private-Network', 'true')
|
|
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.send_header('Access-Control-Allow-Private-Network', 'true')
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
def _send_error(self, status, message):
|
|
self._send_json({'error': message}, status)
|
|
|
|
def _read_json(self):
|
|
length = int(self.headers.get('Content-Length', 0))
|
|
if length == 0:
|
|
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('/')
|
|
return path, parse_qs(parsed.query)
|
|
|
|
def _serve_static(self):
|
|
web_dir = getattr(self.server, 'web_dir', None)
|
|
if not web_dir:
|
|
self._send_error(404, 'Not found')
|
|
return
|
|
rel = self.path.split('?', 1)[0].lstrip('/')
|
|
if rel in ('', '/'):
|
|
rel = 'index.html'
|
|
if '..' in rel.split('/') or rel.startswith('/'):
|
|
self._send_error(404, 'Not found')
|
|
return
|
|
fs_path = os.path.join(web_dir, rel)
|
|
if not os.path.isfile(fs_path):
|
|
self._send_error(404, 'Not found')
|
|
return
|
|
with open(fs_path, 'rb') as f:
|
|
data = f.read()
|
|
content_type = _STATIC_MIME.get(os.path.splitext(rel)[1].lower(), 'application/octet-stream')
|
|
self.send_response(200)
|
|
self.send_header('Content-Type', content_type)
|
|
self.send_header('Content-Length', str(len(data)))
|
|
if rel in ('index.html', 'sw.js'):
|
|
self.send_header('Cache-Control', 'no-cache')
|
|
self.end_headers()
|
|
self.wfile.write(data)
|
|
|
|
# --- CORS preflight ---
|
|
def do_OPTIONS(self):
|
|
self.send_response(204)
|
|
self.send_header('Access-Control-Allow-Origin', '*')
|
|
self.send_header('Access-Control-Allow-Methods', 'GET, POST, PATCH, DELETE, OPTIONS')
|
|
self.send_header('Access-Control-Allow-Headers', 'Content-Type')
|
|
self.send_header('Access-Control-Allow-Private-Network', 'true')
|
|
self.end_headers()
|
|
|
|
# --- GET ---
|
|
def do_GET(self):
|
|
path, params = self._parse_path()
|
|
|
|
if path == '/api/status':
|
|
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)
|
|
elif path == '/api/capture/latest':
|
|
self._handle_capture_latest(params)
|
|
elif path == '/api/capture/status':
|
|
self._handle_capture_status()
|
|
elif path.startswith('/api/apdu/search'):
|
|
session_id = int(params.get('session_id', [0])[0])
|
|
query = params.get('q', [''])[0]
|
|
self._handle_search(session_id, query)
|
|
elif path.startswith('/api/apdu/filter'):
|
|
session_id = int(params.get('session_id', [0])[0])
|
|
msg_type = params.get('type', [''])[0]
|
|
self._handle_filter(session_id, msg_type)
|
|
elif path.startswith('/api/apdu/'):
|
|
msg_id = int(path.split('/')[-1])
|
|
self._handle_get_apdu(msg_id)
|
|
else:
|
|
self._serve_static()
|
|
|
|
# --- POST ---
|
|
def do_POST(self):
|
|
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')
|
|
|
|
# --- PATCH ---
|
|
def do_PATCH(self):
|
|
path, _ = self._parse_path()
|
|
|
|
if path.startswith('/api/sessions/') and path.endswith('/name'):
|
|
session_id = int(path.split('/')[-2])
|
|
body = self._read_json()
|
|
self._handle_rename_session(session_id, body.get('name', ''))
|
|
else:
|
|
self._send_error(404, 'Not found')
|
|
|
|
# --- DELETE ---
|
|
def do_DELETE(self):
|
|
path, _ = self._parse_path()
|
|
|
|
if path.startswith('/api/sessions/'):
|
|
session_id = int(path.split('/')[-1])
|
|
self._handle_delete_session(session_id)
|
|
else:
|
|
self._send_error(404, 'Not found')
|
|
|
|
# --- Handlers ---
|
|
|
|
def _handle_status(self):
|
|
active_session = self.db.get_active_session()
|
|
capture = self.capture
|
|
self._send_json({
|
|
'server': 'simtrace-analyser-server',
|
|
'version': __version__,
|
|
'capture_mode': self.capture_mode,
|
|
'capture_active': capture.active if capture else False,
|
|
'session_id': capture.session_id if capture else None,
|
|
'mode': active_session['mode'] if active_session else None,
|
|
'messages_count': self.db.count_messages(capture.session_id) if capture and capture.session_id else 0,
|
|
'device_connected': capture.device_connected if capture else None,
|
|
})
|
|
|
|
def _handle_list_sessions(self):
|
|
sessions = self.db.list_sessions()
|
|
self._send_json({
|
|
'sessions': sessions,
|
|
'total_sessions': self.db.count_sessions(),
|
|
'db_size': self.db.db_size(),
|
|
})
|
|
|
|
def _handle_get_session(self, session_id, params):
|
|
session = self.db.get_session(session_id)
|
|
if session is None:
|
|
self._send_error(404, 'Session not found')
|
|
return
|
|
offset = int(params.get('offset', [0])[0])
|
|
limit = int(params.get('limit', [200])[0])
|
|
messages = self.db.get_messages(session_id, offset=offset, limit=limit)
|
|
total = self.db.count_messages(session_id)
|
|
type_counts = self.db.get_type_counts(session_id)
|
|
self._send_json({
|
|
'session': session,
|
|
'messages': messages,
|
|
'total': total,
|
|
'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 or not self.capture.active:
|
|
self._send_json({'messages': [], 'next_after': after_id, 'active': False,
|
|
'device_connected': self.capture.device_connected if self.capture else None})
|
|
return
|
|
msg_id = self.capture.latest_msg_id
|
|
messages = self.db.get_messages_after(self.capture.session_id, after_id)
|
|
self._send_json({
|
|
'messages': messages,
|
|
'next_after': msg_id,
|
|
'active': True,
|
|
'session_id': self.capture.session_id,
|
|
'device_connected': self.capture.device_connected,
|
|
})
|
|
|
|
def _handle_capture_status(self):
|
|
self._send_json({
|
|
'active': self.capture.active if self.capture else False,
|
|
'session_id': self.capture.session_id if self.capture else None,
|
|
'device_connected': self.capture.device_connected if self.capture else None,
|
|
})
|
|
|
|
def _handle_capture_start(self):
|
|
if self.capture_mode == 'disabled' or self.capture is None:
|
|
self._send_error(403, 'Capture disabled')
|
|
return
|
|
if self.capture.active:
|
|
self.capture.stop_session()
|
|
session_id = self.capture.start_session()
|
|
started = self.db.get_session(session_id)['started']
|
|
self._send_json({'session_id': session_id, 'started': started})
|
|
|
|
def _handle_capture_stop(self):
|
|
if self.capture_mode == 'disabled' or self.capture is None:
|
|
self._send_error(403, 'Capture disabled')
|
|
return
|
|
if not self.capture.active:
|
|
self._send_error(400, 'No active capture')
|
|
return
|
|
session_id = self.capture.stop_session()
|
|
if session_id is None:
|
|
self._send_json({'session_id': None, 'messages_count': 0, 'discarded': True})
|
|
return
|
|
session = self.db.get_session(session_id)
|
|
self._send_json({
|
|
'session_id': session_id,
|
|
'ended': session['ended'],
|
|
'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]
|
|
last_ts = packets[-1][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)
|
|
# Use the packet timestamps as the session window when they are real
|
|
# (classic pcap / pcapng EPB). pcapng SPB-only files have synthetic
|
|
# 0,1,2,… timestamps, so fall back to the import wall-clock time.
|
|
_REAL_TS_MIN = 946684800 # 2000-01-01
|
|
if first_ts is not None and first_ts > _REAL_TS_MIN:
|
|
self.db.set_session_times_from_ts(session_id, first_ts, last_ts)
|
|
else:
|
|
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})
|
|
|
|
def _handle_delete_session(self, session_id):
|
|
if self.capture and self.capture.session_id == session_id:
|
|
self.capture.stop_session()
|
|
self.db.delete_session(session_id)
|
|
self.db.vacuum()
|
|
self._send_json({'ok': True})
|
|
|
|
def _handle_search(self, session_id, query):
|
|
if not query:
|
|
self._send_json({'messages': []})
|
|
return
|
|
messages = self.db.search_messages(session_id, query)
|
|
self._send_json({'messages': messages})
|
|
|
|
def _handle_filter(self, session_id, msg_type):
|
|
if not msg_type:
|
|
self._send_json({'messages': []})
|
|
return
|
|
messages = self.db.filter_messages(session_id, msg_type)
|
|
self._send_json({'messages': messages})
|
|
|
|
def _handle_get_apdu(self, msg_id):
|
|
msg = self.db.get_message(msg_id)
|
|
if msg is None:
|
|
self._send_error(404, 'Message not found')
|
|
return
|
|
self._send_json(msg)
|