simtrace-analyser/simtrace_analyser/server.py

"""HTTP API + static PWA server for SIMtrace Analyser."""

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 .capture import CaptureManager, GsmtapListener, DirectSniffer, _log
from .database import Database
from .decode import looks_like_apdu
from .comprion import (ComprionError, decode_text, looks_like_comprion,
                       parse_comprion)
from simtrace2_pysniff.gsmtap import build_gsmtap_packet, GSMTAP_SIM_ATR, GSMTAP_SIM_APDU
from simtrace2_pysniff.pcap import build_pcap, parse_pcap, parse_pcapng
from .version 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'
    log_requests: bool = False

    def log_message(self, fmt, *args):
        if self.log_requests:
            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, code=None):
        body = {'error': message}
        if code:
            body['code'] = code
        self._send_json(body, 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/') and path.endswith('/files/image'):
            session_id = int(path.split('/')[-3])
            self._handle_session_files_image(session_id, params)
        elif path.startswith('/api/sessions/') and path.endswith('/files'):
            session_id = int(path.split('/')[-2])
            self._handle_session_files(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_session(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)
        _log(f'Session opened: session={session_id} '
             f'name={session.get("name") or "(untitled)"} messages={total}')
        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_session(self, params):
        """Import a saved trace: PCAP/PCAPNG (default) or a COMPRION text
        export (`?format=comprion`)."""
        raw = self._read_body()
        fmt = (params.get('format', [''])[0] or '').strip().lower()
        if fmt in ('', 'pcap', 'pcapng'):
            self._handle_import_pcap(params, raw)
        elif fmt == 'comprion':
            self._handle_import_comprion(params, raw)
        else:
            self._send_error(400, f"Unknown import format '{fmt}' "
                                  "(use 'pcap' or 'comprion')")

    def _handle_import_pcap(self, params, raw=None):
        if raw is None:
            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:
            if looks_like_comprion(raw):
                self._send_error(
                    400,
                    "This looks like a COMPRION export — use 'Import "
                    "COMPRION trace'.",
                    code='comprion_export')
                return
            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 = []
        skipped = 0
        for ts, msg_type, payload, flags in packets:
            # A captured 'tpdu' must at least be APDU-shaped; mis-sampled or
            # character-level streams would only pollute the decode.
            if msg_type == 'tpdu' and not looks_like_apdu(payload):
                skipped += 1
                continue
            elapsed = round(ts - first_ts, 6) if ts is not None else 0.0
            rows.append((max(0.0, elapsed), msg_type, payload, flags))
        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:
            duration = (last_ts - first_ts) if last_ts is not None and first_ts is not None else 0
            self.db.close_session(session_id, duration)

        _log(f'Session imported: session={session_id} '
             f'name={name or "(untitled)"} messages={len(rows)} '
             f'skipped={skipped} '
             f'bytes={sum(len(r[2]) for r in rows)}')
        self._send_json({
            'session_id': session_id,
            'name': name,
            'message_count': len(rows),
            'skipped': skipped,
        })

    def _handle_import_comprion(self, params, raw):
        """Import a COMPRION 'ISO Protocol Layer' text export.

        The body is the export text, or a JSON object ``{text, times_text}``
        when the matching ISO Application Layer export is supplied as the
        time base.
        """
        name = (params.get('name', [''])[0]).strip()

        if not raw:
            self._send_error(400, 'Empty file')
            return

        text = None
        times_text = None
        if raw.lstrip()[:1] == b'{':
            try:
                payload = json.loads(raw.decode('utf-8'))
            except (ValueError, UnicodeDecodeError):
                payload = None
            if isinstance(payload, dict):
                if not isinstance(payload.get('text'), str):
                    self._send_error(400, "JSON import body must carry a "
                                          "'text' string")
                    return
                text = payload['text']
                times_text = payload.get('times_text') or None
                if times_text is not None and not isinstance(times_text, str):
                    self._send_error(400, 'times_text must be a string')
                    return
        if text is None:
            text = decode_text(raw)

        try:
            result = parse_comprion(text, times_text=times_text)
        except ComprionError as e:
            self._send_error(400, e.message, code=e.code)
            return

        meta = result['metadata']
        # The file's own Session Name beats the uploaded file name.
        session_name = meta.get('session_name') or name
        session_id = self.db.create_session('comprion')
        if session_name:
            self.db.rename_session(session_id, session_name)

        rows = [(elapsed, msg_type, data, 0)
                for elapsed, msg_type, data in result['messages']]
        self.db.insert_messages(session_id, rows)
        started = result.get('started')
        if started:
            # Wall-clock window from the Session Name; with the Application
            # Layer time base the span is exact, otherwise the delay estimate.
            self.db.set_session_times_from_ts(
                session_id, started, started + result['duration'])
        else:
            self.db.close_session(session_id, result['duration'])

        for note in result['warnings']:
            _log(f'COMPRION import warning: {note}')
        _log(f'Session imported: session={session_id} '
             f'name={session_name or "(untitled)"} format=comprion '
             f'time_base={result["time_base"]} '
             f'exchanges={result["exchange_count"]} apdus={result["apdu_count"]} '
             f'messages={result["message_count"]} '
             f'skipped={len(result["skipped"])}')
        self._send_json({
            'session_id': session_id,
            'name': session_name,
            'message_count': result['message_count'],
            'format': 'comprion',
            'time_base': result['time_base'],
            'apdu_count': result['apdu_count'],
            'skipped': result['skipped'],
            'warnings': result['warnings'],
        })

    def _handle_session_files(self, session_id):
        """Per-file tracking for a session (counters, changes, timeline)."""
        session = self.db.get_session(session_id)
        if session is None:
            self._send_error(404, 'Session not found')
            return
        data = self.db.track_files(session_id)
        _log(f'Session files: session={session_id} '
             f'files={len(data["files"])} operations={data["timeline_total"]}')
        self._send_json(data)

    def _handle_session_files_image(self, session_id, params):
        """Before/after images of one change (the Files view's Details)."""
        session = self.db.get_session(session_id)
        if session is None:
            self._send_error(404, 'Session not found')
            return
        key = (params.get('key', [''])[0]).strip()
        try:
            idx = int(params.get('change', ['0'])[0])
        except ValueError:
            self._send_error(400, 'change must be an index')
            return
        if not key or idx < 0:
            self._send_error(400, 'key and change are required')
            return
        data = self.db.change_images(session_id, key, idx)
        if data is None:
            self._send_error(404, 'No such file or change')
            return
        _log(f'Session file change: session={session_id} key={key} '
             f'change={idx} op={data["op"]} index={data["index"]}')
        self._send_json(data)

    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()
        _log(f'Session deleted: session={session_id}')
        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)