diff --git a/frontend/index.html b/frontend/index.html index a9ea738..8580231 100644 --- a/frontend/index.html +++ b/frontend/index.html @@ -171,6 +171,8 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(- .settings-view .conn-status.connected { background: var(--accent-bg); color: var(--green); } .settings-view .conn-status.disconnected { background: var(--warn-bg); color: var(--error); } .conn-hint { font-size: 12px; color: var(--text-muted); margin: 8px 0 0; line-height: 1.5; } +.capture-mode { font-size: 12px; color: var(--text-muted); margin: 8px 0 0; line-height: 1.5; } +.capture-mode:empty { display: none; } .version-warn { font-size: 12px; color: var(--warn); margin: 8px 0 0; line-height: 1.5; } .version-warn:empty { display: none; } .conn-instructions { margin-top: 16px; font-size: 13px; color: var(--text-faint); line-height: 1.7; } @@ -272,6 +274,7 @@ body { font: 14px/1.5 system-ui, sans-serif; background: var(--bg); color: var(-
⬤ Not connected — ensure the server is running on the URL above
+
@@ -313,6 +316,7 @@ const STATE = { typeCounts: [], pendingDeleteId: null, serverAvailable: false, + captureMode: '', }; const API_BASE = () => STATE.serverUrl; @@ -760,9 +764,14 @@ function renderMessages() { populateFilterOptions(); if (!STATE.captureActive && !STATE.sessionViewId) { - list.innerHTML = `

${t('No capture active')}

-
`; - $('#trace-start-btn', list).addEventListener('click', startCapture); + if (STATE.captureMode === 'disabled') { + list.innerHTML = `

${t('Capture unavailable')}

+

${t('Open a saved session from the Sessions tab.')}

`; + } else { + list.innerHTML = `

${t('No capture active')}

+
`; + $('#trace-start-btn', list).addEventListener('click', startCapture); + } updateStatus(); return; } @@ -837,6 +846,7 @@ function stopPolling() { // ──────────────────── Capture ──────────────────── async function startCapture() { + if (STATE.captureMode === 'disabled') return; try { const data = await API.startCapture(); STATE.captureActive = true; @@ -864,7 +874,14 @@ async function stopCapture() { function updateCaptureButton() { const btn = el('capture-btn'); + const newCapture = el('new-capture-btn'); btn.classList.remove('stop', 'connect'); + const captureOff = STATE.captureMode === 'disabled' && STATE.serverAvailable; + newCapture.style.display = captureOff ? 'none' : ''; + if (captureOff) { + btn.style.display = 'none'; + return; + } if (STATE.sessionViewId && !STATE.captureActive) { btn.style.display = 'none'; return; @@ -1062,6 +1079,23 @@ function updateVersionWarn(serverVersion) { elv.textContent = ''; } +function updateCaptureModeLine() { + const elm = el('capture-mode'); + switch (STATE.captureMode) { + case 'gsmtap': + elm.textContent = t('Capture mode: GSMTAP — the server is listening for incoming UDP/GSMTAP traffic. Ensure a GSMTAP source (e.g. simtrace2-pysniff or simtrace2-sniff) is feeding the server.'); + break; + case 'direct': + elm.textContent = t('Capture mode: direct — the server captures from SIMtrace2 hardware. Keep the device connected.'); + break; + case 'disabled': + elm.textContent = t('Capture is disabled — only pre-captured traces (saved sessions / PCAP import) are available.'); + break; + default: + elm.textContent = ''; + } +} + function connectServer() { if (_connTimer) return; _connTimer = setInterval(checkConnection, 5000); @@ -1076,6 +1110,7 @@ function disconnectServer() { STATE.serverAvailable = false; el('conn-status').textContent = '⬤ ' + t('Disconnected — click Connect to resume'); el('conn-status').className = 'conn-status disconnected'; + el('capture-mode').textContent = ''; el('version-warn').textContent = ''; el('conn-hint').textContent = ''; el('conn-instructions').innerHTML = ''; @@ -1096,9 +1131,11 @@ async function checkConnection() { try { const data = await API.status(); STATE.serverAvailable = true; - el('conn-status').textContent = '⬤ ' + t('Connected') + ' — ' + data.server; + STATE.captureMode = data.capture_mode || ''; + el('conn-status').textContent = '⬤ ' + t('Connected') + ' — ' + data.server + ' v' + (data.version || '?'); el('conn-status').className = 'conn-status connected'; updateVersionWarn(data.version); + updateCaptureModeLine(); el('conn-hint').textContent = ''; el('conn-instructions').innerHTML = ''; el('connect-btn').textContent = t('Disconnect'); @@ -1112,8 +1149,10 @@ async function checkConnection() { updateCaptureButton(); } catch (e) { STATE.serverAvailable = false; + STATE.captureMode = ''; el('conn-status').textContent = '⬤ ' + t('Not connected — ensure the server is running on the URL above'); el('conn-status').className = 'conn-status disconnected'; + el('capture-mode').textContent = ''; el('version-warn').textContent = ''; const url = STATE.serverUrl || location.origin; el('conn-hint').textContent = t(isLikelyBlocked() ? HINT_BLOCKED : HINT_UNREACHABLE).replace('{url}', url); @@ -1167,7 +1206,7 @@ function onFilterChange() { } // ──────────────────── Versioning ──────────────────── -const PWA_VERSION = '1.1.0'; +const PWA_VERSION = '1.2.0'; const MIN_SERVER_VERSION = '1.0.0'; function parseVersion(v) { @@ -1261,6 +1300,12 @@ const LANG_RU = { 'Server version unknown — update simtrace2-pysniff-server.': 'Версия сервера неизвестна — обновите simtrace2-pysniff-server.', 'Server version {v} is outdated. Please update simtrace2-pysniff-server.': 'Версия сервера {v} устарела. Обновите simtrace2-pysniff-server.', 'Server version {v} is newer than expected. Update simtrace2-pysniff for full compatibility.': 'Версия сервера {v} новее ожидаемой. Обновите simtrace2-pysniff для полной совместимости.', + 'Capture mode: GSMTAP — the server is listening for incoming UDP/GSMTAP traffic. Ensure a GSMTAP source (e.g. simtrace2-pysniff or simtrace2-sniff) is feeding the server.': 'Режим захвата: GSMTAP — сервер ожидает входящий UDP/GSMTAP-трафик. Убедитесь, что источник GSMTAP (например, simtrace2-pysniff или simtrace2-sniff) передаёт данные серверу.', + 'Capture mode: direct — the server captures from SIMtrace2 hardware. Keep the device connected.': 'Режим захвата: direct — сервер захватывает данные с устройства SIMtrace2. Не отключайте устройство.', + 'Capture is disabled — only pre-captured traces (saved sessions / PCAP import) are available.': 'Захват отключён — доступны только ранее записанные трассировки (сохранённые сессии / импорт PCAP).', + 'Capture unavailable': 'Захват недоступен', + 'Open a saved session from the Sessions tab.': 'Откройте сохранённую сессию на вкладке «Сессии».', + 'Capture disabled': 'Захват отключён', }; function t(text) { diff --git a/frontend/manifest.json b/frontend/manifest.json index fbc3875..66c1fe5 100644 --- a/frontend/manifest.json +++ b/frontend/manifest.json @@ -2,7 +2,7 @@ "name": "SIMtrace Analyser", "short_name": "SIMtrace", "description": "PWA frontend for browsing SIMtrace2 APDU traces", - "version": "1.1.0", + "version": "1.2.0", "start_url": "index.html", "display": "standalone", "background_color": "#0f172a", diff --git a/pyproject.toml b/pyproject.toml index 38e041f..5f1d0b7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.backends._legacy:_Backend" [project] name = "simtrace2-pysniff" -version = "1.1.0" +version = "1.2.0" description = "Python-based sniffing tool for SIMtrace2 hardware (SIM card communication sniffer)" readme = {file = "README.md", content-type = "text/markdown"} license = {text = "GPL-2.0-or-later"} diff --git a/simtrace2_pysniff/__init__.py b/simtrace2_pysniff/__init__.py index 82cb9e9..65f3412 100644 --- a/simtrace2_pysniff/__init__.py +++ b/simtrace2_pysniff/__init__.py @@ -6,7 +6,7 @@ from .gsmtap import GsmtapSender, GsmtapReceiver from .pcap import PcapWriter from .dump import FileDumper, format_message -__version__ = '1.1.0' +__version__ = '1.2.0' __all__ = [ 'SniffSession', 'DeviceDisconnected', diff --git a/simtrace2_pysniff/server/__main__.py b/simtrace2_pysniff/server/__main__.py index e52ced0..5cfa6f0 100644 --- a/simtrace2_pysniff/server/__main__.py +++ b/simtrace2_pysniff/server/__main__.py @@ -29,8 +29,8 @@ def main(): help='HTTP server port (default: 8081)') p.add_argument('--db', default=DEFAULT_DB_PATH, help=f'SQLite database path (default: {DEFAULT_DB_PATH})') - p.add_argument('--capture', choices=['gsmtap', 'direct'], default='gsmtap', - help='Capture mode: gsmtap (listen UDP 4729) or direct (SIMtrace2 USB)') + p.add_argument('--capture', choices=['gsmtap', 'direct', 'disabled'], default='gsmtap', + help='Capture mode: gsmtap (listen UDP 4729), direct (SIMtrace2 USB), or disabled (no capture)') p.add_argument('--gsmtap-port', type=int, default=4729, help='UDP port for GSMTAP listener (default: 4729)') p.add_argument('--web-dir', default=_default_web_dir(), metavar='PATH', @@ -39,15 +39,17 @@ def main(): db = Database(args.db) + capture = None if args.capture == 'gsmtap': backend = GsmtapListener(bind_port=args.gsmtap_port) - else: + capture = CaptureManager(backend, db) + elif args.capture == 'direct': backend = DirectSniffer() - - capture = CaptureManager(backend, db) + capture = CaptureManager(backend, db) RequestHandler.db = db RequestHandler.capture = capture + RequestHandler.capture_mode = args.capture server = HTTPServer(('127.0.0.1', args.port), RequestHandler) server.web_dir = args.web_dir @@ -65,7 +67,7 @@ def main(): except KeyboardInterrupt: pass finally: - if capture.active: + if capture and capture.active: capture.stop_session() server.server_close() diff --git a/simtrace2_pysniff/server/server.py b/simtrace2_pysniff/server/server.py index bfddafa..496a30b 100644 --- a/simtrace2_pysniff/server/server.py +++ b/simtrace2_pysniff/server/server.py @@ -42,6 +42,7 @@ def _content_disposition(filename): 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) @@ -197,13 +198,15 @@ class RequestHandler(BaseHTTPRequestHandler): def _handle_status(self): active_session = self.db.get_active_session() + capture = self.capture self._send_json({ 'server': 'simtrace-analyser-server', 'version': __version__, - 'capture_active': self.capture.active, - 'session_id': self.capture.session_id, + '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(self.capture.session_id) if self.capture.session_id else 0, + 'messages_count': self.db.count_messages(capture.session_id) if capture and capture.session_id else 0, }) def _handle_list_sessions(self): @@ -259,7 +262,7 @@ class RequestHandler(BaseHTTPRequestHandler): def _handle_capture_latest(self, params): after_id = int(params.get('after', [0])[0]) - if not self.capture.active: + if not self.capture or not self.capture.active: self._send_json({'messages': [], 'next_after': after_id, 'active': False}) return msg_id = self.capture.latest_msg_id @@ -273,11 +276,14 @@ class RequestHandler(BaseHTTPRequestHandler): def _handle_capture_status(self): self._send_json({ - 'active': self.capture.active, - 'session_id': self.capture.session_id, + 'active': self.capture.active if self.capture else False, + 'session_id': self.capture.session_id 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() @@ -285,6 +291,9 @@ class RequestHandler(BaseHTTPRequestHandler): 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 @@ -336,7 +345,7 @@ class RequestHandler(BaseHTTPRequestHandler): self._send_json({'ok': True}) def _handle_delete_session(self, session_id): - if self.capture.session_id == 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() diff --git a/tests/test_pcap_import.py b/tests/test_pcap_import.py index dc3e09f..6546e65 100644 --- a/tests/test_pcap_import.py +++ b/tests/test_pcap_import.py @@ -140,7 +140,7 @@ class TestImportEndpoint(unittest.TestCase): class TestVersion(unittest.TestCase): def test_package_version(self): from simtrace2_pysniff import __version__ - self.assertEqual(__version__, '1.1.0') + self.assertEqual(__version__, '1.2.0') def test_status_includes_version(self): import os @@ -171,5 +171,57 @@ class TestVersion(unittest.TestCase): self.assertEqual(h._json['version'], __version__) +class TestCaptureDisabled(unittest.TestCase): + def _handler(self, capture_mode): + import os + from simtrace2_pysniff.server.server import RequestHandler + from simtrace2_pysniff.server.database import Database + + class Fake(RequestHandler): + def __init__(self, db, capture_mode): + self.db = db + self.capture = None + self.capture_mode = capture_mode + self._status = None + self._json = None + + 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(db, capture_mode) + + def test_status_reports_disabled(self): + h = self._handler('disabled') + h._handle_status() + self.assertEqual(h._json['capture_mode'], 'disabled') + self.assertFalse(h._json['capture_active']) + self.assertIsNone(h._json['session_id']) + + def test_capture_start_403(self): + h = self._handler('disabled') + h._handle_capture_start() + self.assertEqual(h._status, 403) + self.assertIn('Capture disabled', h._json['error']) + + def test_capture_stop_403(self): + h = self._handler('disabled') + h._handle_capture_stop() + self.assertEqual(h._status, 403) + + def test_status_reports_gsmtap(self): + h = self._handler('gsmtap') + h._handle_status() + self.assertEqual(h._json['capture_mode'], 'gsmtap') + + if __name__ == '__main__': unittest.main()