mirror of
https://github.com/CyberMind-FR/secubox-deb.git
synced 2026-07-29 09:14:33 +00:00
feat(sentinelle-gsm): v0.3.5 + v0.3.6 — scan-auto multi-cell + async-job pattern (closes #356, #357) (#358)
* 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 <gandalf@Gk2.net>
This commit is contained in:
CyberMind
GitHub
parent
604099f7bd
commit
fe834d4fe8
|
|
@ -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)."""
|
||||
|
|
|
|||
163
packages/secubox-sentinelle-gsm/api/tests/test_livemon_fleet.py
Normal file
163
packages/secubox-sentinelle-gsm/api/tests/test_livemon_fleet.py
Normal file
|
|
@ -0,0 +1,163 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
# 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
|
||||
|
|
@ -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"] == []
|
||||
|
|
|
|||
265
packages/secubox-sentinelle-gsm/api/tests/test_scan_auto_job.py
Normal file
265
packages/secubox-sentinelle-gsm/api/tests/test_scan_auto_job.py
Normal file
|
|
@ -0,0 +1,265 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
# 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
|
||||
181
packages/secubox-sentinelle-gsm/api/tests/test_scanner.py
Normal file
181
packages/secubox-sentinelle-gsm/api/tests/test_scanner.py
Normal file
|
|
@ -0,0 +1,181 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
# 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()
|
||||
|
|
@ -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 <devel@cybermind.fr> 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 <devel@cybermind.fr> 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
|
||||
|
|
|
|||
|
|
@ -0,0 +1,136 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
# 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
|
||||
|
|
@ -0,0 +1,163 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
# 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
|
||||
187
packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scanner.py
Normal file
187
packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scanner.py
Normal file
|
|
@ -0,0 +1,187 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
# 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<arfcn>\d+),\s*"
|
||||
r"Freq:\s*(?P<freq>[\d.]+)M,\s*"
|
||||
r"CID:\s*(?P<cid>\d+),\s*"
|
||||
r"LAC:\s*(?P<lac>\d+),\s*"
|
||||
r"MCC:\s*(?P<mcc>\d+),\s*"
|
||||
r"MNC:\s*(?P<mnc>\d+),\s*"
|
||||
r"Pwr:\s*(?P<pwr>-?\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"))
|
||||
Loading…
Reference in New Issue
Block a user