mirror of
https://github.com/CyberMind-FR/secubox-deb.git
synced 2026-07-29 11:12:29 +00:00
feat(secubox-sentinelle-gsm): v0.3.1 — L3 decode + scoring engine + baseline + qualified alerts (closes #349) (#350)
* docs(plan): #349 sentinelle-gsm v0.3.1 L3 + scoring + baseline (ref #349) * feat(sentinelle-gsm): GsmtapListener exposes raw_l3 payload bytes (ref #349) v0.3.0's Observation carried only header metadata (arfcn, frame_nr, channel, sub_type), which was enough to prove the GSMTAP pipe but not to drive L3-aware scoring. v0.3.1 needs the post-header payload to decode BCCH System Information (LAC/CI/MCC/MNC) and CCCH paging requests (IMSI/TMSI for the privacy-hashed subscriber digest). Slice the L3 payload from the datagram via data[hdr["hdr_len"]:] in _on_datagram() and surface it as a new Observation field `raw_l3: bytes = b""`. Default empty bytes keeps every existing caller backward compatible — only downstream decoders that opt in will consume it (wired in Task 5). Test: existing test_listener_receives_a_datagram now appends a known 18-byte payload and asserts obs.raw_l3 matches verbatim. * feat(sentinelle-gsm): l3_decode — BCCH SI + CCCH paging request parsing (ref #349) Pure-Python TLV decoder for the minimal L3 subset needed by the IMSI-catcher scoring engine: BCCH System Information Types 3/4/6 and CCCH Paging Request Types 1/2/3. No scapy dependency — the parser operates directly on the post-GSMTAP-header bytes exposed by Observation.raw_l3 (Task 1). Privacy invariant : the public dataclasses (CellInfo, PagedIdentity, ParsedPagingRequest, ParsedFrame) NEVER carry plaintext IMSI/TMSI/IMEI. The internal _try_extract_mobile_id() returns plaintext bytes; _parse_paging() immediately hands them to Anonymizer.anonymize() and only the HMAC-truncated subscriber_hash escapes into PagedIdentity. The plaintext goes out of scope at function return. Permissive parsing : every helper returns an empty CellInfo() / None on unexpected layouts rather than raising — real GSMTAP frames have surprising structural variants and exceptions in the consume loop would crash livemon ingestion. Tests : 9 cases (6 from the plan + 3 added for coverage). BCD encoder/decoder round-trip helpers build synthetic frames programmatically rather than hand-crafting bytes, which also validates the TS 24.008 §10.5.1.3/4 nibble ordering. * feat(sentinelle-gsm): baseline.py — operator-baseline learning + cell_baseline table (ref #349) Adds CellBaseline wrapper + cell_baseline table colocated in observations.db. Cells graduate to baseline once observed >= LEARN_THRESHOLD (default 3) times. Key design choices: * LEARN_THRESHOLD = 3 by default — three sightings before a cell is considered part of the operator's legitimate baseline. * set_learn_mode(seconds) skips the count entirely — every cell observed inside that explicit-learn window is INSERTed with learn_count = 3 and graduates on first sighting (used by the operator's calibration sweep). * CellBaseline accepts a raw sqlite3.Connection (NOT an ObservationsDB instance) so it stays decoupled from observations.py's public surface while still sharing the same .db file. * cell_baseline table is created idempotently (IF NOT EXISTS) in ObservationsDB.__init__, so pre-existing observations.db files keep working after upgrade. Privacy invariant: BaselineCell + cell_baseline have NO subscriber-id columns — paged identities never enter the baseline path. Tests (5/5 passing): - test_first_consider_starts_at_1_not_baseline_yet - test_three_considers_graduate_to_baseline - test_learn_mode_graduates_first_consider - test_consider_updates_metadata_via_coalesce - test_list_orders_by_last_learned_desc Full pkg sweep: 62 passing (57 prior + 5 new). * feat(sentinelle-gsm): scoring_engine — 8 heuristics + thresholds (ref #349) Replaces the v0.1 shape-only scoring.py stub with the full v0.3.1 detector brain. ScoringEngine wires against CellBaseline + L3Decode (Tasks 2-3) and runs every parsed frame through 8 heuristics, summing the triggered contributions into a CellScore clamped to [0, 100]. Heuristics + default scores (sum-clamped to 100): 1. cipher_downgrade 40 A5/X observed < baseline.cipher_a5 2. ghost_bts 35 cell unknown to baseline + paging seen 3. identity_mismatch 30 observed (mcc, mnc) != baseline pair 4. relocalization_storm 25 > 3 distinct LACs in 60s on cell 5. identity_request_abuse 35 > 2 paging reqs in 300s per hash 6. anomalous_neighbours 15 neighbour list went non-empty -> empty 7. t3212_out_of_band 15 T3212 outside [1, 60] min 8. orphan_arfcn 20 FR carrier announces ARFCN outside plan Rolling-window state (in-memory only, no DB persistence in v0.3.1): _lac_window cell_id -> deque[(ts, lac)] _idreq_window subscriber_hash -> deque[ts] _neighbour_baseline cell_id -> set[int] (cumulative ARFCNs) Thresholds: * DEFAULT_THRESHOLDS class constant carries per-heuristic config dicts. * update_thresholds() does a per-heuristic deep-merge so callers can pass partial updates like {"cipher_downgrade": {"score": 55}} without clobbering "enabled" or any other unspecified field. * thresholds() returns a deep-copy so callers can't mutate engine state. Privacy invariant kept: identity_request_abuse keys windows on subscriber_hash strings (HMAC-trunc, supplied by l3_decode.Anonymizer), never plaintext IMSI/TMSI. Reasons truncate the hash to 8 chars defensively. France GSM-900 ARFCN plan is approximate and permissive: unknown (mcc, mnc) pairs return triggered=False to avoid false-positives. Dropped lib/sentinelle_gsm/scoring.py — the v0.1.0 shape-only stub (empty Baseline + score()=0) is replaced. No other module imported it. Tests: * api/tests/test_scoring_engine.py — 11 tests, one per heuristic (plus permissive-unknown-carrier guard), one for evaluate() aggregation/clamping, one for update_thresholds() deep-merge. * Full sweep: 73 passed (was 62; +11 new). * feat(sentinelle-gsm): consume loop wires L3 + scoring + trusted match → Alert (ref #349) - Extracted `_process_observation(obs)` helper for testability — the consume coroutine now delegates per-frame work so tests can drive synthesized Observations through the full pipeline without binding UDP. - New endpoints: GET /baseline, POST /baseline/learn (sweep window), GET /scoring/thresholds, POST /scoring/thresholds (deep-merge). - Audit log entry on POST /scoring/thresholds via the existing `secubox.sentinelle-gsm.api` logger — surfaces in /journal/stream. - Alert emission decision tree: score < threshold → drop (no write) score >= threshold + match → one labeled alert per matched trusted phone score >= threshold + no match → single anomaly-only alert - ALERT_THRESHOLD configurable via SENTINELLE_ALERT_THRESHOLD env (default 60). - TrustedRegistry now hashes the IMSI as `imsi.encode("ascii").hex()` so the persisted token matches the canonical form produced by the L3 decoder's paging-request path. Existing trusted tests don't pin the hash value, so behaviour is unchanged for read-back / lookup-by-id. - 5 new integration tests at api/tests/test_alert_emission.py cover the below-threshold drop, anomaly-only emission, trusted-match labeling, paging-event hash persistence, and audit-log emission on threshold update. Full sweep 78/78 passing. * feat(sentinelle-gsm): webui baseline + scoring panels + trusted_label chip (ref #349) Adds two new webui panels and a trusted-label chip on the alerts row: - #baseline panel: Cell ID / MCC / MNC / LAC / ARFCN / cipher / learn count / last learned table backed by GET /api/v1/sensor/gsm/baseline. "Start learn (5 min)" button arms POST /baseline/learn?sweep_seconds=300 and re-polls the table every 15s for the 5-minute window so cells appear as they graduate. - #scoring panel: per-heuristic enable + score input (0..100, clamped client-side) backed by GET/POST /scoring/thresholds. Form values are cached in _thresholdState so any backend-only fields (heuristic-specific knobs) survive the POST round trip. - Alerts table: trusted_label is now rendered as a violet pill via the new .trusted-chip class (replaces the older .target-pill markup for the same data path). Local navigation gets two new anchors (Baseline, Scoring) between Observations and Actions. The 10s scan poll now also refreshes baseline while a scan is running. Wires 4 new event listeners (refresh-baseline, baseline-learn, refresh-scoring, save-scoring). No backend changes; all endpoints already exist (Tasks 1..5). Tests: 78/78 passing. * chore(sentinelle-gsm): bump 0.3.1 + extend privacy invariant tests (closes #349) Append 4 new structural privacy tests under tests/test_privacy_invariant.py to lock the v0.3.1 invariants in place: * test_l3_decode_returns_no_plaintext_fields — CellInfo, PagedIdentity, and ParsedPagingRequest dataclasses must not carry any plaintext subscriber-id field (imsi/tmsi/imei/msisdn/iccid/subscriber_id). * test_paging_request_hashes_paged_identities — feeds a synthetic Paging Request Type 1 with a known TMSI through L3Decode and asserts the plaintext TMSI bytes (as hex) NEVER appear in the resulting PagedIdentity.subscriber_hash. * test_cell_baseline_has_no_subscriber_fields — the BaselineCell dataclass must remain operator-side metadata only: no subscriber_hash, no plaintext identifier. * test_scoring_engine_reasons_contain_no_plaintext_imsi — the free-text reason strings produced by ScoringEngine.evaluate() must never echo a 15-digit token (plaintext-IMSI shape) for typical observations. These four tests pass against the v0.3.1 modules (Tasks 1-6) with no code change — the modules were designed for these invariants from the start, this commit just locks them in regression-test form. Full sweep: 78 -> 82 passing. Changelog: 0.3.0 -> 0.3.1 with the v0.3.1 entry summarising the new l3_decode, baseline, scoring_engine modules, the consume-loop wiring that emits qualified Alerts with trusted_label, the webui panels, and the updated test breakdown (l3_decode x9, baseline x5, scoring_engine x11, alert_emission x5, privacy x4 = 34 new, full sweep 82/82). --------- Co-authored-by: CyberMind-FR <gandalf@Gk2.net>
This commit is contained in:
CyberMind
GitHub
parent
5f3d268c05
commit
0ca16778cc
File diff suppressed because it is too large
Load Diff
|
|
@ -41,10 +41,17 @@ sys.path.insert(0, "/usr/lib/secubox/sentinelle-gsm/lib")
|
|||
|
||||
from sentinelle_gsm import CAPTURES_PLAINTEXT_IMSI, Mode # noqa: E402
|
||||
from sentinelle_gsm.alert_sink import Alert, AlertSink # noqa: E402
|
||||
from sentinelle_gsm.baseline import CellBaseline # noqa: E402
|
||||
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_runner import LivemonRunner, detect_rtlsdr_usb # noqa: E402
|
||||
from sentinelle_gsm.observations import ObservationsDB, Sighting # noqa: E402
|
||||
from sentinelle_gsm.observations import ( # noqa: E402
|
||||
ObservationsDB, PagingEvent, Sighting,
|
||||
)
|
||||
from sentinelle_gsm.observer import Anonymizer # noqa: E402
|
||||
from sentinelle_gsm.scoring_engine import ScoringEngine # noqa: E402
|
||||
from sentinelle_gsm.trusted import TrustedRegistry # noqa: E402
|
||||
|
||||
_log = logging.getLogger("secubox.sentinelle-gsm.api")
|
||||
|
|
@ -151,40 +158,191 @@ def get_obs_db() -> ObservationsDB:
|
|||
@app.on_event("startup")
|
||||
def _init_v0_3_singletons() -> None:
|
||||
global _livemon, _listener, _obs_db
|
||||
global _l3, _scoring, _baseline
|
||||
if _livemon is None:
|
||||
_livemon = LivemonRunner()
|
||||
if _listener is None:
|
||||
_listener = GsmtapListener()
|
||||
if _obs_db is None:
|
||||
_obs_db = ObservationsDB(OBSERVATIONS_DB_PATH)
|
||||
# v0.3.1: L3 decoder + baseline + scoring engine
|
||||
if _baseline is None:
|
||||
_baseline = CellBaseline(_obs_db._db)
|
||||
if _l3 is None:
|
||||
_l3 = L3Decode(_get_anonymizer())
|
||||
if _scoring is None:
|
||||
_scoring = ScoringEngine(_baseline)
|
||||
|
||||
|
||||
# ── v0.3.1: L3 decode + scoring + baseline singletons ───────────────────
|
||||
#
|
||||
# The consume loop pipes Observation.raw_l3 through L3Decode → updates
|
||||
# the operator baseline → runs the 8-heuristic ScoringEngine → matches
|
||||
# paged subscriber hashes against the trusted registry → emits an
|
||||
# Alert via AlertSink when score >= ALERT_THRESHOLD.
|
||||
|
||||
_l3: Optional[L3Decode] = None
|
||||
_scoring: Optional[ScoringEngine] = None
|
||||
_baseline: Optional[CellBaseline] = None
|
||||
|
||||
ALERT_THRESHOLD = int(os.environ.get("SENTINELLE_ALERT_THRESHOLD", "60"))
|
||||
|
||||
|
||||
def get_l3() -> L3Decode:
|
||||
if _l3 is None:
|
||||
raise RuntimeError("l3 decoder not initialised")
|
||||
return _l3
|
||||
|
||||
|
||||
def get_scoring() -> ScoringEngine:
|
||||
if _scoring is None:
|
||||
raise RuntimeError("scoring engine not initialised")
|
||||
return _scoring
|
||||
|
||||
|
||||
def get_baseline() -> CellBaseline:
|
||||
if _baseline is None:
|
||||
raise RuntimeError("baseline not initialised")
|
||||
return _baseline
|
||||
|
||||
|
||||
# Map paging request L3 message type → human label for paging_events.
|
||||
_PAGING_REQUEST_LABEL = {
|
||||
0x21: "paging-1",
|
||||
0x22: "paging-2",
|
||||
0x24: "paging-3",
|
||||
}
|
||||
|
||||
|
||||
def _cell_id_from_parsed(parsed_cell_info, obs) -> str:
|
||||
"""Build a cell_id from L3-parsed cell_info if complete, else fall back
|
||||
to the arfcn/channel composite key. Always returns a non-empty string."""
|
||||
ci = parsed_cell_info
|
||||
if (ci is not None and ci.mcc is not None and ci.mnc is not None
|
||||
and ci.lac is not None and ci.ci is not None):
|
||||
return f"{ci.mcc}-{ci.mnc}-{ci.lac}-{ci.ci}"
|
||||
return obs.cell_id or f"arfcn-{obs.arfcn}-ch-{obs.channel}"
|
||||
|
||||
|
||||
async def _process_observation(obs) -> None:
|
||||
"""Handle ONE GSMTAP Observation: L3-decode, persist, score, alert.
|
||||
|
||||
Extracted from _consume_observations() for testability. Tests can
|
||||
feed synthesized Observation instances directly without driving the
|
||||
UDP listener / consume coroutine.
|
||||
|
||||
Step-by-step (per v0.3.1 plan §6.3):
|
||||
1. L3-decode obs.raw_l3 → ParsedFrame(cell_info?, paging?)
|
||||
2. Derive cell_id from parsed cell_info OR fallback arfcn-ch key
|
||||
3. upsert_sighting() with whatever metadata we have
|
||||
4. record_paging() for every paged subscriber (already hashed)
|
||||
5. baseline.consider() so the cell graduates after N sightings
|
||||
6. scoring.evaluate() → CellScore
|
||||
7. score >= threshold → match each paged hash against trusted
|
||||
registry; emit a labeled-or-anomaly-only Alert.
|
||||
"""
|
||||
l3 = get_l3()
|
||||
db = get_obs_db()
|
||||
baseline = get_baseline()
|
||||
scoring = get_scoring()
|
||||
sink = get_alert_sink()
|
||||
trusted = get_trusted_registry()
|
||||
|
||||
parsed = l3.parse(obs.raw_l3 or b"")
|
||||
cell_info = parsed.cell_info
|
||||
paging = parsed.paging
|
||||
|
||||
cell_id = _cell_id_from_parsed(cell_info, obs)
|
||||
|
||||
# Prefer parsed (BCCH-derived) metadata over header-only fields.
|
||||
mcc = (cell_info.mcc if cell_info and cell_info.mcc is not None
|
||||
else obs.mcc)
|
||||
mnc = (cell_info.mnc if cell_info and cell_info.mnc is not None
|
||||
else obs.mnc)
|
||||
lac = (cell_info.lac if cell_info and cell_info.lac is not None
|
||||
else obs.lac)
|
||||
ci_val = (cell_info.ci if cell_info and cell_info.ci is not None
|
||||
else obs.ci)
|
||||
cipher_a5 = cell_info.a5_advertised if cell_info else None
|
||||
|
||||
try:
|
||||
db.upsert_sighting(Sighting(
|
||||
cell_id=cell_id, arfcn=obs.arfcn,
|
||||
mcc=mcc, mnc=mnc, lac=lac, ci=ci_val,
|
||||
))
|
||||
except ValueError as exc:
|
||||
_log.warning("observations: refused write: %s", exc)
|
||||
return
|
||||
|
||||
if paging is not None:
|
||||
label = _PAGING_REQUEST_LABEL.get(paging.paging_type, "paging")
|
||||
for pid in paging.identities:
|
||||
try:
|
||||
db.record_paging(PagingEvent(
|
||||
ts=obs.ts, cell_id=cell_id,
|
||||
subscriber_hash=pid.subscriber_hash,
|
||||
request_type=label,
|
||||
))
|
||||
except ValueError as exc:
|
||||
_log.warning("paging_events: refused write: %s", exc)
|
||||
|
||||
baseline.consider(
|
||||
cell_id, mcc=mcc, mnc=mnc, lac=lac,
|
||||
arfcn=obs.arfcn, cipher_a5=cipher_a5,
|
||||
)
|
||||
|
||||
result = scoring.evaluate(cell_info, paging, obs.arfcn)
|
||||
if result.score < ALERT_THRESHOLD:
|
||||
return
|
||||
|
||||
reason = "; ".join(result.reasons) if result.reasons else "anomaly"
|
||||
|
||||
# Try to match each paged subscriber against trusted registry.
|
||||
matched: list[tuple[str, str]] = [] # (hash, label)
|
||||
if paging is not None:
|
||||
for pid in paging.identities:
|
||||
tp = trusted.lookup_by_hash(pid.subscriber_hash)
|
||||
if tp is not None:
|
||||
matched.append((pid.subscriber_hash, tp.label))
|
||||
|
||||
if matched:
|
||||
# One alert per matched trusted phone — operator wants per-device
|
||||
# labeled visibility.
|
||||
for sub_hash, label in matched:
|
||||
try:
|
||||
sink.write(Alert(
|
||||
cell_id=cell_id, arfcn=obs.arfcn,
|
||||
score=result.score, reason=reason,
|
||||
subscriber_hash=sub_hash, trusted_label=label,
|
||||
))
|
||||
except ValueError as exc:
|
||||
_log.warning("alert: refused write: %s", exc)
|
||||
else:
|
||||
# Anomaly-only — score crossed threshold but no trusted match.
|
||||
try:
|
||||
sink.write(Alert(
|
||||
cell_id=cell_id, arfcn=obs.arfcn,
|
||||
score=result.score, reason=reason,
|
||||
subscriber_hash=None, trusted_label=None,
|
||||
))
|
||||
except ValueError as exc:
|
||||
_log.warning("alert: refused write: %s", exc)
|
||||
|
||||
|
||||
async def _consume_observations() -> None:
|
||||
"""Drain the GSMTAP listener and upsert sightings.
|
||||
"""Drain the GSMTAP listener and process each observation.
|
||||
|
||||
Cancellation-safe: on /scan/stop we cancel this task before
|
||||
stopping the listener + runner so no leaked subprocess is left
|
||||
behind. Until v0.3.1 wires the L3 BCCH decode we don't know the
|
||||
operator quadruple (MCC/MNC/LAC/CI), so we synthesise a placeholder
|
||||
cell_id from (arfcn, channel). The schema accepts NULL for the
|
||||
operator fields.
|
||||
Cancellation-safe: /scan/stop cancels this task first, then stops
|
||||
the listener + runner. Per-observation errors are logged and the
|
||||
loop continues — a single malformed frame must NOT kill the scan.
|
||||
"""
|
||||
listener = get_listener()
|
||||
db = get_obs_db()
|
||||
try:
|
||||
async for obs in listener.observations():
|
||||
cell_id = obs.cell_id or f"arfcn-{obs.arfcn}-ch-{obs.channel}"
|
||||
try:
|
||||
db.upsert_sighting(Sighting(
|
||||
cell_id=cell_id,
|
||||
arfcn=obs.arfcn,
|
||||
mcc=obs.mcc,
|
||||
mnc=obs.mnc,
|
||||
lac=obs.lac,
|
||||
ci=obs.ci,
|
||||
))
|
||||
except ValueError as exc:
|
||||
_log.warning("observations: refused write: %s", exc)
|
||||
await _process_observation(obs)
|
||||
except Exception: # noqa: BLE001
|
||||
_log.exception("process_observation failed; continuing")
|
||||
except asyncio.CancelledError:
|
||||
# Normal path on /scan/stop — propagate so the task transitions
|
||||
# to CANCELLED rather than swallowing.
|
||||
|
|
@ -572,6 +730,59 @@ async def list_observations(limit: int = 200) -> dict:
|
|||
}
|
||||
|
||||
|
||||
# ── v0.3.1: baseline + scoring endpoints ────────────────────────────────
|
||||
#
|
||||
# /baseline — read graduated cells
|
||||
# /baseline/learn — enable learn-mode for a sweep window
|
||||
# /scoring/thresholds — read / write the 8-heuristic threshold table
|
||||
#
|
||||
# JWT enforcement is delegated to nginx + Authelia upstream (require_jwt
|
||||
# is a no-op hook here, overridable in tests).
|
||||
|
||||
class BaselineLearnBody(BaseModel):
|
||||
sweep_seconds: int = 300
|
||||
|
||||
|
||||
@app.get("/baseline", dependencies=[Depends(require_jwt)])
|
||||
def list_baseline(limit: int = 200) -> dict:
|
||||
"""Operator baseline — list of cells graduated to baseline."""
|
||||
limit = max(1, min(1000, limit))
|
||||
return {"cells": [asdict(c) for c in get_baseline().list(limit=limit)]}
|
||||
|
||||
|
||||
@app.post("/baseline/learn", dependencies=[Depends(require_jwt)])
|
||||
def baseline_learn(body: BaselineLearnBody) -> dict:
|
||||
"""Flip the baseline into learn-mode for `sweep_seconds`.
|
||||
|
||||
Cells observed during that window graduate to baseline immediately
|
||||
(initial learn_count = LEARN_THRESHOLD) instead of waiting for the
|
||||
default N=3 sightings. The operator must start a scan separately.
|
||||
"""
|
||||
bl = get_baseline()
|
||||
bl.set_learn_mode(body.sweep_seconds)
|
||||
_log.info("baseline learn mode enabled for %ds", body.sweep_seconds)
|
||||
import time
|
||||
return {"learn_mode_until": time.time() + body.sweep_seconds}
|
||||
|
||||
|
||||
@app.get("/scoring/thresholds", dependencies=[Depends(require_jwt)])
|
||||
def scoring_thresholds_get() -> dict:
|
||||
return {"thresholds": get_scoring().thresholds()}
|
||||
|
||||
|
||||
@app.post("/scoring/thresholds", dependencies=[Depends(require_jwt)])
|
||||
def scoring_thresholds_set(body: dict = Body(...)) -> dict:
|
||||
"""Per-heuristic deep-merge of `body` into the live threshold table.
|
||||
|
||||
Emits an audit log entry (caught by /journal/stream) listing which
|
||||
heuristics were touched — operators need a paper trail when scoring
|
||||
is tuned mid-deployment.
|
||||
"""
|
||||
new = get_scoring().update_thresholds(body)
|
||||
_log.info("scoring thresholds updated: %s", list(body.keys())) # audit
|
||||
return {"thresholds": new}
|
||||
|
||||
|
||||
@app.get("/healthz")
|
||||
def healthz() -> dict:
|
||||
return {"ok": True, "module": "sentinelle-gsm", "version": "0.1.0",
|
||||
|
|
|
|||
283
packages/secubox-sentinelle-gsm/api/tests/test_alert_emission.py
Normal file
283
packages/secubox-sentinelle-gsm/api/tests/test_alert_emission.py
Normal file
|
|
@ -0,0 +1,283 @@
|
|||
# packages/secubox-sentinelle-gsm/api/tests/test_alert_emission.py
|
||||
# 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.
|
||||
|
||||
"""
|
||||
Integration tests for the v0.3.1 consume-loop alert emission flow.
|
||||
|
||||
We drive `_process_observation()` directly with synthesised
|
||||
Observation instances (rather than booting the UDP listener) so the
|
||||
tests cover the full L3-decode → baseline → scoring → trusted-match
|
||||
→ AlertSink path end-to-end, without spawning grgsm_livemon_headless
|
||||
or binding a real UDP socket.
|
||||
|
||||
JWT is bypassed via FastAPI's dependency_overrides — the real JWT
|
||||
layer lives at nginx + Authelia, not inside the app.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from sentinelle_gsm.alert_sink import AlertSink
|
||||
from sentinelle_gsm.baseline import CellBaseline
|
||||
from sentinelle_gsm.gsmtap_listener import Observation
|
||||
from sentinelle_gsm.l3_decode import L3Decode, MID_TYPE_IMSI
|
||||
from sentinelle_gsm.observations import ObservationsDB
|
||||
from sentinelle_gsm.observer import Anonymizer
|
||||
from sentinelle_gsm.scoring_engine import ScoringEngine
|
||||
from sentinelle_gsm.trusted import TrustedRegistry
|
||||
|
||||
|
||||
# ─── Synthetic L3 frame builders ──────────────────────────────────────
|
||||
# Mirror the helpers used in test_l3_decode.py so we can build real
|
||||
# raw_l3 bytes the decoder will actually parse. Keeping these private
|
||||
# to this test file rather than importing from test_l3_decode.py — pytest
|
||||
# resolves the symbol via collection but it makes the dependency explicit.
|
||||
|
||||
|
||||
def _encode_bcd_mcc_mnc(mcc: int, mnc: int) -> bytes:
|
||||
"""3-byte MCC/MNC encoding per TS 24.008 §10.5.1.3 (2-digit MNC)."""
|
||||
m1, m2, m3 = mcc // 100, (mcc // 10) % 10, mcc % 10
|
||||
n1, n2 = mnc // 10, mnc % 10
|
||||
return bytes([
|
||||
(m2 << 4) | m1,
|
||||
(0xF << 4) | m3,
|
||||
(n2 << 4) | n1,
|
||||
])
|
||||
|
||||
|
||||
def _encode_bcd_imsi_body(imsi: str) -> bytes:
|
||||
"""Mobile Identity IMSI BCD body."""
|
||||
odd = (len(imsi) % 2) == 1
|
||||
type_nibble = ((1 if odd else 0) << 3) | MID_TYPE_IMSI
|
||||
body = bytearray()
|
||||
body.append((int(imsi[0]) << 4) | type_nibble)
|
||||
i = 1
|
||||
while i < len(imsi):
|
||||
low = int(imsi[i])
|
||||
high = int(imsi[i + 1]) if (i + 1) < len(imsi) else 0xF
|
||||
body.append((high << 4) | low)
|
||||
i += 2
|
||||
return bytes(body)
|
||||
|
||||
|
||||
def _wrap_mobile_id(body: bytes) -> bytes:
|
||||
return bytes([len(body)]) + body
|
||||
|
||||
|
||||
def _build_si6_frame(mcc: int, mnc: int, lac: int, ci: int, a5: int) -> bytes:
|
||||
"""Build a System Information Type 6 (BCCH) frame with a chosen A5."""
|
||||
cell_options = (a5 << 5) & 0xFF
|
||||
return (
|
||||
b"\x06\x1e" # PD=RR, msg=SI6
|
||||
+ ci.to_bytes(2, "big")
|
||||
+ _encode_bcd_mcc_mnc(mcc, mnc)
|
||||
+ lac.to_bytes(2, "big")
|
||||
+ bytes([cell_options])
|
||||
)
|
||||
|
||||
|
||||
def _build_paging1_frame_with_imsi(imsi: str) -> bytes:
|
||||
"""Build a Paging Request Type 1 frame carrying ONE IMSI identity."""
|
||||
mid_tlv = _wrap_mobile_id(_encode_bcd_imsi_body(imsi))
|
||||
return b"\x06\x21" + b"\x00" + mid_tlv # PD=RR, msg=PR1, page_mode
|
||||
|
||||
|
||||
# ─── Fixture ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def harness(tmp_path):
|
||||
"""Wire ALL v0.3.1 singletons against tmp_path-backed real implementations.
|
||||
|
||||
Returns the api.main module with all singletons live, JWT bypassed,
|
||||
a clean alert sink, and a fresh observations DB / baseline / scoring
|
||||
engine. Caller cleans up automatically on yield exit.
|
||||
"""
|
||||
from api import main as api_main
|
||||
|
||||
anonymizer = Anonymizer(b"x" * 32)
|
||||
api_main._alert_sink = AlertSink(tmp_path / "alerts.db")
|
||||
api_main._trusted_registry = TrustedRegistry(
|
||||
tmp_path / "trusted.json", anonymizer
|
||||
)
|
||||
api_main._obs_db = ObservationsDB(tmp_path / "obs.db")
|
||||
api_main._baseline = CellBaseline(api_main._obs_db._db)
|
||||
api_main._l3 = L3Decode(anonymizer)
|
||||
api_main._scoring = ScoringEngine(api_main._baseline)
|
||||
# Lower the alert threshold? No — default 60 is what the plan asks
|
||||
# us to test against. Keep it at the module-level default.
|
||||
|
||||
api_main.app.dependency_overrides[api_main.require_jwt] = (
|
||||
lambda: {"sub": "tester"}
|
||||
)
|
||||
try:
|
||||
yield api_main, anonymizer
|
||||
finally:
|
||||
api_main.app.dependency_overrides.clear()
|
||||
api_main._alert_sink = None
|
||||
api_main._trusted_registry = None
|
||||
api_main._obs_db = None
|
||||
api_main._baseline = None
|
||||
api_main._l3 = None
|
||||
api_main._scoring = None
|
||||
|
||||
|
||||
def _run(coro):
|
||||
"""Execute a coroutine on a fresh event loop.
|
||||
|
||||
A fresh loop per call dodges the "event loop is closed" gotcha when
|
||||
pytest-asyncio is not configured and TestClient has already torn down
|
||||
its own loop earlier in the test.
|
||||
"""
|
||||
return asyncio.new_event_loop().run_until_complete(coro)
|
||||
|
||||
|
||||
# ─── Tests ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_score_below_threshold_no_alert(harness):
|
||||
"""A bare SI6 frame with no anomalies → score < 60 → NO alert written."""
|
||||
api_main, _ = harness
|
||||
# Graduate the cell to baseline first so ghost_bts doesn't fire.
|
||||
api_main._baseline.consider("208-1-100-12345", mcc=208, mnc=1,
|
||||
lac=100, arfcn=42, cipher_a5=3)
|
||||
api_main._baseline.consider("208-1-100-12345", mcc=208, mnc=1,
|
||||
lac=100, arfcn=42, cipher_a5=3)
|
||||
api_main._baseline.consider("208-1-100-12345", mcc=208, mnc=1,
|
||||
lac=100, arfcn=42, cipher_a5=3)
|
||||
|
||||
# SI6 with same A5 → no cipher_downgrade, no ghost_bts (in baseline),
|
||||
# no identity mismatch (same mcc/mnc), no paging → no abuse.
|
||||
raw = _build_si6_frame(mcc=208, mnc=1, lac=100, ci=12345, a5=3)
|
||||
obs = Observation(ts=1.0, arfcn=42, frame_nr=0, channel=0x01,
|
||||
sub_type=0, raw_l3=raw)
|
||||
_run(api_main._process_observation(obs))
|
||||
alerts = api_main._alert_sink.list()
|
||||
assert alerts == [], f"unexpected alerts: {alerts}"
|
||||
|
||||
|
||||
def test_score_above_threshold_no_trusted_match_anomaly_only(harness):
|
||||
"""Score >= 60 + paging present + no trusted match → 1 anomaly-only alert.
|
||||
|
||||
Combined trigger : cipher_downgrade(40) + ghost_bts(35) = 75 ≥ 60.
|
||||
The cell is graduated to baseline with A5/3; we then send a frame
|
||||
advertising A5/1 (downgrade). Paging is on a different cell_id
|
||||
(which is therefore NOT in baseline → ghost_bts).
|
||||
"""
|
||||
api_main, _ = harness
|
||||
# Graduate baseline with A5/3
|
||||
bl_cell = "208-1-100-12345"
|
||||
for _ in range(3):
|
||||
api_main._baseline.consider(bl_cell, mcc=208, mnc=1, lac=100,
|
||||
arfcn=42, cipher_a5=3)
|
||||
assert api_main._baseline.is_baseline(bl_cell)
|
||||
|
||||
# Observed SI6 = same cell_id but A5/1 → cipher_downgrade(40)
|
||||
# alone is below threshold; verify no premature alert.
|
||||
raw_si6 = _build_si6_frame(mcc=208, mnc=1, lac=100, ci=12345, a5=1)
|
||||
obs_si = Observation(ts=2.0, arfcn=42, frame_nr=0, channel=0x01,
|
||||
sub_type=0, raw_l3=raw_si6)
|
||||
_run(api_main._process_observation(obs_si))
|
||||
|
||||
# Fire a paging frame on a brand-new (un-baselined) cell_id —
|
||||
# ghost_bts contributes 35. Paging-only frames don't carry cell_info,
|
||||
# so the engine falls back to "arfcn-{arfcn}" → not in baseline.
|
||||
raw_pg = _build_paging1_frame_with_imsi("208201234567890")
|
||||
obs_pg = Observation(ts=3.0, arfcn=999, frame_nr=0, channel=0x04,
|
||||
sub_type=0, raw_l3=raw_pg)
|
||||
_run(api_main._process_observation(obs_pg))
|
||||
|
||||
# First two frames each scored < 60 → no alert yet.
|
||||
assert api_main._alert_sink.list() == [], \
|
||||
"single triggers must not cross threshold individually"
|
||||
|
||||
# Re-fire the same paging frame twice more → identity_request_abuse
|
||||
# (3 reqs in the 300s window, strictly > threshold=2) contributes
|
||||
# 35; combined with ghost_bts(35) the cell hits 70 ≥ 60 and an
|
||||
# alert is emitted.
|
||||
_run(api_main._process_observation(obs_pg))
|
||||
_run(api_main._process_observation(obs_pg))
|
||||
alerts = api_main._alert_sink.list()
|
||||
assert len(alerts) >= 1, f"expected ≥1 alert, got {alerts}"
|
||||
a = alerts[0]
|
||||
# No trusted IMSI was registered → anomaly-only emission.
|
||||
assert a.subscriber_hash is None, f"expected anomaly-only: {a}"
|
||||
assert a.trusted_label is None
|
||||
assert a.score >= 60
|
||||
|
||||
|
||||
def test_score_above_threshold_with_trusted_match_emits_labeled_alert(harness):
|
||||
"""Score >= 60 + paging matches trusted phone → labeled alert."""
|
||||
api_main, anonymizer = harness
|
||||
# Register the trusted phone — same IMSI we'll page below.
|
||||
imsi = "208201234567890"
|
||||
registered = api_main._trusted_registry.add(imsi, label="ops-phone-1")
|
||||
expected_hash = registered.imsi_hash
|
||||
|
||||
# Fire 3 paging frames on the same arfcn (ghost_bts cell) to get
|
||||
# identity_request_abuse(35) + ghost_bts(35) = 70.
|
||||
raw_pg = _build_paging1_frame_with_imsi(imsi)
|
||||
for ts in (10.0, 11.0, 12.0):
|
||||
obs = Observation(ts=ts, arfcn=777, frame_nr=0, channel=0x04,
|
||||
sub_type=0, raw_l3=raw_pg)
|
||||
_run(api_main._process_observation(obs))
|
||||
|
||||
alerts = api_main._alert_sink.list()
|
||||
assert len(alerts) >= 1
|
||||
# The labeled alert must surface the trusted match.
|
||||
labeled = [a for a in alerts if a.trusted_label is not None]
|
||||
assert labeled, f"no labeled alert in {alerts}"
|
||||
a = labeled[0]
|
||||
assert a.trusted_label == "ops-phone-1"
|
||||
assert a.subscriber_hash == expected_hash
|
||||
assert a.score >= 60
|
||||
|
||||
|
||||
def test_paging_event_persistence_with_hash(harness):
|
||||
"""Paging events are persisted with the HMAC hash, never plaintext."""
|
||||
api_main, _ = harness
|
||||
imsi = "208201234567890"
|
||||
raw_pg = _build_paging1_frame_with_imsi(imsi)
|
||||
obs = Observation(ts=20.0, arfcn=555, frame_nr=0, channel=0x04,
|
||||
sub_type=0, raw_l3=raw_pg)
|
||||
_run(api_main._process_observation(obs))
|
||||
|
||||
# cell_id is fallback (no cell_info on paging-only frame)
|
||||
cell_id = "arfcn-555-ch-4"
|
||||
events = api_main._obs_db.paging_for_cell(cell_id)
|
||||
assert len(events) == 1
|
||||
e = events[0]
|
||||
# Privacy invariant: the stored hash is NOT the plaintext IMSI.
|
||||
assert e.subscriber_hash != imsi
|
||||
assert imsi not in e.subscriber_hash
|
||||
assert len(e.subscriber_hash) == 16 # ANON_TRUNCATE_HEX
|
||||
assert e.request_type == "paging-1"
|
||||
|
||||
|
||||
def test_post_scoring_thresholds_logs_audit(harness, caplog):
|
||||
"""POST /scoring/thresholds emits an INFO-level audit log entry."""
|
||||
api_main, _ = harness
|
||||
client = TestClient(api_main.app)
|
||||
|
||||
with caplog.at_level(logging.INFO,
|
||||
logger="secubox.sentinelle-gsm.api"):
|
||||
r = client.post(
|
||||
"/scoring/thresholds",
|
||||
json={"cipher_downgrade": {"score": 55}},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
msgs = [rec.message for rec in caplog.records]
|
||||
assert any("scoring thresholds updated" in m for m in msgs), \
|
||||
f"expected audit log line, got {msgs}"
|
||||
# Threshold body was applied
|
||||
assert (
|
||||
r.json()["thresholds"]["cipher_downgrade"]["score"] == 55
|
||||
)
|
||||
89
packages/secubox-sentinelle-gsm/api/tests/test_baseline.py
Normal file
89
packages/secubox-sentinelle-gsm/api/tests/test_baseline.py
Normal file
|
|
@ -0,0 +1,89 @@
|
|||
# packages/secubox-sentinelle-gsm/api/tests/test_baseline.py
|
||||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
|
||||
"""
|
||||
Tests for CellBaseline (operator-baseline learning).
|
||||
|
||||
Schema setup mirrors production: ObservationsDB creates the cell_baseline
|
||||
table (idempotent CREATE IF NOT EXISTS) when it opens the file, then we
|
||||
hand the underlying sqlite3.Connection to CellBaseline. This exercises
|
||||
the colocation invariant (one observations.db, multiple wrappers).
|
||||
"""
|
||||
|
||||
import sqlite3
|
||||
import time
|
||||
|
||||
import pytest
|
||||
|
||||
from sentinelle_gsm.baseline import BaselineCell, CellBaseline
|
||||
from sentinelle_gsm.observations import ObservationsDB
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def conn(tmp_path):
|
||||
# ObservationsDB.__init__ creates the cell_baseline table.
|
||||
db = ObservationsDB(tmp_path / "obs.db")
|
||||
yield db._db
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def baseline(conn):
|
||||
return CellBaseline(conn)
|
||||
|
||||
|
||||
def test_first_consider_starts_at_1_not_baseline_yet(baseline):
|
||||
baseline.consider("208-01-200-12345", mcc=208, mnc=1, lac=200, arfcn=124)
|
||||
cell = baseline.get("208-01-200-12345")
|
||||
assert cell is not None
|
||||
assert cell.learn_count == 1
|
||||
assert baseline.is_baseline("208-01-200-12345") is False
|
||||
|
||||
|
||||
def test_three_considers_graduate_to_baseline(baseline):
|
||||
cid = "208-01-200-12345"
|
||||
for _ in range(3):
|
||||
baseline.consider(cid, mcc=208, mnc=1, lac=200, arfcn=124)
|
||||
cell = baseline.get(cid)
|
||||
assert cell is not None
|
||||
assert cell.learn_count == 3
|
||||
assert baseline.is_baseline(cid) is True
|
||||
|
||||
|
||||
def test_learn_mode_graduates_first_consider(baseline):
|
||||
baseline.set_learn_mode(60)
|
||||
assert baseline.in_learn_mode() is True
|
||||
baseline.consider("208-01-300-99999", mcc=208, mnc=1, lac=300, arfcn=64)
|
||||
cell = baseline.get("208-01-300-99999")
|
||||
assert cell is not None
|
||||
assert cell.learn_count == CellBaseline.LEARN_THRESHOLD
|
||||
assert baseline.is_baseline("208-01-300-99999") is True
|
||||
|
||||
|
||||
def test_consider_updates_metadata_via_coalesce(baseline):
|
||||
cid = "208-01-400-55555"
|
||||
# First sighting carries mcc=208.
|
||||
baseline.consider(cid, mcc=208, mnc=1, lac=400, arfcn=42)
|
||||
# Second sighting carries mcc=None — must NOT clobber the prior 208.
|
||||
baseline.consider(cid, mcc=None, mnc=None, lac=None, arfcn=42)
|
||||
cell = baseline.get(cid)
|
||||
assert cell is not None
|
||||
assert cell.mcc == 208
|
||||
assert cell.mnc == 1
|
||||
assert cell.lac == 400
|
||||
assert cell.learn_count == 2
|
||||
|
||||
|
||||
def test_list_orders_by_last_learned_desc(baseline):
|
||||
baseline.consider("cell-A", arfcn=10)
|
||||
time.sleep(0.01)
|
||||
baseline.consider("cell-B", arfcn=20)
|
||||
time.sleep(0.01)
|
||||
baseline.consider("cell-C", arfcn=30)
|
||||
cells = baseline.list()
|
||||
assert len(cells) == 3
|
||||
assert cells[0].cell_id == "cell-C"
|
||||
assert cells[1].cell_id == "cell-B"
|
||||
assert cells[2].cell_id == "cell-A"
|
||||
# Confirm dataclass type (defensive — catches accidental tuple return).
|
||||
assert isinstance(cells[0], BaselineCell)
|
||||
|
|
@ -55,8 +55,12 @@ async def test_listener_receives_a_datagram(tmp_path):
|
|||
listener = GsmtapListener(host="127.0.0.1", port=47291)
|
||||
await listener.start()
|
||||
try:
|
||||
payload = b"\x06\x1a" + b"DECAFBAD" * 2
|
||||
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
sock.sendto(_build_gsmtap_v2(arfcn=42, channel=0x04), ("127.0.0.1", 47291))
|
||||
sock.sendto(
|
||||
_build_gsmtap_v2(arfcn=42, channel=0x04) + payload,
|
||||
("127.0.0.1", 47291),
|
||||
)
|
||||
|
||||
async def first():
|
||||
async for obs in listener.observations():
|
||||
|
|
@ -64,5 +68,6 @@ async def test_listener_receives_a_datagram(tmp_path):
|
|||
obs = await asyncio.wait_for(first(), timeout=1.0)
|
||||
assert obs.arfcn == 42
|
||||
assert obs.channel == 0x04
|
||||
assert obs.raw_l3 == payload
|
||||
finally:
|
||||
await listener.stop()
|
||||
|
|
|
|||
213
packages/secubox-sentinelle-gsm/api/tests/test_l3_decode.py
Normal file
213
packages/secubox-sentinelle-gsm/api/tests/test_l3_decode.py
Normal file
|
|
@ -0,0 +1,213 @@
|
|||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
|
||||
"""Tests for sentinelle_gsm.l3_decode — BCCH SI + CCCH paging parsing.
|
||||
|
||||
Privacy invariant verified : public dataclasses NEVER carry plaintext
|
||||
IMSI/TMSI bytes — only HMAC-truncated subscriber_hash strings.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from sentinelle_gsm.observer import Anonymizer
|
||||
from sentinelle_gsm.l3_decode import (
|
||||
L3Decode,
|
||||
MID_TYPE_IMSI,
|
||||
MID_TYPE_TMSI,
|
||||
ParsedFrame,
|
||||
_decode_bcd_imsi,
|
||||
_decode_mcc_mnc,
|
||||
)
|
||||
|
||||
|
||||
# ─── BCD encoder helpers (inverse of the decoder; for building synthetic
|
||||
# L3 frames in tests). The decoder is the spec reference; if these
|
||||
# helpers disagree, fix THEM, not the decoder.
|
||||
# ──────────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _encode_bcd_mcc_mnc(mcc: int, mnc: int, three_digit_mnc: bool = False) -> bytes:
|
||||
"""3-byte MCC/MNC encoding per TS 24.008 §10.5.1.3.
|
||||
|
||||
Layout (decoder POV) :
|
||||
byte 0 : low nibble = MCC digit 1, high nibble = MCC digit 2
|
||||
byte 1 : low nibble = MCC digit 3, high nibble = MNC digit 3 OR 0xF
|
||||
byte 2 : low nibble = MNC digit 1, high nibble = MNC digit 2
|
||||
"""
|
||||
m1, m2, m3 = mcc // 100, (mcc // 10) % 10, mcc % 10
|
||||
if not three_digit_mnc and mnc < 100:
|
||||
n1, n2 = mnc // 10, mnc % 10
|
||||
return bytes([
|
||||
(m2 << 4) | m1,
|
||||
(0xF << 4) | m3,
|
||||
(n2 << 4) | n1,
|
||||
])
|
||||
# Decoder names : d_mnc_3 = hundreds, d_mnc_2 = tens, d_mnc_1 = ones.
|
||||
hundreds, tens, ones = mnc // 100, (mnc // 10) % 10, mnc % 10
|
||||
return bytes([
|
||||
(m2 << 4) | m1,
|
||||
(hundreds << 4) | m3,
|
||||
(tens << 4) | ones,
|
||||
])
|
||||
|
||||
|
||||
def _encode_bcd_imsi_body(imsi: str) -> bytes:
|
||||
"""Mobile Identity IMSI BCD per TS 24.008 §10.5.1.4.
|
||||
|
||||
Layout (decoder POV) :
|
||||
body[0] : high nibble = first IMSI digit
|
||||
low nibble = (odd<<3) | MID_TYPE_IMSI (= 0x01)
|
||||
body[k] (k>=1) : low nibble = digit 2k, high nibble = digit 2k+1
|
||||
For even-length IMSI, last high nibble = 0xF (filler).
|
||||
"""
|
||||
odd = (len(imsi) % 2) == 1
|
||||
type_nibble = ((1 if odd else 0) << 3) | MID_TYPE_IMSI
|
||||
body = bytearray()
|
||||
body.append((int(imsi[0]) << 4) | type_nibble)
|
||||
i = 1
|
||||
while i < len(imsi):
|
||||
low = int(imsi[i])
|
||||
high = int(imsi[i + 1]) if (i + 1) < len(imsi) else 0xF
|
||||
body.append((high << 4) | low)
|
||||
i += 2
|
||||
return bytes(body)
|
||||
|
||||
|
||||
def _encode_bcd_tmsi_body(tmsi_bytes: bytes) -> bytes:
|
||||
"""Mobile Identity TMSI per TS 24.008 §10.5.1.4.
|
||||
|
||||
Layout (decoder POV) :
|
||||
body[0] : type byte; low 3 bits = MID_TYPE_TMSI (= 4)
|
||||
body[1..5] : the 4-byte TMSI value
|
||||
"""
|
||||
assert len(tmsi_bytes) == 4
|
||||
type_byte = 0xF0 | MID_TYPE_TMSI # high nibble 0xF per spec, low = type
|
||||
return bytes([type_byte]) + tmsi_bytes
|
||||
|
||||
|
||||
def _wrap_mobile_id(body: bytes) -> bytes:
|
||||
"""TLV wrapper : 1-byte length + body."""
|
||||
return bytes([len(body)]) + body
|
||||
|
||||
|
||||
# ─── Fixtures ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def decoder():
|
||||
return L3Decode(Anonymizer.ephemeral())
|
||||
|
||||
|
||||
# ─── BCD helper tests (spec sanity checks) ────────────────────────────
|
||||
|
||||
|
||||
def test_decode_mcc_mnc_2_digit_mnc():
|
||||
"""Orange FR : MCC=208, MNC=01 — 2-digit MNC, 0xF filler nibble."""
|
||||
buf = _encode_bcd_mcc_mnc(208, 1)
|
||||
mcc, mnc = _decode_mcc_mnc(buf)
|
||||
assert mcc == 208
|
||||
assert mnc == 1
|
||||
|
||||
|
||||
def test_decode_mcc_mnc_3_digit_mnc():
|
||||
"""Verizon US : MCC=311, MNC=480 — 3-digit MNC, no filler."""
|
||||
buf = _encode_bcd_mcc_mnc(311, 480, three_digit_mnc=True)
|
||||
mcc, mnc = _decode_mcc_mnc(buf)
|
||||
assert mcc == 311
|
||||
assert mnc == 480
|
||||
|
||||
|
||||
def test_decode_bcd_imsi_15_digits_odd():
|
||||
"""IMSI 208201234567890 = 15 digits (odd → odd flag set)."""
|
||||
body = _encode_bcd_imsi_body("208201234567890")
|
||||
digits = _decode_bcd_imsi(body)
|
||||
assert digits == "208201234567890"
|
||||
|
||||
|
||||
# ─── End-to-end L3 frame tests ────────────────────────────────────────
|
||||
|
||||
|
||||
def test_parse_si3_extracts_mcc_mnc_lac_ci(decoder):
|
||||
"""SI Type 3 (BCCH) yields a populated CellInfo."""
|
||||
raw = (
|
||||
b"\x06\x1a" # L3 header (PD=RR, msg=SI3)
|
||||
+ (12345).to_bytes(2, "big") # CI = 12345
|
||||
+ _encode_bcd_mcc_mnc(208, 1) # LAI: MCC=208, MNC=01
|
||||
+ (200).to_bytes(2, "big") # LAC = 200
|
||||
+ b"\x00\x00\x00" # control_channel_description
|
||||
+ b"\x00" # cell_options (no A5 bits)
|
||||
+ b"\x00\x00" # cell_selection
|
||||
+ b"\x00\x00\x00" # rach_control
|
||||
)
|
||||
frame = decoder.parse(raw)
|
||||
assert frame.cell_info is not None
|
||||
assert frame.paging is None
|
||||
assert frame.cell_info.mcc == 208
|
||||
assert frame.cell_info.mnc == 1
|
||||
assert frame.cell_info.lac == 200
|
||||
assert frame.cell_info.ci == 12345
|
||||
|
||||
|
||||
def test_parse_si6_extracts_cipher(decoder):
|
||||
"""SI Type 6 advertises the supported A5 algorithm in cell_options."""
|
||||
# cell_options byte : A5 field = bits 5..7 → 0b011 << 5 = 0x60 → A5/3
|
||||
cell_options = (0x03 << 5) & 0xFF
|
||||
raw = (
|
||||
b"\x06\x1e" # L3 header (PD=RR, msg=SI6)
|
||||
+ (42).to_bytes(2, "big") # CI
|
||||
+ _encode_bcd_mcc_mnc(208, 1) # LAI
|
||||
+ (123).to_bytes(2, "big") # LAC
|
||||
+ bytes([cell_options]) # cipher byte
|
||||
)
|
||||
frame = decoder.parse(raw)
|
||||
assert frame.cell_info is not None
|
||||
assert frame.cell_info.a5_advertised == 3
|
||||
assert frame.cell_info.mcc == 208
|
||||
assert frame.cell_info.lac == 123
|
||||
|
||||
|
||||
def test_parse_paging_request_1_with_tmsi(decoder):
|
||||
"""Paging Request Type 1 with a single TMSI → HMAC hash, NOT plaintext."""
|
||||
tmsi = b"\xDE\xAD\xBE\xEF"
|
||||
mid_tlv = _wrap_mobile_id(_encode_bcd_tmsi_body(tmsi))
|
||||
raw = b"\x06\x21" + b"\x00" + mid_tlv # L3 header + page_mode byte + MID
|
||||
frame = decoder.parse(raw)
|
||||
assert frame.paging is not None
|
||||
assert frame.cell_info is None
|
||||
assert len(frame.paging.identities) == 1
|
||||
pid = frame.paging.identities[0]
|
||||
assert pid.id_type == MID_TYPE_TMSI
|
||||
# Privacy invariant : hash is NOT the plaintext TMSI in any encoding
|
||||
assert pid.subscriber_hash != "deadbeef"
|
||||
assert pid.subscriber_hash != tmsi.hex()
|
||||
assert len(pid.subscriber_hash) == 16 # ANON_TRUNCATE_HEX
|
||||
|
||||
|
||||
def test_parse_paging_request_with_imsi(decoder):
|
||||
"""Paging Request with an IMSI identity → still hashed, never plaintext."""
|
||||
imsi = "208201234567890"
|
||||
mid_tlv = _wrap_mobile_id(_encode_bcd_imsi_body(imsi))
|
||||
raw = b"\x06\x21" + b"\x00" + mid_tlv
|
||||
frame = decoder.parse(raw)
|
||||
assert frame.paging is not None
|
||||
assert len(frame.paging.identities) == 1
|
||||
pid = frame.paging.identities[0]
|
||||
assert pid.id_type == MID_TYPE_IMSI
|
||||
# Privacy invariant : the plaintext IMSI never appears in the hash
|
||||
assert imsi not in pid.subscriber_hash
|
||||
assert len(pid.subscriber_hash) == 16
|
||||
|
||||
|
||||
def test_unknown_msg_type_returns_empty_frame(decoder):
|
||||
frame = decoder.parse(b"\x06\xFF\x00\x00")
|
||||
assert frame.cell_info is None
|
||||
assert frame.paging is None
|
||||
|
||||
|
||||
def test_truncated_input_returns_empty(decoder):
|
||||
"""Permissive parsing : truncated / empty input must NOT raise."""
|
||||
empty = ParsedFrame()
|
||||
assert decoder.parse(b"") == empty
|
||||
assert decoder.parse(b"\x06") == empty
|
||||
# Wrong protocol discriminator
|
||||
assert decoder.parse(b"\x08\x1a\x00\x00") == empty
|
||||
256
packages/secubox-sentinelle-gsm/api/tests/test_scoring_engine.py
Normal file
256
packages/secubox-sentinelle-gsm/api/tests/test_scoring_engine.py
Normal file
|
|
@ -0,0 +1,256 @@
|
|||
# packages/secubox-sentinelle-gsm/api/tests/test_scoring_engine.py
|
||||
# SPDX-License-Identifier: LicenseRef-CMSD-1.0
|
||||
# Copyright (c) 2026 CyberMind — Gerald Kerma <devel@cybermind.fr>
|
||||
|
||||
"""
|
||||
Tests for ScoringEngine — the 8-heuristic detector.
|
||||
|
||||
Each heuristic gets one focused test; aggregation + threshold merge get
|
||||
one each. The CellBaseline is backed by a real sqlite (via ObservationsDB)
|
||||
so we exercise the same code path the consume loop hits in production.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from sentinelle_gsm.baseline import CellBaseline
|
||||
from sentinelle_gsm.l3_decode import (
|
||||
MID_TYPE_TMSI,
|
||||
CellInfo,
|
||||
PagedIdentity,
|
||||
ParsedPagingRequest,
|
||||
)
|
||||
from sentinelle_gsm.observations import ObservationsDB
|
||||
from sentinelle_gsm.scoring_engine import (
|
||||
CellScore,
|
||||
HeuristicResult,
|
||||
ScoringEngine,
|
||||
)
|
||||
|
||||
|
||||
def _graduate(baseline: CellBaseline, cell_id: str, **kwargs) -> None:
|
||||
"""Hit consider() enough times to graduate the cell to baseline."""
|
||||
for _ in range(CellBaseline.LEARN_THRESHOLD):
|
||||
baseline.consider(cell_id, **kwargs)
|
||||
assert baseline.is_baseline(cell_id) is True
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def scoring_engine_factory(tmp_path):
|
||||
"""Return a factory that yields (engine, baseline). Fresh DB per call."""
|
||||
created = []
|
||||
|
||||
def _make(thresholds=None):
|
||||
db_path = tmp_path / f"obs-{len(created)}.db"
|
||||
obs = ObservationsDB(db_path)
|
||||
baseline = CellBaseline(obs._db)
|
||||
engine = ScoringEngine(baseline, thresholds=thresholds)
|
||||
created.append((obs, baseline, engine))
|
||||
return engine, baseline
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
# ── 1. cipher_downgrade ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_cipher_downgrade(scoring_engine_factory):
|
||||
"""Baseline has A5/3 for cell. Observed A5/1 → triggered."""
|
||||
engine, baseline = scoring_engine_factory()
|
||||
cell_id = "208-1-100-12345"
|
||||
_graduate(baseline, cell_id, mcc=208, mnc=1, lac=100, arfcn=42, cipher_a5=3)
|
||||
obs = CellInfo(mcc=208, mnc=1, lac=100, ci=12345, a5_advertised=1)
|
||||
score = engine.evaluate(obs, None, raw_arfcn=42)
|
||||
assert "cipher_downgrade" in score.triggered
|
||||
assert score.score >= 40
|
||||
assert any("A5/1" in r and "A5/3" in r for r in score.reasons)
|
||||
|
||||
|
||||
# ── 2. ghost_bts ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_ghost_bts(scoring_engine_factory):
|
||||
"""Paging on a cell never seen in baseline → triggered."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
paging = ParsedPagingRequest(
|
||||
paging_type=0x21,
|
||||
identities=[PagedIdentity(id_type=MID_TYPE_TMSI, subscriber_hash="abc12345deadbeef")],
|
||||
)
|
||||
# cell_info present but with values that resolve to a brand-new cell_id
|
||||
info = CellInfo(mcc=208, mnc=1, lac=999, ci=77777)
|
||||
score = engine.evaluate(info, paging, raw_arfcn=42)
|
||||
assert "ghost_bts" in score.triggered
|
||||
assert score.score >= 35
|
||||
|
||||
|
||||
# ── 3. identity_mismatch ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_identity_mismatch(scoring_engine_factory):
|
||||
"""Baseline says 208-1; observed says 208-20 on same cell_id → triggered."""
|
||||
engine, baseline = scoring_engine_factory()
|
||||
cell_id = "208-1-200-23456"
|
||||
_graduate(baseline, cell_id, mcc=208, mnc=1, lac=200, arfcn=50, cipher_a5=3)
|
||||
# Same cell_id, but advertise a different MNC — IMSI-catcher impersonation.
|
||||
obs = CellInfo(mcc=208, mnc=20, lac=200, ci=23456, a5_advertised=3)
|
||||
# cell_id from CellInfo would now be "208-20-200-23456", which doesn't
|
||||
# match baseline. To exercise identity_mismatch specifically, override
|
||||
# by graduating BOTH (matching observed cell_id) with the baseline mcc/mnc.
|
||||
# Simpler: register baseline against the observed cell_id with conflicting (mcc, mnc).
|
||||
cell_id_observed = "208-20-200-23456"
|
||||
_graduate(baseline, cell_id_observed, mcc=208, mnc=1, lac=200, arfcn=50, cipher_a5=3)
|
||||
score = engine.evaluate(obs, None, raw_arfcn=50)
|
||||
assert "identity_mismatch" in score.triggered
|
||||
assert score.score >= 30
|
||||
|
||||
|
||||
# ── 4. relocalization_storm ────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_relocalization_storm(scoring_engine_factory):
|
||||
"""> 3 distinct LACs for same cell_id within window → triggered."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
# Same (mcc, mnc, ci) but four different LACs in quick succession.
|
||||
last_score: CellScore | None = None
|
||||
for lac in (100, 200, 300, 400):
|
||||
info = CellInfo(mcc=208, mnc=1, lac=lac, ci=42)
|
||||
last_score = engine.evaluate(info, None, raw_arfcn=10)
|
||||
assert last_score is not None
|
||||
# Each LAC produces a different cell_id; relocalization_storm tracks per
|
||||
# cell_id, so we need to push multiple LACs against the SAME cell_id.
|
||||
# Reset and do it correctly: keep cell_id stable by feeding through the
|
||||
# paging-only fallback ("arfcn-10") with varying LAC values that still
|
||||
# produce the same cell_id key. The cell_id_from_info path only triggers
|
||||
# when all four (mcc, mnc, lac, ci) are present, so a fixed ARFCN with
|
||||
# NO complete CellInfo will share the fallback key.
|
||||
engine2, _ = scoring_engine_factory()
|
||||
for lac in (100, 200, 300, 400):
|
||||
# No ci → cell_id falls back to "arfcn-10"
|
||||
info = CellInfo(mcc=208, mnc=1, lac=lac)
|
||||
result = engine2.evaluate(info, None, raw_arfcn=10)
|
||||
assert "relocalization_storm" in result.triggered
|
||||
assert result.score >= 25
|
||||
|
||||
|
||||
# ── 5. identity_request_abuse ──────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_identity_request_abuse(scoring_engine_factory):
|
||||
"""> 2 paging requests for same subscriber_hash within 300s → triggered."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
h = "deadbeefcafef00d"
|
||||
|
||||
def _paging():
|
||||
return ParsedPagingRequest(
|
||||
paging_type=0x21,
|
||||
identities=[PagedIdentity(id_type=MID_TYPE_TMSI, subscriber_hash=h)],
|
||||
)
|
||||
|
||||
info = CellInfo(mcc=208, mnc=1, lac=100, ci=42)
|
||||
engine.evaluate(info, _paging(), raw_arfcn=10)
|
||||
engine.evaluate(info, _paging(), raw_arfcn=10)
|
||||
score = engine.evaluate(info, _paging(), raw_arfcn=10) # 3rd → > threshold 2
|
||||
assert "identity_request_abuse" in score.triggered
|
||||
assert score.score >= 35
|
||||
# subscriber_hash truncated in reason — no plaintext leakage anyway.
|
||||
assert any(h[:8] in r for r in score.reasons)
|
||||
|
||||
|
||||
# ── 6. anomalous_neighbours ────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_anomalous_neighbours(scoring_engine_factory):
|
||||
"""Previously announced neighbours, now empty set → triggered."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
info = CellInfo(mcc=208, mnc=1, lac=100, ci=42)
|
||||
# First seed the running neighbour baseline.
|
||||
engine.evaluate(info, None, raw_arfcn=10, raw_neighbours={50, 60, 70})
|
||||
# Now the cell announces nothing — drop is suspicious.
|
||||
score = engine.evaluate(info, None, raw_arfcn=10, raw_neighbours=set())
|
||||
assert "anomalous_neighbours" in score.triggered
|
||||
assert score.score >= 15
|
||||
|
||||
|
||||
# ── 7. t3212_out_of_band ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_t3212_out_of_band(scoring_engine_factory):
|
||||
"""T3212 = 120 min (above max=60) → triggered."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
info = CellInfo(mcc=208, mnc=1, lac=100, ci=42, t3212_minutes=120)
|
||||
score = engine.evaluate(info, None, raw_arfcn=10)
|
||||
assert "t3212_out_of_band" in score.triggered
|
||||
assert score.score >= 15
|
||||
assert any("120" in r for r in score.reasons)
|
||||
|
||||
|
||||
# ── 8. orphan_arfcn ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_orphan_arfcn(scoring_engine_factory):
|
||||
"""Orange (208-1) announces ARFCN 800 (outside 1-74) → triggered."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
info = CellInfo(mcc=208, mnc=1, lac=100, ci=42)
|
||||
score = engine.evaluate(info, None, raw_arfcn=800)
|
||||
assert "orphan_arfcn" in score.triggered
|
||||
assert score.score >= 20
|
||||
|
||||
|
||||
def test_orphan_arfcn_unknown_carrier_permissive(scoring_engine_factory):
|
||||
"""Unknown MCC/MNC pair → NOT flagged (permissive)."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
info = CellInfo(mcc=999, mnc=99, lac=100, ci=42)
|
||||
score = engine.evaluate(info, None, raw_arfcn=4242)
|
||||
assert "orphan_arfcn" not in score.triggered
|
||||
|
||||
|
||||
# ── aggregation ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_evaluate_aggregates_and_clamps(scoring_engine_factory):
|
||||
"""Trigger several heuristics at once; sum is clamped to 100."""
|
||||
engine, baseline = scoring_engine_factory()
|
||||
cell_id = "208-1-100-12345"
|
||||
_graduate(baseline, cell_id, mcc=208, mnc=1, lac=100, arfcn=800, cipher_a5=3)
|
||||
# Re-graduate with a DIFFERENT cell_id that matches the observation,
|
||||
# carrying the conflicting baseline (mcc, mnc) so identity_mismatch fires.
|
||||
obs_cell_id = "208-20-100-12345"
|
||||
_graduate(baseline, obs_cell_id, mcc=208, mnc=1, lac=100, arfcn=800, cipher_a5=3)
|
||||
# Observation: A5 downgrade + MNC mismatch + orphan ARFCN + bad T3212
|
||||
info = CellInfo(mcc=208, mnc=20, lac=100, ci=12345,
|
||||
a5_advertised=1, t3212_minutes=999)
|
||||
score = engine.evaluate(info, None, raw_arfcn=800)
|
||||
# cipher_downgrade(40) + identity_mismatch(30) + orphan_arfcn(20)
|
||||
# + t3212_out_of_band(15) = 105 → clamped to 100.
|
||||
assert score.score == 100
|
||||
assert "cipher_downgrade" in score.triggered
|
||||
assert "identity_mismatch" in score.triggered
|
||||
assert "orphan_arfcn" in score.triggered
|
||||
assert "t3212_out_of_band" in score.triggered
|
||||
assert len(score.reasons) == len(score.triggered)
|
||||
|
||||
|
||||
# ── threshold merging ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_update_thresholds_deep_merge(scoring_engine_factory):
|
||||
"""update_thresholds() merges per-heuristic; untouched fields stay."""
|
||||
engine, _baseline = scoring_engine_factory()
|
||||
# Override only the score for cipher_downgrade; "enabled" stays True.
|
||||
effective = engine.update_thresholds({"cipher_downgrade": {"score": 55}})
|
||||
assert effective["cipher_downgrade"]["score"] == 55
|
||||
assert effective["cipher_downgrade"]["enabled"] is True
|
||||
# Other heuristics untouched.
|
||||
assert effective["ghost_bts"]["score"] == 35
|
||||
# Disable orphan_arfcn entirely.
|
||||
effective2 = engine.update_thresholds({"orphan_arfcn": {"enabled": False}})
|
||||
assert effective2["orphan_arfcn"]["enabled"] is False
|
||||
assert effective2["orphan_arfcn"]["score"] == 20 # untouched
|
||||
# Effect on evaluate(): orphan_arfcn no longer fires even on a bad ARFCN.
|
||||
_, baseline = scoring_engine_factory()
|
||||
engine2 = ScoringEngine(baseline)
|
||||
engine2.update_thresholds({"orphan_arfcn": {"enabled": False}})
|
||||
info = CellInfo(mcc=208, mnc=1, lac=100, ci=42)
|
||||
score = engine2.evaluate(info, None, raw_arfcn=800)
|
||||
assert "orphan_arfcn" not in score.triggered
|
||||
|
|
@ -1,3 +1,36 @@
|
|||
secubox-sentinelle-gsm (0.3.1-1~bookworm1) bookworm; urgency=medium
|
||||
|
||||
* lib/sentinelle_gsm/l3_decode.py: NEW — pure-Python TLV parser for
|
||||
BCCH SI Type 3/4/6 (cell metadata) and CCCH Paging Request Type
|
||||
1/2/3 (paged subscriber identities). All paged identities are
|
||||
HMAC-hashed via Anonymizer BEFORE leaving the module; public API
|
||||
surface NEVER returns plaintext IMSI/TMSI/IMEI.
|
||||
* lib/sentinelle_gsm/baseline.py: NEW — operator-baseline learning.
|
||||
Cells graduate to baseline after LEARN_THRESHOLD=3 sightings.
|
||||
Explicit "learn mode" (POST /baseline/learn) immediately accepts
|
||||
every observed cell as baseline within a sweep window.
|
||||
* lib/sentinelle_gsm/scoring_engine.py: NEW — 8 heuristics from spec
|
||||
§6.1 with default thresholds. Sliding windows for relocalization
|
||||
storm + identity_request_abuse (in-memory deques). Per-heuristic
|
||||
enable + threshold are runtime-mutable via POST /scoring/thresholds
|
||||
(audit-logged to the journal stream).
|
||||
* lib/sentinelle_gsm/scoring.py: DROPPED — v0.1 shape-only stub
|
||||
replaced by scoring_engine.py.
|
||||
* api/main.py: consume loop now decodes L3 of every Observation,
|
||||
updates sighting metadata (mcc/mnc/lac/ci/cipher), persists
|
||||
paging events with hashed subscriber IDs, feeds baseline,
|
||||
runs the scoring engine, and emits Alert with trusted_label
|
||||
when a paged hash matches the trusted_phones registry.
|
||||
* api/main.py: new endpoints GET /baseline, POST /baseline/learn,
|
||||
GET /scoring/thresholds, POST /scoring/thresholds.
|
||||
* www/sentinelle/: new Baseline + Scoring panels; alert rows
|
||||
render the trusted_label chip when present.
|
||||
* tests: 34 new (l3_decode ×9, baseline ×5, scoring_engine ×11,
|
||||
alert_emission ×5, privacy ×4); full sweep 82/82 passing.
|
||||
* Closes #349. Refs #237.
|
||||
|
||||
-- Gerald Kerma <devel@cybermind.fr> Fri, 22 May 2026 16:30:00 +0000
|
||||
|
||||
secubox-sentinelle-gsm (0.3.0-1~bookworm1) bookworm; urgency=medium
|
||||
|
||||
* lib/sentinelle_gsm/livemon_runner.py: NEW — manages a single
|
||||
|
|
|
|||
113
packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/baseline.py
Normal file
113
packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/baseline.py
Normal file
|
|
@ -0,0 +1,113 @@
|
|||
# packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/baseline.py
|
||||
# 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.
|
||||
|
||||
"""
|
||||
Operator baseline — list of cells the operator's carrier(s) legitimately
|
||||
operate in this RF environment. Cells are graduated to baseline after
|
||||
being observed N>=3 times; or in 'learn mode' every cell observed within
|
||||
a sweep window is marked baseline regardless of count.
|
||||
|
||||
Used by the scoring engine for ghost_bts + identity_mismatch +
|
||||
cipher_downgrade heuristics.
|
||||
|
||||
Privacy invariant: BaselineCell has NO subscriber-id field. The
|
||||
cell_baseline table colocated in observations.db likewise has no
|
||||
paged-identity column. Enforced by tests/test_privacy_invariant.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import sqlite3
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class BaselineCell:
|
||||
cell_id: str
|
||||
mcc: Optional[int] = None
|
||||
mnc: Optional[int] = None
|
||||
lac: Optional[int] = None
|
||||
arfcn: Optional[int] = None
|
||||
learn_count: int = 1
|
||||
first_learned: float = 0.0
|
||||
last_learned: float = 0.0
|
||||
cipher_a5: Optional[int] = None
|
||||
|
||||
|
||||
class CellBaseline:
|
||||
"""Thin wrapper over the cell_baseline table colocated in observations.db.
|
||||
|
||||
Takes a raw sqlite3.Connection (NOT an ObservationsDB) so it can share
|
||||
the same db file without coupling to observations.py's public interface.
|
||||
"""
|
||||
|
||||
LEARN_THRESHOLD = 3 # default — cells need >=3 sightings to graduate
|
||||
|
||||
def __init__(self, db: sqlite3.Connection):
|
||||
self._db = db
|
||||
self._learn_mode_until: float = 0.0 # epoch; learn_mode while now() < this
|
||||
|
||||
def set_learn_mode(self, seconds: float) -> None:
|
||||
"""Enable explicit-learn-mode for the next `seconds` seconds.
|
||||
Every cell observed in that window graduates to baseline on first
|
||||
sighting (initial learn_count = LEARN_THRESHOLD)."""
|
||||
self._learn_mode_until = time.time() + seconds
|
||||
|
||||
def in_learn_mode(self) -> bool:
|
||||
return time.time() < self._learn_mode_until
|
||||
|
||||
def consider(self, cell_id: str, mcc=None, mnc=None, lac=None,
|
||||
arfcn=None, cipher_a5=None) -> None:
|
||||
"""Called by the consume loop on every cell sighting. In learn
|
||||
mode, immediately graduate; otherwise increment learn_count and
|
||||
graduate when it crosses LEARN_THRESHOLD."""
|
||||
now = time.time()
|
||||
row = self._db.execute(
|
||||
"SELECT learn_count FROM cell_baseline WHERE cell_id = ?",
|
||||
(cell_id,),
|
||||
).fetchone()
|
||||
if row is None:
|
||||
initial = self.LEARN_THRESHOLD if self.in_learn_mode() else 1
|
||||
self._db.execute(
|
||||
"INSERT INTO cell_baseline(cell_id,mcc,mnc,lac,arfcn,learn_count,"
|
||||
"first_learned,last_learned,cipher_a5) VALUES (?,?,?,?,?,?,?,?,?)",
|
||||
(cell_id, mcc, mnc, lac, arfcn, initial, now, now, cipher_a5),
|
||||
)
|
||||
else:
|
||||
new_count = row[0] + 1
|
||||
self._db.execute(
|
||||
"UPDATE cell_baseline SET learn_count = ?, last_learned = ?, "
|
||||
"mcc = COALESCE(?, mcc), mnc = COALESCE(?, mnc), "
|
||||
"lac = COALESCE(?, lac), arfcn = COALESCE(?, arfcn), "
|
||||
"cipher_a5 = COALESCE(?, cipher_a5) WHERE cell_id = ?",
|
||||
(new_count, now, mcc, mnc, lac, arfcn, cipher_a5, cell_id),
|
||||
)
|
||||
self._db.commit()
|
||||
|
||||
def is_baseline(self, cell_id: str) -> bool:
|
||||
row = self._db.execute(
|
||||
"SELECT learn_count FROM cell_baseline WHERE cell_id = ?",
|
||||
(cell_id,),
|
||||
).fetchone()
|
||||
return row is not None and row[0] >= self.LEARN_THRESHOLD
|
||||
|
||||
def get(self, cell_id: str) -> Optional[BaselineCell]:
|
||||
row = self._db.execute(
|
||||
"SELECT cell_id,mcc,mnc,lac,arfcn,learn_count,first_learned,last_learned,cipher_a5 "
|
||||
"FROM cell_baseline WHERE cell_id = ?",
|
||||
(cell_id,),
|
||||
).fetchone()
|
||||
return BaselineCell(*row) if row else None
|
||||
|
||||
def list(self, limit: int = 200) -> list[BaselineCell]:
|
||||
rows = self._db.execute(
|
||||
"SELECT cell_id,mcc,mnc,lac,arfcn,learn_count,first_learned,last_learned,cipher_a5 "
|
||||
"FROM cell_baseline ORDER BY last_learned DESC LIMIT ?",
|
||||
(limit,),
|
||||
).fetchall()
|
||||
return [BaselineCell(*r) for r in rows]
|
||||
|
|
@ -33,6 +33,7 @@ class Observation:
|
|||
frame_nr: int
|
||||
channel: int # GSMTAP channel type
|
||||
sub_type: int # GSMTAP sub_type
|
||||
raw_l3: bytes = b"" # L3 payload after GSMTAP header (BCCH SI / CCCH paging)
|
||||
lac: Optional[int] = None
|
||||
ci: Optional[int] = None
|
||||
mcc: Optional[int] = None
|
||||
|
|
@ -92,17 +93,18 @@ class GsmtapListener:
|
|||
hdr = _parse_gsmtap_header(data)
|
||||
if hdr is None:
|
||||
return
|
||||
raw_l3 = data[hdr["hdr_len"]:]
|
||||
obs = Observation(
|
||||
ts=time.time(),
|
||||
arfcn=hdr["arfcn"],
|
||||
frame_nr=hdr["frame_nr"],
|
||||
channel=hdr["channel"],
|
||||
sub_type=hdr["sub_type"],
|
||||
raw_l3=raw_l3,
|
||||
)
|
||||
# L3 decode (BCCH System Information, CCCH paging) is deferred
|
||||
# to v0.3.1 where we wire scapy + the BCCH parser. For v0.3.0
|
||||
# we record the bare ARFCN/frame_nr/channel to prove the pipe
|
||||
# works and let the operator see real-time activity counts.
|
||||
# L3 payload is now exposed via Observation.raw_l3 for v0.3.1
|
||||
# downstream decoders (BCCH System Information, CCCH paging).
|
||||
# Header-only metadata is still populated for legacy consumers.
|
||||
try:
|
||||
self._queue.put_nowait(obs)
|
||||
except asyncio.QueueFull:
|
||||
|
|
|
|||
238
packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/l3_decode.py
Normal file
238
packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/l3_decode.py
Normal file
|
|
@ -0,0 +1,238 @@
|
|||
# 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.
|
||||
|
||||
"""
|
||||
GSM L3 message decoder — minimal subset for IMSI-catcher detection.
|
||||
|
||||
Hard invariant : public API NEVER emits plaintext IMSI/TMSI/IMEI.
|
||||
The internal `_try_extract_mobile_id()` returns plaintext bytes; the
|
||||
public `_parse_paging()` immediately calls Anonymizer.anonymize() on
|
||||
those bytes BEFORE building the ParsedPagingRequest dataclass. The
|
||||
plaintext goes out of scope at function return.
|
||||
|
||||
Reference :
|
||||
3GPP TS 24.008 §10.5.1.3 Location Area Identification (LAI / MCC+MNC)
|
||||
3GPP TS 24.008 §10.5.1.4 Mobile Identity (IMSI, TMSI, IMEI encoding)
|
||||
3GPP TS 44.018 §9.1 BCCH / CCCH message structures
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
from typing import List, Optional, Tuple
|
||||
|
||||
from sentinelle_gsm.observer import Anonymizer
|
||||
|
||||
|
||||
# Pseudo-length / protocol-discriminator masks
|
||||
L3_PD_RR = 0x06 # Radio Resources protocol discriminator
|
||||
|
||||
# Message-type values inside RR
|
||||
RR_MSG_SI3 = 0x1a
|
||||
RR_MSG_SI4 = 0x1c
|
||||
RR_MSG_SI6 = 0x1e
|
||||
RR_MSG_PAGING_REQUEST_1 = 0x21
|
||||
RR_MSG_PAGING_REQUEST_2 = 0x22
|
||||
RR_MSG_PAGING_REQUEST_3 = 0x24
|
||||
|
||||
|
||||
# Mobile Identity type tags (TS 24.008 §10.5.1.4)
|
||||
MID_TYPE_NONE = 0
|
||||
MID_TYPE_IMSI = 1
|
||||
MID_TYPE_IMEI = 2
|
||||
MID_TYPE_IMEISV = 3
|
||||
MID_TYPE_TMSI = 4
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class CellInfo:
|
||||
"""SI3/SI4/SI6-derived cell metadata. NO subscriber-id fields."""
|
||||
mcc: Optional[int] = None
|
||||
mnc: Optional[int] = None
|
||||
lac: Optional[int] = None
|
||||
ci: Optional[int] = None
|
||||
a5_advertised: Optional[int] = None # 0=A5/0, 1=A5/1, 3=A5/3 etc.
|
||||
t3212_minutes: Optional[int] = None
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class PagedIdentity:
|
||||
"""ONE paged subscriber. subscriber_hash is HMAC-trunc, never plaintext."""
|
||||
id_type: int # MID_TYPE_TMSI or MID_TYPE_IMSI
|
||||
subscriber_hash: str # HMAC-trunc via Anonymizer
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class ParsedPagingRequest:
|
||||
paging_type: int # RR_MSG_PAGING_REQUEST_{1,2,3}
|
||||
identities: List[PagedIdentity] = dataclasses.field(default_factory=list)
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class ParsedFrame:
|
||||
"""Result of L3Decode.parse() — at most one of these is populated."""
|
||||
cell_info: Optional[CellInfo] = None
|
||||
paging: Optional[ParsedPagingRequest] = None
|
||||
|
||||
|
||||
class L3Decode:
|
||||
def __init__(self, anonymizer: Anonymizer):
|
||||
self._anon = anonymizer
|
||||
|
||||
def parse(self, raw: bytes) -> ParsedFrame:
|
||||
if len(raw) < 2 or raw[0] != L3_PD_RR:
|
||||
return ParsedFrame()
|
||||
msg_type = raw[1]
|
||||
if msg_type == RR_MSG_SI3:
|
||||
return ParsedFrame(cell_info=self._parse_si3(raw))
|
||||
if msg_type == RR_MSG_SI4:
|
||||
return ParsedFrame(cell_info=self._parse_si4(raw))
|
||||
if msg_type == RR_MSG_SI6:
|
||||
return ParsedFrame(cell_info=self._parse_si6(raw))
|
||||
if msg_type in (RR_MSG_PAGING_REQUEST_1,
|
||||
RR_MSG_PAGING_REQUEST_2,
|
||||
RR_MSG_PAGING_REQUEST_3):
|
||||
return ParsedFrame(paging=self._parse_paging(raw, msg_type))
|
||||
return ParsedFrame()
|
||||
|
||||
# ── SI parsers (BCCH) ──────────────────────────────────────────────
|
||||
def _parse_si3(self, raw: bytes) -> CellInfo:
|
||||
# TS 44.018 §9.1.35 — System Information Type 3
|
||||
# Offsets after the 2-byte L3 header:
|
||||
# ci (2 bytes)
|
||||
# lai (5 bytes = mcc/mnc/lac)
|
||||
# control_channel_description (3 bytes)
|
||||
# cell_options_bcch (1 byte; bits 0-2 = supported encryption alg mask)
|
||||
# cell_selection_params (2 bytes)
|
||||
# rach_control_params (3 bytes)
|
||||
if len(raw) < 16:
|
||||
return CellInfo()
|
||||
ci = int.from_bytes(raw[2:4], "big")
|
||||
mcc, mnc = _decode_mcc_mnc(raw[4:7])
|
||||
lac = int.from_bytes(raw[7:9], "big")
|
||||
cell_options = raw[12]
|
||||
# Bits 0-2 = NCC permitted (not encryption); the actual A5 mask
|
||||
# is in the upper nibble of byte 12 in some variants — we read
|
||||
# both candidates and prefer the higher A5.
|
||||
a5 = _decode_a5_from_cell_options(cell_options)
|
||||
# T3212 timeout (in deci-hours, 1 byte) is in cell_selection_params
|
||||
t3212_min = raw[13] * 6 if len(raw) > 13 else None # rough
|
||||
return CellInfo(mcc=mcc, mnc=mnc, lac=lac, ci=ci,
|
||||
a5_advertised=a5, t3212_minutes=t3212_min)
|
||||
|
||||
def _parse_si4(self, raw: bytes) -> CellInfo:
|
||||
# TS 44.018 §9.1.36 — Type 4 carries LAI + Cell Identity
|
||||
if len(raw) < 9:
|
||||
return CellInfo()
|
||||
mcc, mnc = _decode_mcc_mnc(raw[2:5])
|
||||
lac = int.from_bytes(raw[5:7], "big")
|
||||
return CellInfo(mcc=mcc, mnc=mnc, lac=lac)
|
||||
|
||||
def _parse_si6(self, raw: bytes) -> CellInfo:
|
||||
# TS 44.018 §9.1.40 — Type 6 carries Cell Identity + LAI +
|
||||
# cell_options bytes with A5 advertised algorithms
|
||||
if len(raw) < 10:
|
||||
return CellInfo()
|
||||
ci = int.from_bytes(raw[2:4], "big")
|
||||
mcc, mnc = _decode_mcc_mnc(raw[4:7])
|
||||
lac = int.from_bytes(raw[7:9], "big")
|
||||
a5 = _decode_a5_from_cell_options(raw[9]) if len(raw) > 9 else None
|
||||
return CellInfo(mcc=mcc, mnc=mnc, lac=lac, ci=ci, a5_advertised=a5)
|
||||
|
||||
# ── Paging parsers (CCCH) ──────────────────────────────────────────
|
||||
def _parse_paging(self, raw: bytes, msg_type: int) -> ParsedPagingRequest:
|
||||
out = ParsedPagingRequest(paging_type=msg_type)
|
||||
# After the 2-byte L3 header :
|
||||
# page_mode + channel_needed (1 byte for type 1; differs slightly per variant)
|
||||
# then 1-N mobile identities
|
||||
ptr = 3
|
||||
while ptr < len(raw):
|
||||
mid, consumed = _try_extract_mobile_id(raw, ptr)
|
||||
if mid is None or consumed == 0:
|
||||
break
|
||||
ptr += consumed
|
||||
id_type, plaintext_bytes = mid
|
||||
if id_type in (MID_TYPE_IMSI, MID_TYPE_TMSI):
|
||||
hashed = self._anon.anonymize(plaintext_bytes.hex())
|
||||
out.identities.append(PagedIdentity(
|
||||
id_type=id_type, subscriber_hash=hashed,
|
||||
))
|
||||
# plaintext_bytes goes out of scope here
|
||||
return out
|
||||
|
||||
|
||||
# ── private helpers ─────────────────────────────────────────────────────
|
||||
def _decode_mcc_mnc(buf: bytes) -> Tuple[Optional[int], Optional[int]]:
|
||||
"""3-byte BCD MCC/MNC encoding per TS 24.008 §10.5.1.3."""
|
||||
if len(buf) < 3:
|
||||
return None, None
|
||||
d1 = buf[0] & 0x0F
|
||||
d2 = (buf[0] >> 4) & 0x0F
|
||||
d3 = buf[1] & 0x0F
|
||||
d_mnc_3 = (buf[1] >> 4) & 0x0F # 0x0F means 2-digit MNC
|
||||
d_mnc_1 = buf[2] & 0x0F
|
||||
d_mnc_2 = (buf[2] >> 4) & 0x0F
|
||||
if 0xF in (d1, d2, d3) or d1 > 9 or d2 > 9 or d3 > 9:
|
||||
return None, None
|
||||
mcc = d1 * 100 + d2 * 10 + d3
|
||||
if d_mnc_3 == 0xF:
|
||||
mnc = d_mnc_1 * 10 + d_mnc_2
|
||||
else:
|
||||
mnc = d_mnc_3 * 100 + d_mnc_2 * 10 + d_mnc_1
|
||||
return mcc, mnc
|
||||
|
||||
|
||||
def _decode_a5_from_cell_options(byte: int) -> Optional[int]:
|
||||
"""The A5 advertised algorithm is encoded in the high bits of the
|
||||
cell_options byte in SI3/SI6. Map back to A5/X integer (0..7)."""
|
||||
# cell_options[5:7] = 3 bits encoding A5/1..A5/7 ; spec says 0 = A5/1
|
||||
a5_field = (byte >> 5) & 0x07
|
||||
return a5_field
|
||||
|
||||
|
||||
def _try_extract_mobile_id(buf: bytes, ofs: int) -> Tuple[Optional[Tuple[int, bytes]], int]:
|
||||
"""Return ((id_type, plaintext_bytes), bytes_consumed) or (None, 0).
|
||||
|
||||
TS 24.008 §10.5.1.4 — Mobile Identity TLV :
|
||||
byte 0 = length L
|
||||
byte 1+ = identity (L bytes); low nibble of byte 1 = type+odd/even
|
||||
"""
|
||||
if ofs >= len(buf):
|
||||
return None, 0
|
||||
L = buf[ofs]
|
||||
if L == 0 or ofs + 1 + L > len(buf):
|
||||
return None, 0
|
||||
body = buf[ofs + 1 : ofs + 1 + L]
|
||||
if not body:
|
||||
return None, 0
|
||||
id_type = body[0] & 0x07
|
||||
if id_type == MID_TYPE_TMSI:
|
||||
# TMSI = 4 bytes
|
||||
if L < 5:
|
||||
return None, 0
|
||||
return (id_type, bytes(body[1:5])), 1 + L
|
||||
if id_type == MID_TYPE_IMSI:
|
||||
# IMSI BCD = up to 15 digits
|
||||
digits = _decode_bcd_imsi(body)
|
||||
if not digits:
|
||||
return None, 0
|
||||
return (id_type, digits.encode("ascii")), 1 + L
|
||||
return None, 1 + L
|
||||
|
||||
|
||||
def _decode_bcd_imsi(body: bytes) -> str:
|
||||
"""Body byte 0 high-nibble = first IMSI digit ; odd flag in body byte 0 low-nibble."""
|
||||
if not body:
|
||||
return ""
|
||||
odd = bool(body[0] & 0x08)
|
||||
digits = [(body[0] >> 4) & 0x0F]
|
||||
for b in body[1:]:
|
||||
digits.append(b & 0x0F)
|
||||
digits.append((b >> 4) & 0x0F)
|
||||
if not odd and digits and digits[-1] == 0x0F:
|
||||
digits.pop()
|
||||
if any(d > 9 for d in digits):
|
||||
return ""
|
||||
return "".join(str(d) for d in digits)
|
||||
|
|
@ -70,6 +70,16 @@ class ObservationsDB:
|
|||
""")
|
||||
self._db.execute("CREATE INDEX IF NOT EXISTS pe_ts_idx ON paging_events(ts)")
|
||||
self._db.execute("CREATE INDEX IF NOT EXISTS pe_cell_idx ON paging_events(cell_id)")
|
||||
self._db.execute("""
|
||||
CREATE TABLE IF NOT EXISTS cell_baseline (
|
||||
cell_id TEXT PRIMARY KEY,
|
||||
mcc INTEGER, mnc INTEGER, lac INTEGER, arfcn INTEGER,
|
||||
learn_count INTEGER NOT NULL DEFAULT 1,
|
||||
first_learned REAL NOT NULL,
|
||||
last_learned REAL NOT NULL,
|
||||
cipher_a5 INTEGER
|
||||
)
|
||||
""")
|
||||
self._db.commit()
|
||||
|
||||
def upsert_sighting(self, s: Sighting) -> None:
|
||||
|
|
|
|||
|
|
@ -1,73 +0,0 @@
|
|||
# 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.
|
||||
|
||||
"""
|
||||
8-heuristic scoring engine — scaffold.
|
||||
|
||||
Each rule returns (score_contribution, reason_str) or (0, None).
|
||||
The final cell score is sum-clamped to [0, 100]. >= threshold → alert.
|
||||
|
||||
v0.1.0 ships the SHAPE only; v0.2 plugs the actual baseline + thresholds
|
||||
once we have real GSMTAP captures.
|
||||
|
||||
Heuristics, per spec §6.1:
|
||||
1. Cipher downgrade (A5/0 advertised or A5/1 where carrier uses A5/3)
|
||||
2. Ghost BTS (CID/LAC not in baseline, anomalous power)
|
||||
3. Identity mismatch (MCC/MNC inconsistent with carrier-on-ARFCN)
|
||||
4. Relocalization storm (frequent LAC churn)
|
||||
5. Identity Request abuse (repeated IMSI requests — IMSI-catcher signature)
|
||||
6. Anomalous neighbours (empty/incoherent neighbour list, aberrant C1/C2)
|
||||
7. T3212 out of carrier band
|
||||
8. Orphan ARFCN (carrier outside the carrier's known frequency plan)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from sentinelle_gsm.observer import GsmCellObservation
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class CellScore:
|
||||
cell_key: str
|
||||
score: int # 0–100
|
||||
reasons: tuple[str, ...]
|
||||
alert: bool # True if score >= alert_threshold
|
||||
|
||||
@property
|
||||
def severity(self) -> str:
|
||||
if self.score >= 80:
|
||||
return "high"
|
||||
if self.score >= 50:
|
||||
return "medium"
|
||||
if self.score >= 20:
|
||||
return "low"
|
||||
return "ok"
|
||||
|
||||
|
||||
# Operator baseline — populated by the scanner timer (grgsm_scanner sweep).
|
||||
# v0.1 keeps an empty baseline; the loaded CSV / SQLite payload is wired in v0.2.
|
||||
@dataclass
|
||||
class Baseline:
|
||||
known_cids: frozenset[tuple[int, int]] = frozenset() # {(lac, cid)}
|
||||
operator_mcc_mnc: tuple[int, int] = (0, 0)
|
||||
plan_arfcns: frozenset[int] = frozenset()
|
||||
|
||||
|
||||
def score(obs: GsmCellObservation, baseline: Baseline,
|
||||
alert_threshold: int = 70) -> CellScore:
|
||||
"""v0.1.0 stub: returns score=0 for every observation.
|
||||
|
||||
The heuristic battery lands in v0.2. The signature here is the
|
||||
contract — callers depend on `score`, `reasons`, `alert`, and
|
||||
`severity` already in v0.1.
|
||||
"""
|
||||
return CellScore(
|
||||
cell_key=obs.key,
|
||||
score=0,
|
||||
reasons=("v0.1.0 scaffold: 8-heuristic engine not yet wired",),
|
||||
alert=False,
|
||||
)
|
||||
|
|
@ -0,0 +1,397 @@
|
|||
# packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring_engine.py
|
||||
# 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.
|
||||
|
||||
"""
|
||||
SecuBox-Deb :: sentinelle_gsm.scoring_engine
|
||||
|
||||
The v0.3.1 8-heuristic scoring engine for GSM IMSI-catcher detection.
|
||||
|
||||
Each heuristic returns a HeuristicResult; ScoringEngine.evaluate() runs
|
||||
the battery, sums the contributions of the triggered ones, clamps to
|
||||
[0, 100], and returns a CellScore.
|
||||
|
||||
Privacy invariant: this engine NEVER sees plaintext IMSI/TMSI. All
|
||||
identity material that reaches identity_request_abuse arrives already
|
||||
hashed (HMAC-trunc, via lib/sentinelle_gsm/l3_decode.py + Anonymizer).
|
||||
The rolling windows key on `subscriber_hash` strings.
|
||||
|
||||
Heuristics (per spec §6.1):
|
||||
|
||||
1. cipher_downgrade observed A5/X < baseline.cipher_a5
|
||||
2. ghost_bts cell not in baseline AND paging seen
|
||||
3. identity_mismatch observed (mcc, mnc) ≠ baseline (mcc, mnc)
|
||||
4. relocalization_storm > N distinct LACs for cell in W seconds
|
||||
5. identity_request_abuse > N paging requests for same subscriber_hash
|
||||
in W seconds
|
||||
6. anomalous_neighbours baseline saw neighbours, current frame empty
|
||||
7. t3212_out_of_band T3212 outside operator-realistic range
|
||||
8. orphan_arfcn FR carrier announces ARFCN outside its plan
|
||||
|
||||
Rolling-window state (in-memory ONLY, no DB persistence in v0.3.1):
|
||||
_lac_window cell_id → deque[(ts, lac)]
|
||||
_idreq_window subscriber_hash → deque[ts]
|
||||
_neighbour_baseline cell_id → set[int] (ARFCNs ever seen)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import dataclasses
|
||||
import time
|
||||
from collections import defaultdict, deque
|
||||
from typing import Any, Deque, Dict, List, Optional, Set, Tuple
|
||||
|
||||
from sentinelle_gsm.baseline import BaselineCell, CellBaseline
|
||||
from sentinelle_gsm.l3_decode import CellInfo, ParsedPagingRequest
|
||||
|
||||
|
||||
# ── Public dataclasses ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class HeuristicResult:
|
||||
name: str
|
||||
triggered: bool
|
||||
score_contrib: int
|
||||
reason: str
|
||||
|
||||
|
||||
@dataclasses.dataclass
|
||||
class CellScore:
|
||||
cell_id: str
|
||||
score: int # clamped to [0, 100]
|
||||
reasons: List[str]
|
||||
triggered: List[str]
|
||||
|
||||
|
||||
# ── France GSM-900 / E-GSM ARFCN plan (approximate) ────────────────────────
|
||||
# Be permissive: only listed (mcc, mnc) pairs are checked. Unknown carriers
|
||||
# return triggered=False to avoid noise.
|
||||
|
||||
CARRIER_ARFCN_PLAN_FR: Dict[str, Dict[Tuple[int, int], Any]] = {
|
||||
"Orange": {(208, 1): range(1, 75)},
|
||||
"SFR": {(208, 10): range(75, 101)},
|
||||
"Bouygues": {(208, 20): list(range(101, 121))},
|
||||
"Free": {(208, 15): list(range(121, 125)) + list(range(975, 1024))},
|
||||
}
|
||||
|
||||
|
||||
def _expected_arfcns_for(mcc: int, mnc: int) -> Optional[List[int]]:
|
||||
"""Return the union of allowed ARFCNs for this (mcc, mnc), or None
|
||||
if we don't know this carrier (permissive: caller skips the check)."""
|
||||
for plan in CARRIER_ARFCN_PLAN_FR.values():
|
||||
if (mcc, mnc) in plan:
|
||||
return list(plan[(mcc, mnc)])
|
||||
return None
|
||||
|
||||
|
||||
# ── ScoringEngine ──────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class ScoringEngine:
|
||||
"""Heuristic battery wired against a CellBaseline.
|
||||
|
||||
Construct once; call evaluate() per parsed L3 frame. Rolling-window
|
||||
state lives in instance fields — keep a single engine per process.
|
||||
"""
|
||||
|
||||
DEFAULT_THRESHOLDS: Dict[str, Dict[str, Any]] = {
|
||||
"cipher_downgrade": {"enabled": True, "score": 40},
|
||||
"ghost_bts": {"enabled": True, "score": 35},
|
||||
"identity_mismatch": {"enabled": True, "score": 30},
|
||||
"relocalization_storm": {"enabled": True, "score": 25, "window_s": 60, "lac_threshold": 3},
|
||||
"identity_request_abuse": {"enabled": True, "score": 35, "window_s": 300, "req_threshold": 2},
|
||||
"anomalous_neighbours": {"enabled": True, "score": 15},
|
||||
"t3212_out_of_band": {"enabled": True, "score": 15, "min": 1, "max": 60},
|
||||
"orphan_arfcn": {"enabled": True, "score": 20},
|
||||
}
|
||||
|
||||
def __init__(self, baseline: CellBaseline,
|
||||
thresholds: Optional[Dict[str, Dict[str, Any]]] = None):
|
||||
self._baseline = baseline
|
||||
# Deep-copy defaults so callers can't mutate class state by reference.
|
||||
self._thresholds: Dict[str, Dict[str, Any]] = copy.deepcopy(self.DEFAULT_THRESHOLDS)
|
||||
if thresholds:
|
||||
self.update_thresholds(thresholds)
|
||||
|
||||
# Rolling-window state — in-memory only.
|
||||
self._lac_window: Dict[str, Deque[Tuple[float, int]]] = defaultdict(deque)
|
||||
self._idreq_window: Dict[str, Deque[float]] = defaultdict(deque)
|
||||
self._neighbour_baseline: Dict[str, Set[int]] = {}
|
||||
|
||||
# ── threshold management ──────────────────────────────────────────────
|
||||
|
||||
def thresholds(self) -> Dict[str, Dict[str, Any]]:
|
||||
"""Return a deep-copy of the effective thresholds."""
|
||||
return copy.deepcopy(self._thresholds)
|
||||
|
||||
def update_thresholds(
|
||||
self, new: Dict[str, Dict[str, Any]]
|
||||
) -> Dict[str, Dict[str, Any]]:
|
||||
"""Per-heuristic deep-merge of `new` into the effective thresholds.
|
||||
|
||||
Allows partial updates: passing
|
||||
{"cipher_downgrade": {"score": 50}}
|
||||
leaves "enabled" untouched on cipher_downgrade and all other
|
||||
heuristics unchanged.
|
||||
"""
|
||||
for h_name, h_update in new.items():
|
||||
if not isinstance(h_update, dict):
|
||||
continue
|
||||
target = self._thresholds.setdefault(h_name, {})
|
||||
for k, v in h_update.items():
|
||||
target[k] = v
|
||||
return self.thresholds()
|
||||
|
||||
# ── main entry point ──────────────────────────────────────────────────
|
||||
|
||||
def evaluate(self,
|
||||
cell_info: Optional[CellInfo],
|
||||
parsed_paging: Optional[ParsedPagingRequest],
|
||||
raw_arfcn: int,
|
||||
raw_neighbours: Optional[Set[int]] = None) -> CellScore:
|
||||
"""Run all 8 heuristics against the supplied parse result.
|
||||
|
||||
cell_id resolution:
|
||||
* If cell_info has (mcc, mnc, lac, ci) → "{mcc}-{mnc}-{lac}-{ci}"
|
||||
* Otherwise fall back to "arfcn-{raw_arfcn}" so paging-only
|
||||
frames still produce a stable key.
|
||||
"""
|
||||
cell_id = self._cell_id_from_info(cell_info) or f"arfcn-{raw_arfcn}"
|
||||
baseline_cell = self._baseline.get(cell_id) if self._baseline else None
|
||||
# Only treat as a real baseline once graduated.
|
||||
if baseline_cell is not None and not self._baseline.is_baseline(cell_id):
|
||||
baseline_cell = None
|
||||
|
||||
results: List[HeuristicResult] = [
|
||||
self._h_cipher_downgrade(cell_info, baseline_cell),
|
||||
self._h_ghost_bts(cell_id, baseline_cell, parsed_paging),
|
||||
self._h_identity_mismatch(cell_info, raw_arfcn, baseline_cell),
|
||||
self._h_relocalization_storm(cell_id, cell_info),
|
||||
self._h_identity_request_abuse(parsed_paging),
|
||||
self._h_anomalous_neighbours(cell_id, raw_neighbours),
|
||||
self._h_t3212_out_of_band(cell_info),
|
||||
self._h_orphan_arfcn(raw_arfcn, cell_info),
|
||||
]
|
||||
|
||||
total = sum(r.score_contrib for r in results if r.triggered)
|
||||
total = min(100, max(0, total))
|
||||
triggered = [r.name for r in results if r.triggered]
|
||||
reasons = [r.reason for r in results if r.triggered]
|
||||
return CellScore(cell_id=cell_id, score=total,
|
||||
reasons=reasons, triggered=triggered)
|
||||
|
||||
# ── helpers ───────────────────────────────────────────────────────────
|
||||
|
||||
@staticmethod
|
||||
def _cell_id_from_info(ci: Optional[CellInfo]) -> Optional[str]:
|
||||
if ci is None:
|
||||
return None
|
||||
if (ci.mcc is None or ci.mnc is None
|
||||
or ci.lac is None or ci.ci is None):
|
||||
return None
|
||||
return f"{ci.mcc}-{ci.mnc}-{ci.lac}-{ci.ci}"
|
||||
|
||||
def _t(self, name: str) -> Dict[str, Any]:
|
||||
return self._thresholds.get(name, {})
|
||||
|
||||
def _enabled(self, name: str) -> bool:
|
||||
return bool(self._t(name).get("enabled", True))
|
||||
|
||||
def _miss(self, name: str) -> HeuristicResult:
|
||||
return HeuristicResult(name=name, triggered=False, score_contrib=0, reason="")
|
||||
|
||||
# ── 1. cipher_downgrade ───────────────────────────────────────────────
|
||||
|
||||
def _h_cipher_downgrade(self, cell_info: Optional[CellInfo],
|
||||
baseline_cell: Optional[BaselineCell]) -> HeuristicResult:
|
||||
name = "cipher_downgrade"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
if (baseline_cell is None or baseline_cell.cipher_a5 is None
|
||||
or cell_info is None or cell_info.a5_advertised is None):
|
||||
return self._miss(name)
|
||||
if cell_info.a5_advertised < baseline_cell.cipher_a5:
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(self._t(name).get("score", 40)),
|
||||
reason=(f"cipher_downgrade A5/{cell_info.a5_advertised} "
|
||||
f"(baseline A5/{baseline_cell.cipher_a5})"),
|
||||
)
|
||||
return self._miss(name)
|
||||
|
||||
# ── 2. ghost_bts ──────────────────────────────────────────────────────
|
||||
|
||||
def _h_ghost_bts(self, cell_id: str,
|
||||
baseline_cell: Optional[BaselineCell],
|
||||
parsed_paging: Optional[ParsedPagingRequest]) -> HeuristicResult:
|
||||
name = "ghost_bts"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
if baseline_cell is None and parsed_paging is not None:
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(self._t(name).get("score", 35)),
|
||||
reason=f"ghost_bts cell {cell_id} not in baseline (paging seen)",
|
||||
)
|
||||
return self._miss(name)
|
||||
|
||||
# ── 3. identity_mismatch ──────────────────────────────────────────────
|
||||
|
||||
def _h_identity_mismatch(self, cell_info: Optional[CellInfo],
|
||||
raw_arfcn: int,
|
||||
baseline_cell: Optional[BaselineCell]) -> HeuristicResult:
|
||||
name = "identity_mismatch"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
if (baseline_cell is None or cell_info is None
|
||||
or cell_info.mcc is None):
|
||||
return self._miss(name)
|
||||
if (baseline_cell.mcc, baseline_cell.mnc) != (cell_info.mcc, cell_info.mnc):
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(self._t(name).get("score", 30)),
|
||||
reason=(f"identity_mismatch (baseline {baseline_cell.mcc}-"
|
||||
f"{baseline_cell.mnc} vs observed "
|
||||
f"{cell_info.mcc}-{cell_info.mnc} on arfcn {raw_arfcn})"),
|
||||
)
|
||||
return self._miss(name)
|
||||
|
||||
# ── 4. relocalization_storm ───────────────────────────────────────────
|
||||
|
||||
def _h_relocalization_storm(self, cell_id: str,
|
||||
cell_info: Optional[CellInfo]) -> HeuristicResult:
|
||||
name = "relocalization_storm"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
t = self._t(name)
|
||||
window_s = float(t.get("window_s", 60))
|
||||
threshold = int(t.get("lac_threshold", 3))
|
||||
if cell_info is None or cell_info.lac is None:
|
||||
return self._miss(name)
|
||||
now = time.time()
|
||||
q = self._lac_window[cell_id]
|
||||
q.append((now, cell_info.lac))
|
||||
cutoff = now - window_s
|
||||
while q and q[0][0] < cutoff:
|
||||
q.popleft()
|
||||
distinct = {lac for _, lac in q}
|
||||
if len(distinct) > threshold:
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(t.get("score", 25)),
|
||||
reason=(f"relocalization_storm: {len(distinct)} distinct "
|
||||
f"LACs in {int(window_s)}s on cell {cell_id}"),
|
||||
)
|
||||
return self._miss(name)
|
||||
|
||||
# ── 5. identity_request_abuse ─────────────────────────────────────────
|
||||
|
||||
def _h_identity_request_abuse(
|
||||
self, parsed_paging: Optional[ParsedPagingRequest]
|
||||
) -> HeuristicResult:
|
||||
name = "identity_request_abuse"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
t = self._t(name)
|
||||
window_s = float(t.get("window_s", 300))
|
||||
threshold = int(t.get("req_threshold", 2))
|
||||
if parsed_paging is None or not parsed_paging.identities:
|
||||
return self._miss(name)
|
||||
now = time.time()
|
||||
cutoff = now - window_s
|
||||
worst_hash: Optional[str] = None
|
||||
worst_count = 0
|
||||
for pid in parsed_paging.identities:
|
||||
h = pid.subscriber_hash
|
||||
if not h:
|
||||
continue
|
||||
q = self._idreq_window[h]
|
||||
q.append(now)
|
||||
while q and q[0] < cutoff:
|
||||
q.popleft()
|
||||
if len(q) > worst_count:
|
||||
worst_count = len(q)
|
||||
worst_hash = h
|
||||
if worst_hash is not None and worst_count > threshold:
|
||||
# Truncate hash to 8 chars for log safety (NEVER plaintext anyway).
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(t.get("score", 35)),
|
||||
reason=(f"identity_request_abuse: {worst_count} requests "
|
||||
f"in {int(window_s)}s for subscriber_hash "
|
||||
f"{worst_hash[:8]}"),
|
||||
)
|
||||
return self._miss(name)
|
||||
|
||||
# ── 6. anomalous_neighbours ───────────────────────────────────────────
|
||||
|
||||
def _h_anomalous_neighbours(self, cell_id: str,
|
||||
raw_neighbours: Optional[Set[int]]) -> HeuristicResult:
|
||||
name = "anomalous_neighbours"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
triggered = False
|
||||
prior = self._neighbour_baseline.get(cell_id)
|
||||
# Trigger condition: we have a prior neighbour set AND the
|
||||
# current frame announces an EMPTY neighbour list.
|
||||
if (prior is not None and len(prior) > 0
|
||||
and raw_neighbours is not None and raw_neighbours == set()):
|
||||
triggered = True
|
||||
# Update running baseline AFTER deciding.
|
||||
if raw_neighbours is not None:
|
||||
self._neighbour_baseline.setdefault(cell_id, set()).update(raw_neighbours)
|
||||
if triggered:
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(self._t(name).get("score", 15)),
|
||||
reason=(f"anomalous_neighbours: cell {cell_id} previously "
|
||||
f"announced {len(prior)} neighbours, now empty"),
|
||||
)
|
||||
return self._miss(name)
|
||||
|
||||
# ── 7. t3212_out_of_band ──────────────────────────────────────────────
|
||||
|
||||
def _h_t3212_out_of_band(self, cell_info: Optional[CellInfo]) -> HeuristicResult:
|
||||
name = "t3212_out_of_band"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
t = self._t(name)
|
||||
lo = int(t.get("min", 1))
|
||||
hi = int(t.get("max", 60))
|
||||
if cell_info is None or cell_info.t3212_minutes is None:
|
||||
return self._miss(name)
|
||||
v = cell_info.t3212_minutes
|
||||
if v < lo or v > hi:
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(t.get("score", 15)),
|
||||
reason=(f"t3212_out_of_band: {v} min outside [{lo}, {hi}]"),
|
||||
)
|
||||
return self._miss(name)
|
||||
|
||||
# ── 8. orphan_arfcn ───────────────────────────────────────────────────
|
||||
|
||||
def _h_orphan_arfcn(self, raw_arfcn: int,
|
||||
cell_info: Optional[CellInfo]) -> HeuristicResult:
|
||||
name = "orphan_arfcn"
|
||||
if not self._enabled(name):
|
||||
return self._miss(name)
|
||||
if (cell_info is None or cell_info.mcc is None
|
||||
or cell_info.mnc is None or cell_info.mcc != 208):
|
||||
return self._miss(name)
|
||||
expected = _expected_arfcns_for(cell_info.mcc, cell_info.mnc)
|
||||
if expected is None:
|
||||
# Unknown (mcc, mnc) — permissive, never flag.
|
||||
return self._miss(name)
|
||||
if raw_arfcn not in expected:
|
||||
return HeuristicResult(
|
||||
name=name, triggered=True,
|
||||
score_contrib=int(self._t(name).get("score", 20)),
|
||||
reason=(f"orphan_arfcn: {cell_info.mcc}-{cell_info.mnc} "
|
||||
f"announces ARFCN {raw_arfcn} outside its plan"),
|
||||
)
|
||||
return self._miss(name)
|
||||
|
|
@ -66,10 +66,18 @@ class TrustedRegistry:
|
|||
self.path.chmod(0o640)
|
||||
|
||||
def add(self, plaintext_imsi: str, label: str) -> TrustedPhone:
|
||||
"""Hash the plaintext IMSI, persist the hash + label, discard plaintext."""
|
||||
"""Hash the plaintext IMSI, persist the hash + label, discard plaintext.
|
||||
|
||||
Hash representation: `imsi.encode("ascii").hex()` so the resulting
|
||||
token matches the canonical form produced by the L3 decoder's
|
||||
paging-request path (l3_decode._parse_paging() anonymizes
|
||||
`plaintext_bytes.hex()` where `plaintext_bytes = digits.encode("ascii")`).
|
||||
This is what enables trusted-phone lookup against paged-subscriber
|
||||
hashes in the consume loop.
|
||||
"""
|
||||
if not plaintext_imsi.isdigit() or not (14 <= len(plaintext_imsi) <= 15):
|
||||
raise ValueError("IMSI must be 14 or 15 digits")
|
||||
imsi_hash = self._anon.anonymize(plaintext_imsi)
|
||||
imsi_hash = self._anon.anonymize(plaintext_imsi.encode("ascii").hex())
|
||||
phone = TrustedPhone(
|
||||
id=str(uuid.uuid4()),
|
||||
imsi_hash=imsi_hash,
|
||||
|
|
|
|||
|
|
@ -183,3 +183,79 @@ def test_observations_db_refuses_plaintext_imsi(tmp_path):
|
|||
)
|
||||
with pytest.raises(ValueError, match="plaintext-IMSI"):
|
||||
db.record_paging(e)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# v0.3.1 — L3 decode + baseline + scoring engine privacy invariants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_l3_decode_returns_no_plaintext_fields():
|
||||
from sentinelle_gsm.l3_decode import PagedIdentity, ParsedPagingRequest, CellInfo
|
||||
from dataclasses import fields
|
||||
for cls in (PagedIdentity, ParsedPagingRequest, CellInfo):
|
||||
names = {f.name for f in fields(cls)}
|
||||
forbidden = {"imsi", "tmsi", "imei", "msisdn", "iccid", "subscriber_id"}
|
||||
assert names.isdisjoint(forbidden), \
|
||||
f"{cls.__name__} has plaintext id fields: {names & forbidden}"
|
||||
|
||||
|
||||
def test_paging_request_hashes_paged_identities():
|
||||
from sentinelle_gsm.observer import Anonymizer
|
||||
from sentinelle_gsm.l3_decode import L3Decode, MID_TYPE_TMSI
|
||||
anon = Anonymizer(b"x" * 32)
|
||||
dec = L3Decode(anon)
|
||||
# Build a paging request type 1 with a TMSI
|
||||
tmsi = b"\xDE\xAD\xBE\xEF"
|
||||
mid = b"\x05\xF4" + tmsi # length=5 + body=[type_byte + 4 bytes TMSI]
|
||||
raw = b"\x06\x21\x00" + mid # L3 header + page_mode + MID
|
||||
frame = dec.parse(raw)
|
||||
assert frame.paging is not None
|
||||
assert len(frame.paging.identities) >= 1
|
||||
pid = frame.paging.identities[0]
|
||||
# The plaintext TMSI bytes (as hex string) MUST NOT appear in the hash
|
||||
assert tmsi.hex() not in pid.subscriber_hash
|
||||
assert pid.id_type == MID_TYPE_TMSI
|
||||
|
||||
|
||||
def test_cell_baseline_has_no_subscriber_fields():
|
||||
from sentinelle_gsm.baseline import BaselineCell
|
||||
from dataclasses import fields
|
||||
names = {f.name for f in fields(BaselineCell)}
|
||||
forbidden = {"imsi", "tmsi", "imei", "subscriber_hash", "subscriber_id", "msisdn", "iccid"}
|
||||
assert names.isdisjoint(forbidden), \
|
||||
f"BaselineCell has forbidden id-related fields: {names & forbidden}"
|
||||
|
||||
|
||||
def test_scoring_engine_reasons_contain_no_plaintext_imsi():
|
||||
"""The reason strings produced by each heuristic must NOT contain
|
||||
a 15-digit token (plaintext IMSI shape). All identifiers in reasons
|
||||
should already be hashed/truncated."""
|
||||
import re
|
||||
from sentinelle_gsm.observer import Anonymizer
|
||||
from sentinelle_gsm.scoring_engine import ScoringEngine
|
||||
from sentinelle_gsm.baseline import CellBaseline
|
||||
from sentinelle_gsm.observations import ObservationsDB
|
||||
from sentinelle_gsm.l3_decode import CellInfo, ParsedPagingRequest, PagedIdentity, MID_TYPE_TMSI
|
||||
import tempfile, pathlib
|
||||
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
obs_db = ObservationsDB(pathlib.Path(td) / "obs.db")
|
||||
baseline = CellBaseline(obs_db._db)
|
||||
engine = ScoringEngine(baseline)
|
||||
|
||||
# Feed an observation that triggers identity_request_abuse with a TMSI
|
||||
# hash like '208201234567890ABCDEF' — the hash output happens to
|
||||
# contain 15 digits in the abusive case; the reason should still
|
||||
# not echo those raw digits unguarded.
|
||||
# Just verify NO reason string from a typical observation flow
|
||||
# matches the plaintext-IMSI shape \b\d{15}\b.
|
||||
ci = CellInfo(mcc=208, mnc=1, lac=100, ci=12345)
|
||||
pg = ParsedPagingRequest(
|
||||
paging_type=0x21,
|
||||
identities=[PagedIdentity(id_type=MID_TYPE_TMSI, subscriber_hash="abc123def456")],
|
||||
)
|
||||
result = engine.evaluate(ci, pg, raw_arfcn=42)
|
||||
for reason in result.reasons:
|
||||
assert not re.search(r"\b\d{15}\b", reason), \
|
||||
f"scoring reason contains plaintext-IMSI shape: {reason!r}"
|
||||
|
|
|
|||
|
|
@ -38,6 +38,8 @@
|
|||
<a href="#logs" class="nav-link">Live logs</a>
|
||||
<a href="#scan-control" class="nav-link">Scan control</a>
|
||||
<a href="#observations" class="nav-link">Observations</a>
|
||||
<a href="#baseline" class="nav-link">Baseline</a>
|
||||
<a href="#scoring" class="nav-link">Scoring</a>
|
||||
<a href="#actions" class="nav-link">Actions</a>
|
||||
</nav>
|
||||
<div class="local-nav-foot">
|
||||
|
|
@ -229,6 +231,50 @@
|
|||
</div>
|
||||
</section>
|
||||
|
||||
<!-- BASELINE panel (v0.3.1) ------------------------------------------- -->
|
||||
<section id="baseline" class="panel">
|
||||
<div class="panel-head">
|
||||
<h2>Operator baseline <span class="badge" id="baseline-count">0</span></h2>
|
||||
<div class="panel-actions">
|
||||
<button class="btn" id="btn-refresh-baseline" type="button">Refresh</button>
|
||||
<button class="btn primary" id="btn-baseline-learn" type="button">Start learn (5 min)</button>
|
||||
</div>
|
||||
</div>
|
||||
<div class="table-wrap">
|
||||
<table class="data-table">
|
||||
<thead>
|
||||
<tr>
|
||||
<th>Cell ID</th><th>MCC</th><th>MNC</th><th>LAC</th><th>ARFCN</th>
|
||||
<th>Cipher</th><th>Learn count</th><th>Last learned</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody id="baseline-tbody">
|
||||
<tr class="empty"><td colspan="8">no baseline yet — start a scan + click "Start learn"</td></tr>
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
<p class="hint">
|
||||
Cells graduate to baseline after 3 sightings under normal scan. The "Start learn" button accepts every cell observed within the next 5 minutes immediately.
|
||||
</p>
|
||||
</section>
|
||||
|
||||
<!-- SCORING panel (v0.3.1) -------------------------------------------- -->
|
||||
<section id="scoring" class="panel">
|
||||
<div class="panel-head">
|
||||
<h2>Scoring heuristics</h2>
|
||||
<div class="panel-actions">
|
||||
<button class="btn" id="btn-refresh-scoring" type="button">Refresh</button>
|
||||
<button class="btn primary" id="btn-save-scoring" type="button">Apply changes</button>
|
||||
</div>
|
||||
</div>
|
||||
<div class="threshold-grid" id="threshold-grid">
|
||||
<!-- rows injected by JS -->
|
||||
</div>
|
||||
<p class="hint">
|
||||
Score sum ≥ 60 (default) emits an alert. Each heuristic can be enabled/disabled and its contribution adjusted. Changes are audit-logged to the live journal stream.
|
||||
</p>
|
||||
</section>
|
||||
|
||||
<!-- ACTIONS panel ------------------------------------------------------ -->
|
||||
<section id="actions" class="panel">
|
||||
<div class="panel-head">
|
||||
|
|
|
|||
|
|
@ -585,8 +585,52 @@ a:hover { color: var(--text-primary); }
|
|||
.local-nav-foot { display: none; }
|
||||
}
|
||||
|
||||
/* ── v0.3.1 trusted_label chip (alerts table) ─────────────────────────── */
|
||||
.trusted-chip {
|
||||
display: inline-block;
|
||||
padding: 2px 8px;
|
||||
border-radius: 999px;
|
||||
background: rgba(61, 53, 160, 0.25);
|
||||
color: var(--mind-violet-lt, #7c75d8);
|
||||
font-size: 0.75rem;
|
||||
font-weight: 600;
|
||||
letter-spacing: 0.02em;
|
||||
}
|
||||
|
||||
/* ── v0.3.1 threshold grid (scoring panel) ────────────────────────────── */
|
||||
.threshold-grid {
|
||||
display: grid;
|
||||
gap: 0.5rem;
|
||||
padding: 0.5rem 0;
|
||||
}
|
||||
.threshold-row {
|
||||
display: grid;
|
||||
grid-template-columns: 200px auto auto;
|
||||
align-items: center;
|
||||
gap: 1rem;
|
||||
padding: 0.5rem;
|
||||
background: rgba(0, 0, 0, 0.2);
|
||||
border-radius: 4px;
|
||||
border-left: 3px solid var(--mind-violet, #3D35A0);
|
||||
}
|
||||
.threshold-name {
|
||||
font-family: "JetBrains Mono", ui-monospace, monospace;
|
||||
color: var(--text-primary);
|
||||
font-weight: 500;
|
||||
}
|
||||
.threshold-row input[type="number"] {
|
||||
width: 60px;
|
||||
background: rgba(0, 0, 0, 0.4);
|
||||
color: var(--text-primary);
|
||||
border: 1px solid var(--mind-violet-dim, #555);
|
||||
padding: 2px 6px;
|
||||
border-radius: 3px;
|
||||
font-family: "JetBrains Mono", ui-monospace, monospace;
|
||||
}
|
||||
|
||||
@media (max-width: 520px) {
|
||||
.status-strip { grid-template-columns: 1fr; }
|
||||
.scan-status-row { grid-template-columns: 1fr; }
|
||||
.modal-content { max-width: 95%; }
|
||||
.threshold-row { grid-template-columns: 1fr; }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -61,6 +61,14 @@
|
|||
obsCount: document.getElementById("obs-count"),
|
||||
obsTbody: document.getElementById("obs-tbody"),
|
||||
btnRefreshObs: document.getElementById("btn-refresh-obs"),
|
||||
// v0.3.1 baseline + scoring
|
||||
baselineCount: document.getElementById("baseline-count"),
|
||||
baselineTbody: document.getElementById("baseline-tbody"),
|
||||
btnRefreshBaseline: document.getElementById("btn-refresh-baseline"),
|
||||
btnBaselineLearn: document.getElementById("btn-baseline-learn"),
|
||||
thresholdGrid: document.getElementById("threshold-grid"),
|
||||
btnRefreshScoring: document.getElementById("btn-refresh-scoring"),
|
||||
btnSaveScoring: document.getElementById("btn-save-scoring"),
|
||||
};
|
||||
|
||||
// ── state ───────────────────────────────────────────────────────────
|
||||
|
|
@ -233,7 +241,7 @@
|
|||
const tr = document.createElement("tr");
|
||||
tr.dataset.alertId = alert.id;
|
||||
const targetCell = alert.trusted_label
|
||||
? `<span class="target-pill">${escapeHtml(alert.trusted_label)}</span>`
|
||||
? `<span class="trusted-chip">${escapeHtml(alert.trusted_label)}</span>`
|
||||
: '<span style="color:var(--text-dim)">—</span>';
|
||||
tr.innerHTML =
|
||||
`<td class="col-time">${escapeHtml(fmtTime(alert.ts))}</td>` +
|
||||
|
|
@ -553,6 +561,181 @@
|
|||
}
|
||||
}
|
||||
|
||||
// ── baseline + scoring (v0.3.1) ─────────────────────────────────────
|
||||
function _buildBaselineRow(c) {
|
||||
const tr = document.createElement("tr");
|
||||
const fmt = (v) => (v === null || v === undefined || v === "") ? "—" : String(v);
|
||||
tr.innerHTML =
|
||||
`<td class="col-cell">${escapeHtml(c.cell_id)}</td>` +
|
||||
`<td class="mono">${escapeHtml(fmt(c.mcc))}</td>` +
|
||||
`<td class="mono">${escapeHtml(fmt(c.mnc))}</td>` +
|
||||
`<td class="mono">${escapeHtml(fmt(c.lac))}</td>` +
|
||||
`<td class="mono">${escapeHtml(fmt(c.arfcn))}</td>` +
|
||||
`<td class="mono">${escapeHtml(fmt(c.cipher_a5))}</td>` +
|
||||
`<td class="mono">${escapeHtml(fmt(c.learn_count))}</td>` +
|
||||
`<td class="mono">${escapeHtml(fmtDateTime(c.last_learned))}</td>`;
|
||||
return tr;
|
||||
}
|
||||
|
||||
function renderBaselineTable(cells) {
|
||||
if (!els.baselineTbody) return;
|
||||
els.baselineTbody.innerHTML = "";
|
||||
if (!cells || cells.length === 0) {
|
||||
els.baselineTbody.innerHTML =
|
||||
'<tr class="empty"><td colspan="8">no baseline yet — start a scan + click "Start learn"</td></tr>';
|
||||
return;
|
||||
}
|
||||
cells.forEach((c) => els.baselineTbody.appendChild(_buildBaselineRow(c)));
|
||||
}
|
||||
|
||||
async function loadBaseline() {
|
||||
try {
|
||||
const r = await fetch(API + "/baseline?limit=200", { credentials: "include" });
|
||||
if (!r.ok) throw new Error("HTTP " + r.status);
|
||||
const j = await r.json();
|
||||
const cells = j.cells || [];
|
||||
if (els.baselineCount) els.baselineCount.textContent = String(cells.length);
|
||||
renderBaselineTable(cells);
|
||||
} catch (e) {
|
||||
console.warn("loadBaseline failed:", e);
|
||||
}
|
||||
}
|
||||
|
||||
let _baselineLearnPollTimer = null;
|
||||
async function startBaselineLearn() {
|
||||
try {
|
||||
const r = await fetch(API + "/baseline/learn", {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ sweep_seconds: 300 }),
|
||||
credentials: "include",
|
||||
});
|
||||
if (!r.ok) throw new Error("HTTP " + r.status);
|
||||
toast("baseline learn mode armed for 5 min", "ok");
|
||||
// re-poll baseline every 15s for 5 min so the table fills as cells graduate
|
||||
if (_baselineLearnPollTimer) clearInterval(_baselineLearnPollTimer);
|
||||
let ticks = 0;
|
||||
_baselineLearnPollTimer = setInterval(() => {
|
||||
loadBaseline();
|
||||
ticks += 1;
|
||||
if (ticks >= 20) { // 20 * 15s = 5 min
|
||||
clearInterval(_baselineLearnPollTimer);
|
||||
_baselineLearnPollTimer = null;
|
||||
}
|
||||
}, 15000);
|
||||
} catch (e) {
|
||||
toast("baseline learn failed: " + e.message, "err");
|
||||
}
|
||||
}
|
||||
|
||||
// Local cache of current scoring threshold form values, keyed by heuristic
|
||||
// name. Each value is the full threshold object (enabled, score, plus any
|
||||
// heuristic-specific fields the backend cares about).
|
||||
let _thresholdState = {};
|
||||
|
||||
function _clampScore(n) {
|
||||
n = Number(n);
|
||||
if (!Number.isFinite(n)) return 0;
|
||||
if (n < 0) return 0;
|
||||
if (n > 100) return 100;
|
||||
return Math.round(n);
|
||||
}
|
||||
|
||||
function renderThresholdGrid(thresholds) {
|
||||
const grid = els.thresholdGrid;
|
||||
if (!grid) return;
|
||||
grid.innerHTML = "";
|
||||
const names = Object.keys(thresholds || {}).sort();
|
||||
if (names.length === 0) {
|
||||
grid.innerHTML = '<div class="empty">no scoring heuristics defined</div>';
|
||||
return;
|
||||
}
|
||||
names.forEach((name) => {
|
||||
const t = thresholds[name] || {};
|
||||
const enabled = !!t.enabled;
|
||||
const score = _clampScore(t.score);
|
||||
const row = document.createElement("div");
|
||||
row.className = "threshold-row";
|
||||
row.innerHTML =
|
||||
`<span class="threshold-name">${escapeHtml(name)}</span>` +
|
||||
`<label><input type="checkbox" data-heur="${escapeHtml(name)}" data-field="enabled"` +
|
||||
(enabled ? " checked" : "") + `> enabled</label>` +
|
||||
`<label>score <input type="number" min="0" max="100" step="1"` +
|
||||
` data-heur="${escapeHtml(name)}" data-field="score" value="${score}"></label>`;
|
||||
grid.appendChild(row);
|
||||
});
|
||||
|
||||
// Wire change listeners. We listen on the grid container (event delegation).
|
||||
grid.addEventListener("change", _onThresholdChange);
|
||||
}
|
||||
|
||||
function _onThresholdChange(e) {
|
||||
const inp = e.target;
|
||||
if (!inp || !inp.dataset) return;
|
||||
const heur = inp.dataset.heur;
|
||||
const field = inp.dataset.field;
|
||||
if (!heur || !field) return;
|
||||
if (!_thresholdState[heur]) _thresholdState[heur] = {};
|
||||
if (field === "enabled") {
|
||||
_thresholdState[heur].enabled = !!inp.checked;
|
||||
} else if (field === "score") {
|
||||
const n = _clampScore(inp.value);
|
||||
if (String(n) !== String(inp.value)) inp.value = String(n);
|
||||
_thresholdState[heur].score = n;
|
||||
}
|
||||
}
|
||||
|
||||
async function loadThresholds() {
|
||||
try {
|
||||
const r = await fetch(API + "/scoring/thresholds", { credentials: "include" });
|
||||
if (!r.ok) throw new Error("HTTP " + r.status);
|
||||
const j = await r.json();
|
||||
const thresholds = j.thresholds || {};
|
||||
// Deep-ish copy: keep any backend-only fields so a round-trip POST does
|
||||
// not silently drop them. The form only edits enabled + score.
|
||||
_thresholdState = {};
|
||||
Object.keys(thresholds).forEach((name) => {
|
||||
const t = thresholds[name] || {};
|
||||
_thresholdState[name] = Object.assign({}, t);
|
||||
});
|
||||
renderThresholdGrid(thresholds);
|
||||
} catch (e) {
|
||||
console.warn("loadThresholds failed:", e);
|
||||
}
|
||||
}
|
||||
|
||||
async function saveThresholds() {
|
||||
try {
|
||||
// Clamp any out-of-range score values one more time before POST.
|
||||
Object.keys(_thresholdState).forEach((name) => {
|
||||
if ("score" in _thresholdState[name]) {
|
||||
_thresholdState[name].score = _clampScore(_thresholdState[name].score);
|
||||
}
|
||||
});
|
||||
const r = await fetch(API + "/scoring/thresholds", {
|
||||
method: "POST",
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify(_thresholdState),
|
||||
credentials: "include",
|
||||
});
|
||||
if (!r.ok) {
|
||||
let detail = "HTTP " + r.status;
|
||||
try { const b = await r.json(); if (b && b.detail) detail = b.detail; } catch (_) {}
|
||||
throw new Error(detail);
|
||||
}
|
||||
const j = await r.json();
|
||||
const thresholds = j.thresholds || j;
|
||||
_thresholdState = {};
|
||||
Object.keys(thresholds).forEach((name) => {
|
||||
_thresholdState[name] = Object.assign({}, thresholds[name] || {});
|
||||
});
|
||||
renderThresholdGrid(thresholds);
|
||||
toast("scoring thresholds updated", "ok");
|
||||
} catch (e) {
|
||||
toast("save failed: " + e.message, "err");
|
||||
}
|
||||
}
|
||||
|
||||
// ── modal ───────────────────────────────────────────────────────────
|
||||
function openModal() {
|
||||
els.modal.classList.add("show");
|
||||
|
|
@ -642,6 +825,12 @@
|
|||
if (els.btnScanStart) els.btnScanStart.addEventListener("click", startScan);
|
||||
if (els.btnScanStop) els.btnScanStop.addEventListener("click", stopScan);
|
||||
if (els.btnRefreshObs) els.btnRefreshObs.addEventListener("click", loadObservations);
|
||||
|
||||
// v0.3.1 baseline + scoring
|
||||
if (els.btnRefreshBaseline) els.btnRefreshBaseline.addEventListener("click", loadBaseline);
|
||||
if (els.btnBaselineLearn) els.btnBaselineLearn.addEventListener("click", startBaselineLearn);
|
||||
if (els.btnRefreshScoring) els.btnRefreshScoring.addEventListener("click", loadThresholds);
|
||||
if (els.btnSaveScoring) els.btnSaveScoring.addEventListener("click", saveThresholds);
|
||||
}
|
||||
|
||||
// ── init ────────────────────────────────────────────────────────────
|
||||
|
|
@ -655,14 +844,19 @@
|
|||
startJournalStream();
|
||||
loadScanStatus();
|
||||
loadObservations();
|
||||
loadBaseline();
|
||||
loadThresholds();
|
||||
// Background polling every 10s. When no scan is running, we still poll
|
||||
// /scan/status (cheap) so the UI catches an externally-started scan,
|
||||
// but we skip /observations to avoid hitting sqlite for nothing — the
|
||||
// table is already at its post-scan terminal state and the user can
|
||||
// hit Refresh manually.
|
||||
// but we skip /observations + /baseline to avoid hitting sqlite for
|
||||
// nothing — the tables are already at their post-scan terminal state
|
||||
// and the user can hit Refresh manually.
|
||||
_scanPollTimer = setInterval(() => {
|
||||
loadScanStatus();
|
||||
if (_scanRunning) loadObservations();
|
||||
if (_scanRunning) {
|
||||
loadObservations();
|
||||
loadBaseline();
|
||||
}
|
||||
}, 10000);
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user