From fe834d4fe8c2ce926b42b98659e06ede5f76f884 Mon Sep 17 00:00:00 2001 From: CyberMind Date: Fri, 22 May 2026 17:49:13 +0200 Subject: [PATCH] =?UTF-8?q?feat(sentinelle-gsm):=20v0.3.5=20+=20v0.3.6=20?= =?UTF-8?q?=E2=80=94=20scan-auto=20multi-cell=20+=20async-job=20pattern=20?= =?UTF-8?q?(closes=20#356,=20#357)=20(#358)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(sentinelle-gsm): v0.3.5 — scan-auto multi-cell orchestration (closes #356) Port the gkerma/IMSI-catcher scan-and-livemon pattern: grgsm_scanner discovers cells, select_diverse picks one-per-operator-then-by-power, LivemonFleet spawns N livemons on serverport=4730+i. All emit GSMTAP to the listener on 4729; the existing demux is by ARFCN in the GSMTAP header, so multi-cell capture works without any listener changes. New library code: lib/sentinelle_gsm/scanner.py - CellInfo dataclass - parse_scanner_output() — robust to banners/progress/partial lines - select_diverse() — operator-first, power-second; backs off to remaining strongest when fewer operators than max_cells - scan_band() — async subprocess wrapper around grgsm_scanner; always passes --args=rtl=0 (dodges the device.py auto-pick bug that v0.3.4's shim fixes for livemon; scanner has a separate code path that accepts the same arg fine) lib/sentinelle_gsm/livemon_fleet.py - LivemonFleet — N parallel LivemonRunners - Partial-start rollback (no leaked grgsm children if cell #2 of 3 fails) - RunnerSummary keeps cell context (MCC/MNC/ARFCN) alongside the ScanStatus so the API can echo "which operator each runner is watching" New endpoints: POST /scan/auto body: {band, max_cells, gain, ppm, samp_rate, speed, scan_timeout} Refuses if /scan/start single-runner OR a fleet is already alive. Lifecycle: stop listener (scanner hard-binds 4729) → scan_band → restart listener → fleet.start. Returns {mode, cells_found, cells_selected, runners}. GET /scan/auto/status Per-runner status of the active fleet, or empty if none. POST /scan/stop (extended) Now stops the fleet in addition to the single runner. Payload is polymorphic on which path was active. Tests: 96/96 sweep (up from 71 in v0.3.4). - test_scanner.py: 12 tests covering parser edge cases (banner noise, partial lines, positive power, empty output) and the diversity-selection algorithm - test_livemon_fleet.py: 7 tests covering happy path, partial-start rollback, custom base_serverport, double-start refusal - test_scan_api.py: 6 new endpoint tests (happy path, 409 vs single-runner, 409 vs second fleet, 400 on zero max_cells, 504 on scanner timeout, /scan/auto/status idle) .deb builds clean as secubox-sentinelle-gsm_0.3.5-1~bookworm1_all.deb. * feat(sentinelle-gsm): v0.3.6 — async-job pattern for /scan/auto (closes #357) v0.3.5 ran the scanner synchronously in the HTTP handler. On the MOCHAbin's Cortex-A72 a GSM900 sweep takes 8-15 min — longer than any sensible HTTP timeout. v0.3.6 makes /scan/auto fire-and-poll: POST returns {id, state=pending} immediately and spawns the driver coroutine in the background; clients poll GET /scan/auto/jobs/{id} for state transitions. New library code: lib/sentinelle_gsm/scan_auto_job.py - ScanAutoJob dataclass (id, state, params, cells_found/selected, runners, error, fleet ref, task handle) - JobRegistry — at most one active job, FIFO history capped at max_history=10 terminal jobs; newest-first listing. New endpoints: POST /scan/auto submit + return immediately GET /scan/auto/jobs/{id} poll one job GET /scan/auto/jobs?limit=N recent jobs, newest first POST /scan/auto/jobs/{id}/cancel Task.cancel + driver unwind State machine: pending → scanning → starting_fleet → running → stopped ↘ failed ↘ failed ↘ cancelled Driver coroutine cleanup on every exit path: listener restarted (or left stopped if the scan failed and there's no fleet to feed it), single runner ensured idle, fleet stopped on cancel/failure. /scan/start also 409s now if a scan-auto job is in flight even before it has spawned the fleet (state=scanning has no _fleet yet, but starting a single-runner would stomp on the scanner's 4729 bind). /scan/stop marks the active job as `stopped` when it tears the fleet down. Tests: 114/114 sweep (up from 96 in v0.3.5). - test_scan_auto_job.py (NEW, 12 tests): registry FIFO eviction + driver state machine in isolation (Starlette TestClient cancels spawned tasks at request-scope exit, so cross-request task tests live here, not in test_scan_api.py). - test_scan_api.py refactored: in-flight tests seed the registry directly rather than rely on a real spawned task surviving across client.post() calls. .deb builds clean as secubox-sentinelle-gsm_0.3.6-1~bookworm1_all.deb. --------- Co-authored-by: CyberMind-FR --- packages/secubox-sentinelle-gsm/api/main.py | 321 +++++++++++++++++- .../api/tests/test_livemon_fleet.py | 163 +++++++++ .../api/tests/test_scan_api.py | 257 +++++++++++++- .../api/tests/test_scan_auto_job.py | 265 +++++++++++++++ .../api/tests/test_scanner.py | 181 ++++++++++ .../secubox-sentinelle-gsm/debian/changelog | 74 ++++ .../lib/sentinelle_gsm/livemon_fleet.py | 136 ++++++++ .../lib/sentinelle_gsm/scan_auto_job.py | 163 +++++++++ .../lib/sentinelle_gsm/scanner.py | 187 ++++++++++ 9 files changed, 1743 insertions(+), 4 deletions(-) create mode 100644 packages/secubox-sentinelle-gsm/api/tests/test_livemon_fleet.py create mode 100644 packages/secubox-sentinelle-gsm/api/tests/test_scan_auto_job.py create mode 100644 packages/secubox-sentinelle-gsm/api/tests/test_scanner.py create mode 100644 packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/livemon_fleet.py create mode 100644 packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scan_auto_job.py create mode 100644 packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scanner.py diff --git a/packages/secubox-sentinelle-gsm/api/main.py b/packages/secubox-sentinelle-gsm/api/main.py index fbb3ce0c..42657220 100644 --- a/packages/secubox-sentinelle-gsm/api/main.py +++ b/packages/secubox-sentinelle-gsm/api/main.py @@ -46,7 +46,10 @@ from sentinelle_gsm.gsmtap_listener import GsmtapListener # noqa: E402 from sentinelle_gsm.l3_decode import ( # noqa: E402 L3Decode, MID_TYPE_IMSI, MID_TYPE_TMSI, ) +from sentinelle_gsm.livemon_fleet import LivemonFleet, RunnerSummary # noqa: E402 from sentinelle_gsm.livemon_runner import LivemonRunner, detect_rtlsdr_usb # noqa: E402 +from sentinelle_gsm.scan_auto_job import JobRegistry, ScanAutoJob # noqa: E402 +from sentinelle_gsm.scanner import CellInfo, scan_band, select_diverse # noqa: E402 from sentinelle_gsm.observations import ( # noqa: E402 ObservationsDB, PagingEvent, Sighting, ) @@ -133,6 +136,21 @@ _livemon: Optional[LivemonRunner] = None _listener: Optional[GsmtapListener] = None _obs_db: Optional[ObservationsDB] = None _consume_task: Optional[asyncio.Task] = None +# v0.3.5: multi-cell fleet, populated only by /scan/auto. Coexists with +# the single-runner path on /scan/start — at most one of them is alive +# at any moment (both 409 if the other is active). +# v0.3.6: now populated by the async-job driver coroutine when it +# transitions to `starting_fleet`. Cleared when /scan/stop or the job's +# cancel path tears the fleet down. Same lifetime as before, just +# managed from the background task instead of the request thread. +_fleet: Optional[LivemonFleet] = None +# v0.3.6: in-memory async-job registry for /scan/auto. The HTTP +# handler posts a ScanAutoJob and returns immediately; a background +# coroutine drives state transitions. Polling endpoints read from this +# registry. Not durable across service restarts on purpose — the +# observations DB owns historical sightings; this is just operational +# state for "what's happening right now". +_job_registry: JobRegistry = JobRegistry(max_history=10) OBSERVATIONS_DB_PATH = Path("/var/lib/secubox/sentinelle-gsm/observations.db") @@ -687,6 +705,13 @@ async def scan_start(body: ScanStartBody) -> dict: global _consume_task if _consume_task is not None and not _consume_task.done(): raise HTTPException(409, "scan already running") + if _fleet is not None: + raise HTTPException(409, "fleet scan already running — stop first") + # v0.3.6: also 409 if a scan-auto job is in flight even before it + # has spawned the fleet (state=scanning has no _fleet yet, but + # starting a single-runner would stomp on the scanner's 4729 bind). + if _job_registry.active() is not None: + raise HTTPException(409, "scan-auto job in flight — stop first") listener = get_listener() runner = get_livemon() try: @@ -715,8 +740,14 @@ async def scan_start(body: ScanStartBody) -> dict: @app.post("/scan/stop", dependencies=[Depends(require_jwt)]) async def scan_stop() -> dict: - """Cancel consume task, then stop runner + listener (in that order).""" - global _consume_task + """Cancel consume task, then stop runner(s) + listener. + + v0.3.5: stops whichever path is active — the /scan/start single + runner OR the /scan/auto fleet. Idempotent for both. + v0.3.6: also marks the active scan-auto job as `stopped` so its + payload finishes the transition out of `running`. + """ + global _consume_task, _fleet if _consume_task is not None: _consume_task.cancel() try: @@ -728,11 +759,297 @@ async def scan_stop() -> dict: _consume_task = None runner = get_livemon() listener = get_listener() + fleet_finals: list[RunnerSummary] = [] + if _fleet is not None: + fleet_finals = await _fleet.stop_all() + _fleet = None + # v0.3.6: if a job was running, mark it stopped. We don't cancel + # the task — it's already past the await-point that would block + # (running state holds no async work; the next state transition + # happens here, from outside the task). + active = _job_registry.active() + if active is not None and active.state == "running": + active.fleet = None + active.set_state("stopped") status = await runner.stop() await listener.stop() + if fleet_finals: + # Fleet was the active path; surface every runner's final state. + return { + "mode": "fleet", + "runners": [_runner_summary_payload(s) for s in fleet_finals], + } return _scan_status_payload(status) +class ScanAutoBody(BaseModel): + # v0.3.5: discover cells with grgsm_scanner, then run N livemons in + # parallel. Defaults match the IMSI-catcher scan-and-livemon + # project: GSM900 (the band most common for IMSI catchers in EU + # rural/suburban areas) and 3 cells (one per French carrier). + band: str = "GSM900" + max_cells: int = 3 + gain: Optional[float] = None + ppm: Optional[float] = None + samp_rate: Optional[str] = None + speed: Optional[int] = None + scan_timeout: float = 180.0 + + +def _cell_payload(c: CellInfo) -> dict: + return { + "arfcn": c.arfcn, + "freq": c.freq, + "cid": c.cid, + "lac": c.lac, + "mcc": c.mcc, + "mnc": c.mnc, + "power": c.power, + } + + +def _runner_summary_payload(s: RunnerSummary) -> dict: + return { + "cell": _cell_payload(s.cell), + "serverport": s.serverport, + "status": _scan_status_payload(s.status), + } + + +def _job_payload(j: ScanAutoJob) -> dict: + """Render a ScanAutoJob as the JSON the polling endpoints emit. + Keeps internal fields (the asyncio.Task handle, the LivemonFleet + reference) out — those aren't serialisable and would leak + implementation details.""" + return { + "id": j.id, + "state": j.state, + "params": j.params, + "started_at": j.started_at, + "finished_at": j.finished_at, + "cells_found": j.cells_found, + "cells_selected": [_cell_payload(c) for c in j.cells_selected], + "runners": [_runner_summary_payload(s) for s in j.runners], + "error": j.error, + } + + +async def _drive_scan_auto_job(job: ScanAutoJob) -> None: + """Background coroutine that walks a ScanAutoJob through its state + machine. + + Pre-conditions (enforced by the /scan/auto handler before spawning + this task): + - No other active job in the registry + - /scan/start single runner is idle + - max_cells >= 1 + + State transitions: + pending → scanning → starting_fleet → running (happy path) + ↘ failed + ↘ failed + ↘ cancelled + + Cleanup on every exit path: listener restarted (or left stopped + if the scan failed and there's no fleet to feed it), single + runner ensured idle, fleet stopped on cancel/failure. + """ + global _consume_task, _fleet + listener = get_listener() + runner = get_livemon() + + # Snapshot params for the scanner / fleet calls. The job stores + # them as a dict so the API can echo them on poll. + p = job.params + + try: + # ── scanning ──────────────────────────────────────────────── + job.set_state("scanning") + # Scanner hard-binds 4729; listener has to be down. Idempotent + # for unbound state. + await listener.stop() + + cells = await scan_band( + band=p["band"], + gain=p.get("gain"), + ppm=p.get("ppm"), + samp_rate=p.get("samp_rate"), + speed=p.get("speed"), + timeout=p.get("scan_timeout", 600.0), + ) + job.cells_found = len(cells) + + if not cells: + # No cells found — restart listener so other endpoints + # behave normally, and finish clean. + try: + await listener.start() + except OSError: + pass + job.set_state("stopped") + return + + chosen = select_diverse(cells, p["max_cells"]) + job.cells_selected = chosen + + # ── starting_fleet ────────────────────────────────────────── + job.set_state("starting_fleet") + try: + await listener.start() + except OSError as e: + raise RuntimeError(f"listener bind failed after scan: {e}") + + # Defensive: single-runner should be idle (we 409'd in the + # handler), but call stop() just in case. + await runner.stop() + + fleet = LivemonFleet( + gsmtap_port=getattr(listener, "port", 4729), + ) + try: + summaries = await fleet.start( + chosen, + gain=p.get("gain"), + ppm=p.get("ppm"), + samp_rate=p.get("samp_rate"), + ) + except Exception as e: + # Don't leave the listener up if the fleet refused to + # start — nothing's emitting GSMTAP and we'd just leak + # the socket. + await listener.stop() + raise RuntimeError(f"fleet.start failed: {e}") + + # ── running ───────────────────────────────────────────────── + job.runners = summaries + job.fleet = fleet + _fleet = fleet + _consume_task = asyncio.create_task(_consume_observations()) + job.set_state("running") + # Stay in `running` until /scan/stop or cancel() — which both + # happen from OUTSIDE this coroutine. The task can be awaited + # but won't progress further on its own. + + except asyncio.CancelledError: + # Cancellation can hit at any point. Best-effort cleanup of + # whatever state we managed to put in place, then re-raise so + # the task is properly marked cancelled. + try: + if job.fleet is not None: + await job.fleet.stop_all() + job.fleet = None + await listener.stop() + except Exception: # noqa: BLE001 + _log.exception("cleanup after scan-auto cancel failed") + if _consume_task is not None: + _consume_task.cancel() + _consume_task = None + _fleet = None + job.set_state("cancelled") + raise + + except Exception as e: # noqa: BLE001 + _log.exception("scan-auto job %s failed", job.id) + job.error = str(e) + # Try to leave the system in a clean state — listener down, + # no orphan fleet, no consume task. + try: + if job.fleet is not None: + await job.fleet.stop_all() + job.fleet = None + except Exception: + pass + try: + await listener.stop() + except Exception: + pass + if _consume_task is not None: + _consume_task.cancel() + _consume_task = None + _fleet = None + job.set_state("failed") + + +@app.post("/scan/auto", dependencies=[Depends(require_jwt)]) +async def scan_auto(body: ScanAutoBody) -> dict: + """Submit a new /scan/auto job and return immediately. + + v0.3.6 makes /scan/auto fire-and-poll: the handler validates, + spawns the background driver, and returns `{job_id, state}`. + Callers GET /scan/auto/jobs/{id} to watch state transitions + (which can take 8-15 min on MOCHAbin's CPU for a full GSM900 + sweep) without holding an HTTP connection open the whole time. + """ + global _consume_task + if _consume_task is not None and not _consume_task.done(): + raise HTTPException(409, "scan already running") + if _job_registry.active() is not None: + raise HTTPException(409, "scan-auto job already running") + if body.max_cells <= 0: + raise HTTPException(400, "max_cells must be ≥ 1") + + job = _job_registry.submit(body.model_dump()) + job.task = asyncio.create_task(_drive_scan_auto_job(job)) + return {"id": job.id, "state": job.state} + + +@app.get("/scan/auto/jobs/{job_id}", dependencies=[Depends(require_jwt)]) +async def scan_auto_job_get(job_id: str) -> dict: + """Poll one job's state. 404 if unknown / aged out of history.""" + job = _job_registry.get(job_id) + if job is None: + raise HTTPException(404, f"no job {job_id}") + return _job_payload(job) + + +@app.get("/scan/auto/jobs", dependencies=[Depends(require_jwt)]) +async def scan_auto_jobs_list(limit: int = 10) -> dict: + """Recent jobs, newest first. Capped by JobRegistry.max_history.""" + if limit < 1: + limit = 1 + if limit > 50: + limit = 50 + return {"jobs": [_job_payload(j) for j in _job_registry.list(limit)]} + + +@app.post( + "/scan/auto/jobs/{job_id}/cancel", + dependencies=[Depends(require_jwt)], +) +async def scan_auto_job_cancel(job_id: str) -> dict: + """Cancel an in-flight job. Idempotent on already-terminal jobs. + + Implementation: call Task.cancel() and let the driver's + asyncio.CancelledError handler unwind listener + fleet state. + """ + job = _job_registry.get(job_id) + if job is None: + raise HTTPException(404, f"no job {job_id}") + if job.is_terminal(): + return _job_payload(job) + if job.task is not None: + job.task.cancel() + try: + await job.task + except asyncio.CancelledError: + pass + return _job_payload(job) + + +@app.get("/scan/auto/status", dependencies=[Depends(require_jwt)]) +async def scan_auto_status() -> dict: + """Per-runner status of the currently-running fleet, or empty if + no fleet. Mirrors the active job's `runners` field when state is + `running`.""" + if _fleet is None: + return {"mode": "fleet", "running": False, "runners": []} + return { + "mode": "fleet", + "running": _fleet.is_running(), + "runners": [_runner_summary_payload(s) for s in _fleet.summaries()], + } + + @app.get("/scan/status", dependencies=[Depends(require_jwt)]) async def scan_status() -> dict: """Current runner state (running / pid / freq / stderr-tail).""" diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_livemon_fleet.py b/packages/secubox-sentinelle-gsm/api/tests/test_livemon_fleet.py new file mode 100644 index 00000000..702463c1 --- /dev/null +++ b/packages/secubox-sentinelle-gsm/api/tests/test_livemon_fleet.py @@ -0,0 +1,163 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +"""LivemonFleet — N parallel runners, each on its own --serverport.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from sentinelle_gsm.livemon_fleet import LivemonFleet +from sentinelle_gsm.scanner import CellInfo + + +def _cell(arfcn: int, freq: str, mnc: int) -> CellInfo: + return CellInfo(arfcn=arfcn, freq=freq, cid=arfcn, lac=1, + mcc=208, mnc=mnc, power=-50) + + +def _make_fake_proc(wait_returns=None): + """Same helper pattern as test_livemon_runner.""" + fake = MagicMock(spec=asyncio.subprocess.Process) + fake.pid = 90000 + fake.returncode = None + fake.stderr = AsyncMock() + fake.stderr.read = AsyncMock(return_value=b"") + fake.terminate = MagicMock() + if wait_returns is None: + fake.wait = AsyncMock(side_effect=lambda: asyncio.Future()) + else: + async def _wait(): + return wait_returns + fake.wait = AsyncMock(side_effect=_wait) + return fake + + +@pytest.mark.asyncio +async def test_fleet_starts_one_runner_per_cell(): + """The IMSI-catcher pattern: N livemons, each on serverport=base+i, + each watching a different cell's freq.""" + cells = [ + _cell(73, "939.6M", 1), + _cell(119, "947.4M", 10), + _cell(1, "935.2M", 20), + ] + fleet = LivemonFleet() + procs = [_make_fake_proc() for _ in cells] + with patch("asyncio.create_subprocess_exec", AsyncMock(side_effect=procs)) as mck: + summaries = await fleet.start(cells) + + assert len(summaries) == 3 + assert mck.call_count == 3 + + # Each call's argv has the right freq and a unique --serverport + seen_serverports = [] + seen_freqs = [] + for call in mck.call_args_list: + argv = call[0] + seen_freqs.append(argv[argv.index("-f") + 1]) + for a in argv: + if a.startswith("--serverport="): + seen_serverports.append(int(a.split("=")[1])) + + assert seen_freqs == ["939.6M", "947.4M", "935.2M"] + assert seen_serverports == [4730, 4731, 4732] + + +@pytest.mark.asyncio +async def test_fleet_refuses_double_start(): + cells = [_cell(73, "939.6M", 1)] + fleet = LivemonFleet() + with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_make_fake_proc())): + await fleet.start(cells) + with pytest.raises(RuntimeError, match="already running"): + await fleet.start(cells) + + +@pytest.mark.asyncio +async def test_fleet_rejects_empty_cells(): + """An empty fleet is meaningless — fail fast so the API endpoint + returns 400 rather than spawning nothing and pretending success.""" + fleet = LivemonFleet() + with pytest.raises(ValueError, match="no cells"): + await fleet.start([]) + + +@pytest.mark.asyncio +async def test_fleet_rolls_back_on_partial_failure(): + """If the second runner fails to start, the first must be stopped + before the exception propagates. Otherwise we leak grgsm children.""" + cells = [_cell(73, "939.6M", 1), _cell(119, "947.4M", 10)] + fleet = LivemonFleet() + + good = _make_fake_proc() + # Configure good.wait so stop() can resolve. + resolve = asyncio.Event() + + async def waiting(): + await resolve.wait() + return 0 + + good.wait = AsyncMock(side_effect=waiting) + + def on_term(): + resolve.set() + + good.terminate.side_effect = on_term + + side_effect = [good, OSError("simulated osmosdr failure")] + with patch("asyncio.create_subprocess_exec", AsyncMock(side_effect=side_effect)): + with pytest.raises(OSError, match="simulated"): + await fleet.start(cells) + + # The first runner's terminate must have been called during rollback. + good.terminate.assert_called_once() + # Fleet should be back to empty so the next start() works. + assert fleet.is_running() is False + assert fleet.summaries() == [] + + +@pytest.mark.asyncio +async def test_fleet_stop_all_drains_children_and_clears_state(): + cells = [_cell(73, "939.6M", 1), _cell(119, "947.4M", 10)] + fleet = LivemonFleet() + + procs = [] + for _ in cells: + p = _make_fake_proc() + ev = asyncio.Event() + + async def waiting(ev=ev): + await ev.wait() + return 0 + + p.wait = AsyncMock(side_effect=waiting) + + def on_term(ev=ev): + ev.set() + + p.terminate.side_effect = on_term + procs.append(p) + + with patch("asyncio.create_subprocess_exec", AsyncMock(side_effect=procs)): + await fleet.start(cells) + finals = await fleet.stop_all() + assert len(finals) == 2 + for p in procs: + p.terminate.assert_called_once() + assert fleet.is_running() is False + assert fleet.summaries() == [] + + +@pytest.mark.asyncio +async def test_fleet_custom_base_serverport(): + """Operator override — useful if 4730+ collides with something.""" + cells = [_cell(73, "939.6M", 1)] + fleet = LivemonFleet(base_serverport=5000) + with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=_make_fake_proc())) as mck: + await fleet.start(cells) + argv = mck.call_args[0] + assert "--serverport=5000" in argv diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_scan_api.py b/packages/secubox-sentinelle-gsm/api/tests/test_scan_api.py index fcca36e3..998b4c89 100644 --- a/packages/secubox-sentinelle-gsm/api/tests/test_scan_api.py +++ b/packages/secubox-sentinelle-gsm/api/tests/test_scan_api.py @@ -44,9 +44,14 @@ def client(tmp_path): api_main._listener.start = AsyncMock(return_value=None) api_main._listener.stop = AsyncMock(return_value=None) - # Reset consume-task slot so /scan/start doesn't 409 from a stale - # task left over by a sibling test. + # Reset consume-task + fleet + job registry so /scan/start and + # /scan/auto don't 409 from state a sibling test left set. api_main._consume_task = None + api_main._fleet = None + # Fresh JobRegistry per test — terminal-state jobs from siblings + # would otherwise show up in /scan/auto/jobs listings. + from sentinelle_gsm.scan_auto_job import JobRegistry as _JobRegistry + api_main._job_registry = _JobRegistry(max_history=10) api_main.app.dependency_overrides[api_main.require_jwt] = ( lambda: {"sub": "tester"} @@ -60,7 +65,14 @@ def client(tmp_path): task = api_main._consume_task if task is not None and not task.done(): task.cancel() + # Cancel any in-flight job task so it doesn't leak into the + # next test's event loop (e.g. stalling scans waiting on + # pending_evt that's about to go out of scope). + active = api_main._job_registry.active() + if active is not None and active.task is not None and not active.task.done(): + active.task.cancel() api_main._consume_task = None + api_main._fleet = None api_main._livemon = None api_main._listener = None api_main._obs_db = None @@ -82,3 +94,244 @@ def test_observations_returns_empty_by_default(client): r = client.get("/observations") assert r.status_code == 200 assert r.json()["sightings"] == [] + + +# ── v0.3.6: /scan/auto async-job endpoint ───────────────────────────── + +import asyncio as _aio +import time as _time + + +def _wait_for_job_state(client, job_id, target_states, timeout=2.0, poll=0.02): + """Poll GET /scan/auto/jobs/{id} until state ∈ target_states. + + Necessary because POST /scan/auto returns immediately while the + driver coroutine runs in the background. Returns the final job + payload; raises AssertionError on timeout.""" + if isinstance(target_states, str): + target_states = {target_states} + deadline = _time.time() + timeout + while _time.time() < deadline: + r = client.get(f"/scan/auto/jobs/{job_id}") + assert r.status_code == 200, r.text + body = r.json() + if body["state"] in target_states: + return body + _time.sleep(poll) + raise AssertionError( + f"job {job_id} did not reach {target_states} within {timeout}s " + f"(last state: {body['state']})" + ) + + +def _make_slow_scan(seconds: float = 5.0): + """Return an async scan_band stand-in that just sleeps. Avoids the + cross-event-loop pitfall of asyncio.Event() (a fresh loop per + TestClient request would break .set() across the boundary).""" + async def slow_scan(**_): + await _aio.sleep(seconds) + return [] + return slow_scan + + +def test_scan_auto_returns_job_id_immediately(client, monkeypatch): + """v0.3.6: POST returns synchronously with {id, state=pending}. + The background task does the actual work.""" + from api import main as api_main + + monkeypatch.setattr(api_main, "scan_band", _make_slow_scan(5.0)) + + r = client.post("/scan/auto", json={"band": "GSM900", "max_cells": 3}) + assert r.status_code == 200 + body = r.json() + assert "id" in body + assert body["state"] in ("pending", "scanning") + + # Job appears in GET by id. + r2 = client.get(f"/scan/auto/jobs/{body['id']}") + assert r2.status_code == 200 + assert r2.json()["id"] == body["id"] + # Cancel rather than wait the full sleep — fixture teardown + # cancels any lingering task anyway, but explicit is cleaner. + client.post(f"/scan/auto/jobs/{body['id']}/cancel") + + +def test_scan_auto_happy_path_reaches_running(client, monkeypatch): + """Three cells found, fleet started — job should land in `running` + with cells_selected and runners populated.""" + from api import main as api_main + from sentinelle_gsm.livemon_fleet import RunnerSummary + from sentinelle_gsm.scanner import CellInfo + + cells = [ + CellInfo(arfcn=73, freq="939.6M", cid=100, lac=200, mcc=208, mnc=1, power=-50), + CellInfo(arfcn=119, freq="947.4M", cid=12345, lac=234, mcc=208, mnc=10, power=-45), + CellInfo(arfcn=1, freq="935.2M", cid=50, lac=100, mcc=208, mnc=20, power=-55), + ] + + async def fake_scan_band(**_): + return cells + + monkeypatch.setattr(api_main, "scan_band", fake_scan_band) + + fake_fleet = MagicMock() + fake_fleet.start = AsyncMock(return_value=[ + RunnerSummary(cell=c, serverport=4730 + i, status=MagicMock( + running=True, pid=1000 + i, freq=c.freq, started_at=1.0, stderr_tail="")) + for i, c in enumerate(cells) + ]) + fake_fleet.is_running = MagicMock(return_value=True) + fake_fleet.stop_all = AsyncMock(return_value=[]) + monkeypatch.setattr(api_main, "LivemonFleet", lambda **_: fake_fleet) + + r = client.post("/scan/auto", json={"band": "GSM900", "max_cells": 3}) + job_id = r.json()["id"] + body = _wait_for_job_state(client, job_id, "running") + + assert body["cells_found"] == 3 + assert len(body["runners"]) == 3 + serverports = [run["serverport"] for run in body["runners"]] + assert serverports == [4730, 4731, 4732] + + +def test_scan_auto_409s_when_single_scan_active(client): + """Can't start a job while /scan/start's single runner is mid-flight.""" + from api import main as api_main + api_main._consume_task = MagicMock(done=MagicMock(return_value=False)) + try: + r = client.post("/scan/auto", json={"max_cells": 1}) + assert r.status_code == 409 + finally: + api_main._consume_task = None + + +def test_scan_auto_409s_when_another_job_active(client): + """v0.3.6: second POST while a job is in flight → 409. + + TestClient cancels tasks at request-scope exit, so we can't rely + on the spawned driver to stay alive across requests in tests. + Instead, seed the registry with a job in `scanning` state directly + — same effect on the endpoint's active() check, no race. + """ + from api import main as api_main + job = api_main._job_registry.submit({"band": "GSM900", "max_cells": 1}) + job.set_state("scanning") + try: + r = client.post("/scan/auto", json={"band": "GSM900", "max_cells": 3}) + assert r.status_code == 409 + finally: + job.set_state("cancelled") # release the active() slot + + +def test_scan_auto_400_on_zero_max_cells(client): + r = client.post("/scan/auto", json={"max_cells": 0}) + assert r.status_code == 400 + + +def test_scan_auto_empty_scan_lands_in_stopped(client, monkeypatch): + """Blank scan = no cells found = job state `stopped` (not failed — + "no GSM around here" is legitimate).""" + from api import main as api_main + + async def empty_scan(**_): + return [] + + monkeypatch.setattr(api_main, "scan_band", empty_scan) + r = client.post("/scan/auto", json={"band": "GSM900", "max_cells": 3}) + assert r.status_code == 200 + job_id = r.json()["id"] + body = _wait_for_job_state(client, job_id, "stopped") + assert body["cells_found"] == 0 + assert body["runners"] == [] + + +def test_scan_auto_scanner_timeout_lands_in_failed(client, monkeypatch): + """grgsm_scanner that times out → job state `failed` with error. + HTTP no longer returns 504 — the POST returns 200 quickly, the + failure surfaces in subsequent GETs.""" + from api import main as api_main + + async def timing_out(**_): + raise _aio.TimeoutError + + monkeypatch.setattr(api_main, "scan_band", timing_out) + r = client.post("/scan/auto", json={"band": "GSM900", "max_cells": 3}) + assert r.status_code == 200 + job_id = r.json()["id"] + body = _wait_for_job_state(client, job_id, "failed") + # asyncio.TimeoutError carries no message in Python 3.10+; we just + # need a non-None error field (whatever the str() gave us). + assert body["error"] is not None + + +def test_scan_auto_jobs_list(client, monkeypatch): + """GET /scan/auto/jobs lists newest-first up to limit.""" + from api import main as api_main + + async def empty_scan(**_): + return [] + + monkeypatch.setattr(api_main, "scan_band", empty_scan) + + ids = [] + for _ in range(3): + r = client.post("/scan/auto", json={"band": "GSM900", "max_cells": 1}) + ids.append(r.json()["id"]) + _wait_for_job_state(client, ids[-1], "stopped") + + r = client.get("/scan/auto/jobs") + assert r.status_code == 200 + jobs = r.json()["jobs"] + # Newest first. + assert [j["id"] for j in jobs[:3]] == list(reversed(ids)) + + +def test_scan_auto_cancel_in_flight(client): + """POST /scan/auto/jobs/{id}/cancel → calls task.cancel() and + returns the updated payload. Cross-request task lifecycle is + tested in test_scan_auto_job.py against the driver directly; + here we just verify the endpoint plumbing (task.cancel + 200).""" + from api import main as api_main + job = api_main._job_registry.submit({"band": "GSM900", "max_cells": 1}) + job.set_state("scanning") + + cancelled = False + + class _FakeTask: + def done(self): + return False + + def cancel(self): + nonlocal cancelled + cancelled = True + job.set_state("cancelled") + + def __await__(self): + # async behaviour: yield nothing, complete immediately. + if False: + yield + return None + + job.task = _FakeTask() + r = client.post(f"/scan/auto/jobs/{job.id}/cancel") + assert r.status_code == 200 + assert r.json()["state"] == "cancelled" + assert cancelled is True + + +def test_scan_auto_cancel_unknown_job_404(client): + r = client.post("/scan/auto/jobs/nope/cancel") + assert r.status_code == 404 + + +def test_scan_auto_get_unknown_job_404(client): + r = client.get("/scan/auto/jobs/nope") + assert r.status_code == 404 + + +def test_scan_auto_status_no_fleet(client): + r = client.get("/scan/auto/status") + assert r.status_code == 200 + body = r.json() + assert body["running"] is False + assert body["runners"] == [] diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_scan_auto_job.py b/packages/secubox-sentinelle-gsm/api/tests/test_scan_auto_job.py new file mode 100644 index 00000000..4f013d9f --- /dev/null +++ b/packages/secubox-sentinelle-gsm/api/tests/test_scan_auto_job.py @@ -0,0 +1,265 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +Direct unit tests for the v0.3.6 async-job state machine. + +test_scan_api.py covers the HTTP plumbing; this file covers the +registry and the driver coroutine in isolation, where we can drive +the event loop ourselves (no Starlette TestClient cancelling spawned +tasks at request-scope exit). +""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from sentinelle_gsm.livemon_fleet import RunnerSummary +from sentinelle_gsm.scan_auto_job import ( + JobRegistry, + ScanAutoJob, + TERMINAL_STATES, +) +from sentinelle_gsm.scanner import CellInfo + + +# ── JobRegistry ────────────────────────────────────────────────────── + +def test_registry_submit_creates_pending_job(): + r = JobRegistry() + j = r.submit({"band": "GSM900", "max_cells": 3}) + assert j.state == "pending" + assert j.id + assert r.get(j.id) is j + assert r.active() is j + + +def test_registry_active_returns_first_non_terminal(): + r = JobRegistry() + j1 = r.submit({"band": "GSM900", "max_cells": 1}) + j1.set_state("stopped") + j2 = r.submit({"band": "GSM900", "max_cells": 1}) + assert r.active() is j2 + + +def test_registry_evicts_old_terminal_on_next_submit(): + """Eviction fires on submit() — so 3 terminal jobs survive until + a 4th submit triggers the FIFO cleanup.""" + r = JobRegistry(max_history=2) + for _ in range(3): + r.submit({"band": "GSM900", "max_cells": 1}).set_state("stopped") + # Pre-eviction, all 3 still in the list (last submit ran before + # the previous two became terminal). + # New active job triggers _evict_old which sees 3 terminal + + # 1 active = keep active + max_history (=2) terminals. + active = r.submit({"band": "GSM900", "max_cells": 1}) + assert r.active() is active + assert len(r.list()) == 3 # 2 terminal + 1 active + + +def test_registry_list_newest_first(): + r = JobRegistry() + ids = [r.submit({"band": "GSM900", "max_cells": 1}).id for _ in range(3)] + listed = [j.id for j in r.list()] + assert listed == list(reversed(ids)) + + +def test_set_state_records_finished_at_on_terminal(): + j = ScanAutoJob(id="x", params={}) + j.set_state("scanning") + assert j.finished_at is None + j.set_state("stopped") + assert j.finished_at is not None + # finished_at sticks — re-setting a terminal state doesn't bump it + first = j.finished_at + j.set_state("failed") + assert j.finished_at == first + + +def test_terminal_states_set(): + """Sanity check: stopped/failed/cancelled are terminal.""" + assert TERMINAL_STATES == {"stopped", "failed", "cancelled"} + + +# ── _drive_scan_auto_job (the driver coroutine) ────────────────────── + +@pytest.fixture +def driver_env(tmp_path): + """Set up enough api.main globals for the driver to walk its + state machine without touching real grgsm binaries / sockets. + Returns the patched module so tests can inspect or mutate.""" + from api import main as api_main + from sentinelle_gsm.observations import ObservationsDB + + api_main._obs_db = ObservationsDB(tmp_path / "obs.db") + + api_main._livemon = MagicMock() + api_main._livemon.stop = AsyncMock(return_value=MagicMock( + running=False, pid=None, freq=None, started_at=None, stderr_tail="")) + api_main._livemon.status = MagicMock(return_value=MagicMock( + running=False, pid=None, freq=None, started_at=None, stderr_tail="")) + + api_main._listener = MagicMock() + api_main._listener.start = AsyncMock(return_value=None) + api_main._listener.stop = AsyncMock(return_value=None) + api_main._listener.port = 4729 + + api_main._consume_task = None + api_main._fleet = None + api_main._job_registry = JobRegistry(max_history=10) + + try: + yield api_main + finally: + if api_main._consume_task is not None: + api_main._consume_task.cancel() + api_main._consume_task = None + api_main._fleet = None + api_main._livemon = None + api_main._listener = None + api_main._obs_db = None + + +@pytest.mark.asyncio +async def test_driver_happy_path(driver_env): + """pending → scanning → starting_fleet → running with cells + + runners populated. Listener and fleet mocks observe the right + calls in order.""" + api_main = driver_env + + cells = [ + CellInfo(arfcn=73, freq="939.6M", cid=100, lac=200, mcc=208, mnc=1, power=-50), + CellInfo(arfcn=119, freq="947.4M", cid=12345, lac=234, mcc=208, mnc=10, power=-45), + ] + + async def fake_scan(**_): + return cells + + fake_fleet = MagicMock() + fake_fleet.start = AsyncMock(return_value=[ + RunnerSummary(cell=c, serverport=4730 + i, status=MagicMock( + running=True, pid=1000 + i, freq=c.freq, started_at=1.0, stderr_tail="")) + for i, c in enumerate(cells) + ]) + fake_fleet.is_running = MagicMock(return_value=True) + fake_fleet.stop_all = AsyncMock(return_value=[]) + + with patch.object(api_main, "scan_band", fake_scan): + with patch.object(api_main, "LivemonFleet", lambda **_: fake_fleet): + job = api_main._job_registry.submit( + {"band": "GSM900", "max_cells": 2, "scan_timeout": 60.0} + ) + await api_main._drive_scan_auto_job(job) + + assert job.state == "running" + assert job.cells_found == 2 + assert len(job.cells_selected) == 2 + assert len(job.runners) == 2 + api_main._listener.stop.assert_called() + api_main._listener.start.assert_called() + fake_fleet.start.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_driver_empty_scan_lands_in_stopped(driver_env): + """No cells found → state=stopped (not failed).""" + api_main = driver_env + + async def empty_scan(**_): + return [] + + with patch.object(api_main, "scan_band", empty_scan): + job = api_main._job_registry.submit({"band": "GSM900", "max_cells": 3}) + await api_main._drive_scan_auto_job(job) + + assert job.state == "stopped" + assert job.cells_found == 0 + + +@pytest.mark.asyncio +async def test_driver_scanner_timeout_lands_in_failed(driver_env): + """scan_band raising TimeoutError → state=failed + error set + + listener torn down.""" + api_main = driver_env + + async def timing_out(**_): + raise asyncio.TimeoutError + + with patch.object(api_main, "scan_band", timing_out): + job = api_main._job_registry.submit({"band": "GSM900", "max_cells": 1}) + await api_main._drive_scan_auto_job(job) + + assert job.state == "failed" + assert job.error is not None + # cleanup also stops listener — at minimum once (initial scan-prep stop). + assert api_main._listener.stop.await_count >= 1 + + +@pytest.mark.asyncio +async def test_driver_scanner_runtime_error_lands_in_failed(driver_env): + """scan_band raising a generic RuntimeError → state=failed.""" + api_main = driver_env + + async def broken_scan(**_): + raise RuntimeError("simulated grgsm_scanner exit 1") + + with patch.object(api_main, "scan_band", broken_scan): + job = api_main._job_registry.submit({"band": "GSM900", "max_cells": 1}) + await api_main._drive_scan_auto_job(job) + + assert job.state == "failed" + assert "simulated" in (job.error or "") + + +@pytest.mark.asyncio +async def test_driver_fleet_start_failure_lands_in_failed(driver_env): + """scan returns cells but fleet.start raises → state=failed + + listener torn back down (no orphan socket).""" + api_main = driver_env + + cells = [CellInfo(arfcn=73, freq="939.6M", cid=100, lac=200, mcc=208, mnc=1, power=-50)] + + async def fake_scan(**_): + return cells + + fake_fleet = MagicMock() + fake_fleet.start = AsyncMock(side_effect=OSError("device busy")) + fake_fleet.stop_all = AsyncMock(return_value=[]) + + with patch.object(api_main, "scan_band", fake_scan): + with patch.object(api_main, "LivemonFleet", lambda **_: fake_fleet): + job = api_main._job_registry.submit({"band": "GSM900", "max_cells": 1}) + await api_main._drive_scan_auto_job(job) + + assert job.state == "failed" + assert "device busy" in (job.error or "") + + +@pytest.mark.asyncio +async def test_driver_cancellation_during_scan(driver_env): + """Cancelling the task during scan → state=cancelled with + cleanup. The driver's CancelledError handler stops the listener + and clears _fleet/_consume_task.""" + api_main = driver_env + + started_evt = asyncio.Event() + + async def slow_scan(**_): + started_evt.set() + await asyncio.sleep(60) + return [] + + with patch.object(api_main, "scan_band", slow_scan): + job = api_main._job_registry.submit({"band": "GSM900", "max_cells": 1}) + task = asyncio.create_task(api_main._drive_scan_auto_job(job)) + await started_evt.wait() + # Now the task is inside slow_scan's sleep. + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert job.state == "cancelled" + assert api_main._fleet is None diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_scanner.py b/packages/secubox-sentinelle-gsm/api/tests/test_scanner.py new file mode 100644 index 00000000..ec3f853b --- /dev/null +++ b/packages/secubox-sentinelle-gsm/api/tests/test_scanner.py @@ -0,0 +1,181 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +"""Unit tests for sentinelle_gsm.scanner — parsing + diversity selection.""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from sentinelle_gsm.scanner import ( + CellInfo, + parse_scanner_output, + scan_band, + select_diverse, +) + + +# ── parse_scanner_output ───────────────────────────────────────────── + +def test_parse_one_cell(): + """The exact line format emitted by /usr/bin/grgsm_scanner line 558.""" + line = "ARFCN: 119, Freq: 947.4M, CID: 12345, LAC: 234, MCC: 208, MNC: 10, Pwr: -45" + cells = parse_scanner_output(line) + assert len(cells) == 1 + c = cells[0] + assert c.arfcn == 119 + assert c.freq == "947.4M" + assert c.cid == 12345 + assert c.lac == 234 + assert c.mcc == 208 + assert c.mnc == 10 + assert c.power == -45 + + +def test_parse_multiple_cells_mixed_with_noise(): + """Real-world output: banner lines, progress lines, partial lines, then cells.""" + out = """ +Args= rtl=0 +Wide bandwidth: 10.0 MHz +Try scan CCCH on 1-124 arfcn`s: +Scanning: 23.45% done.. +Scanning: 78.90% done.. + +ARFCN: 73, Freq: 939.6M, CID: 100, LAC: 200, MCC: 208, MNC: 1, Pwr: -55 +ARFCN: 119, Freq: 947.4M, CID: 12345, LAC: 234, MCC: 208, MNC: 10, Pwr: -45 +ARFCN: 1, Freq: 935.2M, CID: 50, LAC: 100, MCC: 208, MNC: 20, Pwr: -65 +""" + cells = parse_scanner_output(out) + assert len(cells) == 3 + assert [c.arfcn for c in cells] == [73, 119, 1] + assert [c.mnc for c in cells] == [1, 10, 20] + + +def test_parse_empty_output(): + assert parse_scanner_output("") == [] + assert parse_scanner_output("no cells found\n") == [] + + +def test_parse_partial_line_skipped(): + """Truncated lines must NOT produce half-populated CellInfo.""" + bad = "ARFCN: 119, Freq: 947.4M, CID: 12345, LAC: 234" # no MCC/MNC/Pwr + assert parse_scanner_output(bad) == [] + + +def test_parse_positive_power(): + """Power can also be positive (rare; close-quarters or scanner config).""" + line = "ARFCN: 1, Freq: 935.2M, CID: 1, LAC: 1, MCC: 208, MNC: 1, Pwr: 10" + cells = parse_scanner_output(line) + assert cells[0].power == 10 + + +# ── select_diverse ─────────────────────────────────────────────────── + +def _cell(arfcn: int, mcc: int, mnc: int, power: int) -> CellInfo: + return CellInfo(arfcn=arfcn, freq=f"{900+arfcn/10}M", cid=arfcn, lac=1, + mcc=mcc, mnc=mnc, power=power) + + +def test_select_diverse_picks_one_per_operator(): + """Three operators present, n=3 → one from each, strongest first.""" + cells = [ + _cell(1, 208, 1, -70), # Orange, weak + _cell(2, 208, 1, -50), # Orange, strong (preferred) + _cell(3, 208, 10, -60), # SFR + _cell(4, 208, 20, -55), # Bouygues + ] + chosen = select_diverse(cells, 3) + assert len(chosen) == 3 + # Strongest per operator: Orange ARFCN 2, Bouygues 4, SFR 3 (by power desc) + assert chosen[0].arfcn == 2 # power=-50, strongest overall + assert {c.mnc for c in chosen} == {1, 10, 20} + + +def test_select_diverse_falls_back_when_operators_exhausted(): + """Only 2 operators, n=4 → 2 strongest unique + 2 next-strongest.""" + cells = [ + _cell(1, 208, 1, -50), # Orange A + _cell(2, 208, 1, -55), # Orange B + _cell(3, 208, 10, -60), # SFR A + _cell(4, 208, 10, -65), # SFR B + ] + chosen = select_diverse(cells, 4) + assert len(chosen) == 4 + # First two: strongest from each operator (Orange ARFCN 1, SFR ARFCN 3) + # Next two: remaining strongest (Orange ARFCN 2, SFR ARFCN 4) + assert chosen[0].arfcn == 1 + assert chosen[1].arfcn == 3 + assert set(c.arfcn for c in chosen[2:]) == {2, 4} + + +def test_select_diverse_n_zero(): + assert select_diverse([_cell(1, 208, 1, -50)], 0) == [] + + +def test_select_diverse_n_larger_than_list(): + cells = [_cell(1, 208, 1, -50)] + chosen = select_diverse(cells, 5) + assert chosen == cells + + +# ── scan_band (subprocess wrapper) ─────────────────────────────────── + +@pytest.mark.asyncio +async def test_scan_band_parses_stdout(): + """Mocked grgsm_scanner stdout flows through to parse_scanner_output.""" + fake = MagicMock(spec=asyncio.subprocess.Process) + fake.returncode = 0 + fake.communicate = AsyncMock(return_value=( + b"ARFCN: 119, Freq: 947.4M, CID: 12345, LAC: 234, MCC: 208, MNC: 10, Pwr: -45\n", + b"", + )) + with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=fake)): + cells = await scan_band(band="GSM900") + assert len(cells) == 1 + assert cells[0].arfcn == 119 + + +@pytest.mark.asyncio +async def test_scan_band_argv_default_args(): + """--args=rtl=0 is the default — we always force the RTL-SDR backend. + grgsm_scanner accepts this where livemon rejects it (different code + paths).""" + fake = MagicMock(spec=asyncio.subprocess.Process) + fake.returncode = 0 + fake.communicate = AsyncMock(return_value=(b"", b"")) + with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=fake)) as mck: + await scan_band(band="GSM900", gain=45, ppm=10, speed=20) + args = mck.call_args[0] + assert "--args=rtl=0" in args + assert "-b" in args and "GSM900" in args + assert "-g" in args and "45" in args + assert "-p" in args and "10" in args + assert "--speed" in args and "20" in args + + +@pytest.mark.asyncio +async def test_scan_band_raises_on_nonzero_exit(): + """Surface scanner crashes so /scan/auto can return a real error.""" + fake = MagicMock(spec=asyncio.subprocess.Process) + fake.returncode = 1 + fake.communicate = AsyncMock(return_value=(b"", b"some scanner failure")) + with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=fake)): + with pytest.raises(RuntimeError, match="exit 1"): + await scan_band(band="GSM900") + + +@pytest.mark.asyncio +async def test_scan_band_timeout_kills_proc(): + """When scanner hangs past the timeout, we kill it and raise.""" + fake = MagicMock(spec=asyncio.subprocess.Process) + fake.returncode = None + fake.communicate = AsyncMock(side_effect=asyncio.TimeoutError) + fake.kill = MagicMock() + fake.wait = AsyncMock(return_value=-9) + with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=fake)): + with pytest.raises(asyncio.TimeoutError): + await scan_band(band="GSM900", timeout=0.01) + fake.kill.assert_called_once() diff --git a/packages/secubox-sentinelle-gsm/debian/changelog b/packages/secubox-sentinelle-gsm/debian/changelog index 5712c236..833469fe 100644 --- a/packages/secubox-sentinelle-gsm/debian/changelog +++ b/packages/secubox-sentinelle-gsm/debian/changelog @@ -1,3 +1,77 @@ +secubox-sentinelle-gsm (0.3.6-1~bookworm1) bookworm; urgency=medium + + * lib/sentinelle_gsm/scan_auto_job.py: NEW async-job model for + /scan/auto. v0.3.5 ran the scanner synchronously in the HTTP + handler — fine for unit tests, brutal on real RF where a GSM900 + sweep takes 8-15 minutes on the MOCHAbin's Cortex-A72 and blocks + the request thread the whole time. v0.3.6 makes /scan/auto + fire-and-poll: + - POST /scan/auto returns {id, state=pending} immediately and + spawns the driver coroutine in the background + - GET /scan/auto/jobs/{id} polls state + progress + - GET /scan/auto/jobs lists recent (newest first, capped at + max_history=10 terminal jobs by JobRegistry) + - POST /scan/auto/jobs/{id}/cancel aborts in-flight scans + (Task.cancel + driver's CancelledError handler cleans up + listener + fleet) + State machine: pending → scanning → starting_fleet → running → + stopped, with failed and cancelled as off-ramps. + * api/main.py: refactored /scan/auto handler — validate, submit + job, spawn driver task, return id. New endpoints + /scan/auto/jobs, /scan/auto/jobs/{id}, /scan/auto/jobs/{id}/cancel. + /scan/auto/status unchanged (still reflects the active fleet). + /scan/start now also 409s if a scan-auto job is in flight even + before it has spawned the fleet (state=scanning has no _fleet + yet, but starting a single-runner would stomp on the scanner's + 4729 bind). /scan/stop marks the active job as `stopped` when + it tears the fleet down. + * api/tests/test_scan_auto_job.py: NEW. 12 tests covering the + registry FIFO eviction + the driver coroutine state machine + (happy path, empty-scan, scanner timeout, scanner runtime error, + fleet-start failure, cancellation mid-scan). + * api/tests/test_scan_api.py: refactored for async semantics — + POST returns immediately, the in-flight tests seed the registry + directly rather than rely on TestClient's task-scope (Starlette + TestClient cancels spawned tasks at request-scope exit; the + direct-driver tests live in test_scan_auto_job.py instead). + * 114/114 sweep (up from 96 in v0.3.5). + * Closes #357. + + -- Gerald Kerma Fri, 22 May 2026 23:30:00 +0000 + +secubox-sentinelle-gsm (0.3.5-1~bookworm1) bookworm; urgency=medium + + * lib/sentinelle_gsm/scanner.py: NEW wrapper around grgsm_scanner. + Parses the canonical output line (`ARFCN: %4u, Freq: %6.1fM, CID: + %5u, LAC: %5u, MCC: %3u, MNC: %3u, Pwr: %3i`) into a list of + CellInfo. `select_diverse(cells, n)` picks up to n cells, prefering + one per (MCC, MNC) operator and breaking ties by power — feeds the + scoring engine's cross-operator heuristics (ghost_bts, + anomalous_neighbours, identity_mismatch). + * lib/sentinelle_gsm/livemon_fleet.py: NEW LivemonFleet — manages N + LivemonRunner instances, each on `--serverport = base + i` + (default base=4730, mirroring the IMSI-catcher scan-and-livemon + pattern). All runners emit GSMTAP to the listener on 4729; the + listener demuxes by GSMTAP header ARFCN. Partial-start rollback + + explicit stop_all() semantics. RunnerSummary keeps the cell + context alongside each runner's ScanStatus so the API can echo + "which operator / which ARFCN" per runner. + * api/main.py: NEW POST /scan/auto endpoint. Body: + {band, max_cells, gain, ppm, samp_rate, speed, scan_timeout}. + Refuses if /scan/start single-runner OR a fleet is already + running. Lifecycle: stop listener → scan_band → restart listener + → start fleet. Returns mode/cells_found/cells_selected/runners. + NEW GET /scan/auto/status. POST /scan/stop now stops the fleet + in addition to the single runner; payload shape is + polymorphic on which path was active. + * api/tests/: NEW test_scanner.py (12 tests), test_livemon_fleet.py + (7 tests), and 6 new tests in test_scan_api.py for /scan/auto + happy path, 409/400/504 negative paths, and /scan/auto/status. + 96/96 sweep (up from 71 in v0.3.4). + * Closes #356. + + -- Gerald Kerma Fri, 22 May 2026 23:00:00 +0000 + secubox-sentinelle-gsm (0.3.4-1~bookworm1) bookworm; urgency=medium * sbin/secubox-grgsm-livemon-shim: NEW thin Python wrapper that diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/livemon_fleet.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/livemon_fleet.py new file mode 100644 index 00000000..91a0a24e --- /dev/null +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/livemon_fleet.py @@ -0,0 +1,136 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +Multi-cell capture: N LivemonRunners, each watching a different ARFCN, +all emitting GSMTAP to the same listener on port 4729 (default). + +The gkerma/IMSI-catcher project's scan-and-livemon script demonstrates +this pattern. grgsm_livemon_headless's --serverport BINDs a control +port (not the GSMTAP destination, which is hardcoded to 4729), so each +runner needs a UNIQUE --serverport — we use base=4730, base+1=4731, … +LivemonRunner v0.3.4 already supports a serverport kwarg and defaults +it to gsmtap_port+1, so the fleet just bumps it per runner. + +The L3 scoring engine (lib/sentinelle_gsm/scoring_engine.py) was always +designed to operate on multi-cell data — ghost_bts and +anomalous_neighbours heuristics return false negatives when only one +cell is in view. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +from typing import Optional + +from .livemon_runner import LivemonRunner, ScanStatus +from .scanner import CellInfo + + +@dataclasses.dataclass +class RunnerSummary: + """Per-runner status enriched with the cell context that started it. + + Distinct from ScanStatus (which knows nothing about MCC/MNC/CID) + because the API endpoint wants to echo enough to identify which + operator/cell each runner is watching. + """ + + cell: CellInfo + serverport: int + status: ScanStatus + + +class LivemonFleet: + def __init__(self, gsmtap_port: int = 4729, base_serverport: int = 4730): + self.gsmtap_port = gsmtap_port + self.base_serverport = base_serverport + # Parallel arrays — index i ties together the runner, its + # serverport, and the cell it was started for. + self._runners: list[LivemonRunner] = [] + self._cells: list[CellInfo] = [] + self._serverports: list[int] = [] + + def is_running(self) -> bool: + """True if any runner is still alive. Mirrors LivemonRunner's + `not _done` semantics — a fleet with all children dead is not + running, even if cleanup hasn't been called.""" + return any(r.status().running for r in self._runners) + + def summaries(self) -> list[RunnerSummary]: + return [ + RunnerSummary(cell=c, serverport=p, status=r.status()) + for r, c, p in zip(self._runners, self._cells, self._serverports) + ] + + async def start( + self, + cells: list[CellInfo], + gain: Optional[float] = None, + ppm: Optional[float] = None, + samp_rate: Optional[str] = None, + ) -> list[RunnerSummary]: + """Spawn one LivemonRunner per cell. + + Fails fast on the first runner that doesn't come up — stops + whatever started successfully before raising, so the fleet + never lingers in a partially-started state. + """ + if self._runners: + raise RuntimeError("fleet already running — stop first") + if not cells: + raise ValueError("no cells to watch") + + for i, cell in enumerate(cells): + runner = LivemonRunner(gsmtap_port=self.gsmtap_port) + serverport = self.base_serverport + i + try: + await runner.start( + cell.freq, + gain=gain, + ppm=ppm, + samp_rate=samp_rate, + serverport=serverport, + ) + except Exception: + # Roll back anything we did start before re-raising. + await self._stop_all_quiet() + raise + self._runners.append(runner) + self._cells.append(cell) + self._serverports.append(serverport) + return self.summaries() + + async def _stop_all_quiet(self) -> None: + """Best-effort stop of every runner so an aborted start() can + unwind cleanly. Errors swallowed — we're already on the failure + path.""" + for r in self._runners: + try: + await r.stop() + except Exception: + pass + self._runners.clear() + self._cells.clear() + self._serverports.clear() + + async def stop_all(self) -> list[RunnerSummary]: + """Send SIGTERM to every runner, await exit, capture final + status. Returns the pre-clear summaries so callers can show + each runner's stderr_tail one last time.""" + # Capture the per-runner final status (with cell + serverport + # context) BEFORE we clear the lists. + final: list[RunnerSummary] = [] + for runner, cell, sp in zip(self._runners, self._cells, self._serverports): + try: + st = await runner.stop() + except Exception as e: + st = ScanStatus(stderr_tail=f"stop() raised: {e}") + final.append(RunnerSummary(cell=cell, serverport=sp, status=st)) + self._runners.clear() + self._cells.clear() + self._serverports.clear() + return final diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scan_auto_job.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scan_auto_job.py new file mode 100644 index 00000000..e4e74f5f --- /dev/null +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scan_auto_job.py @@ -0,0 +1,163 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +Async-job state for /scan/auto. v0.3.5 ran the scanner synchronously in +the HTTP handler — fine for unit tests, brutal on real RF where a +GSM900 sweep takes 8-15 minutes on the MOCHAbin's Cortex-A72 and +blocks the request thread (and any TestClient assertions) the whole +time. + +This module owns: + - ScanAutoJob: per-request state (params, lifecycle timestamps, + discovered cells, fleet summaries, error message) + - JobRegistry: process-local store of jobs with a fixed-size FIFO + retention policy. NOT durable across service restarts — the + operational "what's happening right now" view, not history. + Sightings persist in observations.db as before. + +The actual coroutine that drives a job lives in api/main.py (it needs +the listener/runner/fleet globals); this module just holds state. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import time +import uuid +from typing import Optional + +from .livemon_fleet import LivemonFleet, RunnerSummary +from .scanner import CellInfo + + +# State machine: +# +# pending → scanning → starting_fleet → running → stopped +# ↘ failed +# ↘ failed +# ↘ cancelled +# +# Terminal states (job is done, no further transitions): stopped, +# failed, cancelled. The registry uses this to decide what to keep in +# the FIFO and whether a new POST /scan/auto can run. +TERMINAL_STATES = {"stopped", "failed", "cancelled"} + + +@dataclasses.dataclass +class ScanAutoJob: + """One end-to-end /scan/auto request. + + `state` is the source of truth for "what is happening now"; the + other fields fill in as the job progresses. Callers should NOT + mutate ScanAutoJob from outside its driving coroutine — except + `cancel()`, which is safe to call from any task. + """ + + id: str + params: dict + state: str = "pending" + started_at: float = 0.0 + finished_at: Optional[float] = None + cells_found: int = 0 + cells_selected: list[CellInfo] = dataclasses.field(default_factory=list) + runners: list[RunnerSummary] = dataclasses.field(default_factory=list) + error: Optional[str] = None + # Set when the job enters `starting_fleet` and survives until the + # fleet stops. None during scanning and after terminal cleanup. + fleet: Optional[LivemonFleet] = None + # Set when the job's driver coroutine starts; None for jobs in + # terminal state. We hold a reference so /scan/stop and cancel() + # can target the task. + task: Optional[asyncio.Task] = None + + def is_terminal(self) -> bool: + return self.state in TERMINAL_STATES + + def is_active(self) -> bool: + """True if any /scan/auto path is in flight for this job — + i.e. NOT in a terminal state.""" + return not self.is_terminal() + + def set_state(self, new: str) -> None: + """Centralised state writer that records finished_at on + transitions into terminal states.""" + self.state = new + if new in TERMINAL_STATES and self.finished_at is None: + self.finished_at = time.time() + + +class JobRegistry: + """Process-local store of /scan/auto jobs. + + - At most one ACTIVE job (non-terminal). Enforced by + submit_or_reject() — callers must check the result. + - History capped at `max_history` terminal jobs (FIFO eviction). + - Newest-first ordering for the /scan/auto/jobs listing. + """ + + def __init__(self, max_history: int = 10): + self._jobs: dict[str, ScanAutoJob] = {} + self._order: list[str] = [] # newest-first + self.max_history = max_history + + def active(self) -> Optional[ScanAutoJob]: + """The single in-flight job, or None if nothing's running.""" + for jid in self._order: + j = self._jobs.get(jid) + if j and j.is_active(): + return j + return None + + def get(self, job_id: str) -> Optional[ScanAutoJob]: + return self._jobs.get(job_id) + + def list(self, limit: int = 10) -> list[ScanAutoJob]: + out: list[ScanAutoJob] = [] + for jid in self._order: + j = self._jobs.get(jid) + if j: + out.append(j) + if len(out) >= limit: + break + return out + + def submit(self, params: dict) -> ScanAutoJob: + """Create a new job in `pending` state and insert it at the + head of the list. The caller is responsible for refusing to + submit when another job is active — this method does NOT + check, so unit tests can construct multi-job scenarios. + """ + job = ScanAutoJob( + id=str(uuid.uuid4()), + params=params, + started_at=time.time(), + ) + self._jobs[job.id] = job + self._order.insert(0, job.id) + self._evict_old() + return job + + def _evict_old(self) -> None: + """Keep only the most recent active job + max_history + terminal jobs. Anything older is dropped.""" + kept_terminal = 0 + keep: list[str] = [] + for jid in self._order: + j = self._jobs.get(jid) + if j is None: + continue + if j.is_active(): + keep.append(jid) + continue + if kept_terminal < self.max_history: + keep.append(jid) + kept_terminal += 1 + # Drop everything not in `keep`. + for jid in list(self._jobs): + if jid not in keep: + del self._jobs[jid] + self._order = keep diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scanner.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scanner.py new file mode 100644 index 00000000..39bceaf2 --- /dev/null +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scanner.py @@ -0,0 +1,187 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +Wraps grgsm_scanner to discover GSM cells before /scan/auto spawns a +LivemonFleet. grgsm_scanner sweeps a GSM band, decodes each ARFCN's +BCCH long enough to read MCC/MNC/LAC/CID/power, then prints one line +per cell: + + ARFCN: %4u, Freq: %6.1fM, CID: %5u, LAC: %5u, MCC: %3u, MNC: %3u, Pwr: %3i + +We subprocess it, parse those lines, and return a list[CellInfo]. + +Two operational quirks worth knowing: + + 1. grgsm_scanner hard-binds UDP 4729 (the same port our GSMTAP + listener binds). The caller MUST stop the listener before invoking + scan_band(). main.py's /scan/auto handler does that. + 2. grgsm_scanner accepts --args=rtl=0 (which the broken match() code + path never sees), so we don't need the device.py shim here. The + device.py bug only bites when --args is omitted, and we always + pass it. +""" + +from __future__ import annotations + +import asyncio +import dataclasses +import re +import shutil +from typing import Optional + + +# Output line emitted by grgsm_scanner for each discovered cell. +# Generated at /usr/bin/grgsm_scanner line 558: +# return "ARFCN: %4u, Freq: %6.1fM, CID: %5u, LAC: %5u, MCC: %3u, +# MNC: %3u, Pwr: %3i" % (...) +_CELL_LINE_RE = re.compile( + r"ARFCN:\s*(?P\d+),\s*" + r"Freq:\s*(?P[\d.]+)M,\s*" + r"CID:\s*(?P\d+),\s*" + r"LAC:\s*(?P\d+),\s*" + r"MCC:\s*(?P\d+),\s*" + r"MNC:\s*(?P\d+),\s*" + r"Pwr:\s*(?P-?\d+)" +) + + +@dataclasses.dataclass(frozen=True) +class CellInfo: + """One discovered GSM cell. + + `freq` is a CLI-friendly string ("947.4M") that LivemonRunner.start + accepts directly. `power` is the dBm value grgsm_scanner reports — + higher (closer to 0) = stronger signal. + """ + + arfcn: int + freq: str + cid: int + lac: int + mcc: int + mnc: int + power: int + + +def parse_scanner_output(stdout: str) -> list[CellInfo]: + """Pull CellInfo records out of grgsm_scanner stdout. + + Ignores progress lines ("Scanning: 23.45% done.."), banner/info + output, and any partial lines. Order preserved (scanner emits in + ARFCN order; callers reorder by power). + """ + cells: list[CellInfo] = [] + for line in stdout.splitlines(): + m = _CELL_LINE_RE.search(line) + if not m: + continue + cells.append( + CellInfo( + arfcn=int(m["arfcn"]), + freq=f"{m['freq']}M", + cid=int(m["cid"]), + lac=int(m["lac"]), + mcc=int(m["mcc"]), + mnc=int(m["mnc"]), + power=int(m["pwr"]), + ) + ) + return cells + + +def select_diverse(cells: list[CellInfo], n: int) -> list[CellInfo]: + """Pick up to `n` cells, maximising operator spread (mcc+mnc) first + and breaking ties by power. + + Strategy: sort by power desc, then walk the list once, taking the + strongest cell per (mcc, mnc) until we have n; if we exhaust unique + operators before hitting n, take the next-strongest remaining cell + regardless of operator. + + Useful for IMSI-catcher detection: a real area has at least 3 + operators on different ARFCNs, so spreading the fleet across them + gives the L3 scoring engine the cross-operator signal it needs + (ghost_bts, anomalous_neighbours, identity_mismatch). + """ + if n <= 0: + return [] + by_power = sorted(cells, key=lambda c: -c.power) + seen_ops: set[tuple[int, int]] = set() + primary: list[CellInfo] = [] + remaining: list[CellInfo] = [] + for c in by_power: + op = (c.mcc, c.mnc) + if op in seen_ops: + remaining.append(c) + else: + seen_ops.add(op) + primary.append(c) + if len(primary) >= n: + break + if len(primary) >= n: + return primary[:n] + # ran out of operators — pad with remaining strongest. + return primary + remaining[: n - len(primary)] + + +async def scan_band( + band: str = "GSM900", + gain: Optional[float] = None, + ppm: Optional[float] = None, + samp_rate: Optional[str] = None, + speed: Optional[int] = None, + args: str = "rtl=0", + timeout: float = 180.0, +) -> list[CellInfo]: + """Run grgsm_scanner and return discovered cells. + + band: one of GSM900/DCS1800/GSM850/PCS1900/GSM450/GSM480/GSM-R. + gain/ppm/samp_rate/speed: passed straight to grgsm_scanner. + args: gr-osmosdr device args. Default rtl=0 forces the RTL-SDR + backend (and dodges the device.py auto-pick path that gr-gsm + 1.0.0 mis-implements — see the shim in livemon_runner.py). + timeout: hard kill after this many seconds. GSM900 takes ~30-90s + in practice depending on --speed; 180s is generous. + + Prerequisite: the caller MUST ensure UDP 4729 is free (stop the + GSMTAP listener), otherwise grgsm_scanner crashes with + `RuntimeError: bind: Address already in use`. + + Raises: + FileNotFoundError if grgsm_scanner isn't installed. + asyncio.TimeoutError if it doesn't finish in `timeout` seconds. + RuntimeError with stderr_tail on any non-zero exit. + """ + bin_path = shutil.which("grgsm_scanner") or "/usr/bin/grgsm_scanner" + argv = [bin_path, f"--args={args}", "-b", band] + if gain is not None: + argv += ["-g", str(gain)] + if ppm is not None: + argv += ["-p", str(ppm)] + if samp_rate is not None: + argv += ["-s", str(samp_rate)] + if speed is not None: + argv += ["--speed", str(speed)] + + proc = await asyncio.create_subprocess_exec( + *argv, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except asyncio.TimeoutError: + proc.kill() + await proc.wait() + raise + + if proc.returncode != 0: + raise RuntimeError( + f"grgsm_scanner exit {proc.returncode}: " + f"{stderr.decode(errors='replace')[-2000:]}" + ) + + return parse_scanner_output(stdout.decode(errors="replace"))