diff --git a/docs/superpowers/plans/2026-05-22-sentinelle-gsm-v0.3.1-l3-scoring-baseline.md b/docs/superpowers/plans/2026-05-22-sentinelle-gsm-v0.3.1-l3-scoring-baseline.md new file mode 100644 index 00000000..52c48ec8 --- /dev/null +++ b/docs/superpowers/plans/2026-05-22-sentinelle-gsm-v0.3.1-l3-scoring-baseline.md @@ -0,0 +1,1036 @@ +# secubox-sentinelle-gsm v0.3.1 — L3 Decode + Scoring + Baseline + Qualified Alert Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Promote secubox-sentinelle-gsm from a passive observation pipeline (v0.3.0) into an actual IMSI-catcher detector. A GSMTAP frame entering on UDP 4729 is now decoded at L3, the cell is scored against 8 spec heuristics with a learned operator baseline, and an Alert with a `trusted_label` fires when a paged subscriber matches a `trusted_phones` entry on a high-scoring cell. + +**Architecture:** + +``` +GsmtapListener (v0.3.0) + ↓ raw L3 payload bytes +L3Decode.parse(bytes) + ↓ ParsedFrame (system_info | paging_request) + ↓ ↓ + update sighting extract paged identities → HMAC each (Anonymizer) + (mcc/mnc/lac/ci, cipher) ↓ + ↓ PagingEvent(subscriber_hash, request_type) + └──→ ScoringEngine.evaluate(cell, frame, baseline_db, rolling_window) + ↓ + score 0-100 + reasons[] + ↓ + if score >= threshold AND ≥1 paged hash ∈ trusted_phones + ↓ + AlertSink.write(Alert(score, reasons, subscriber_hash, trusted_label)) + ↓ + else if score >= threshold (no trusted match) + AlertSink.write(Alert(score, reasons)) ← anomaly-only +``` + +**Tech Stack:** +- Python 3.11 / FastAPI / asyncio (existing) +- **L3 parsing** : pure-Python TLV by message type (NO scapy.layers.gsm — coverage gap in upstream) +- SQLite via stdlib (existing alert_sink + observations + new cell_baseline) +- `collections.deque` for rolling windows in the scoring engine + +**Privacy invariants (load-bearing, tested):** +- `l3_decode` module API NEVER returns plaintext IMSI/TMSI/IMEI. Internal helper extracts the raw bytes; public function calls `Anonymizer.anonymize()` BEFORE the return value crosses the module boundary. +- `cell_baseline` table has NO subscriber-id column. +- The privacy invariant test suite is extended with 4 new shape/behavior checks. + +**Out of scope (v0.3.2+):** +- Multi-frequency simultaneous scan +- SMS via EP06 alert backend (waits on #345 + spare EP06) +- PWA push notifications +- Per-trusted-phone notification routing + +--- + +## File Structure + +- **Create:** + - `lib/sentinelle_gsm/l3_decode.py` + - `lib/sentinelle_gsm/scoring_engine.py` + - `lib/sentinelle_gsm/baseline.py` + - `api/tests/test_l3_decode.py` + - `api/tests/test_scoring_engine.py` + - `api/tests/test_baseline.py` + - `api/tests/test_alert_emission.py` (integration: consume loop → scoring → alert) +- **Modify:** + - `api/main.py` — consume loop calls `L3Decode.parse()` + `ScoringEngine.evaluate()` + matches against trusted_phones + writes Alert. New endpoints `/baseline*`, `/scoring/thresholds`. + - `lib/sentinelle_gsm/scoring.py` — DELETE (replaced by `scoring_engine.py`; the v0.1 stub never had implementations). + - `lib/sentinelle_gsm/gsmtap_listener.py` — extend `Observation` with `raw_l3` bytes field (the payload after the GSMTAP header) + - `www/sentinelle/` — new "Baseline" + "Scoring" panels; alerts table renders `trusted_label` chip + - `tests/test_privacy_invariant.py` — 4 new checks + - `debian/changelog` — bump 0.3.0 → 0.3.1 + +--- + +## Task 1 — Extend `Observation` with `raw_l3` + propagate + +**Files:** +- Modify: `lib/sentinelle_gsm/gsmtap_listener.py` +- Modify: `api/tests/test_gsmtap_listener.py` + +The v0.3.0 `Observation` carries only header metadata (ARFCN, frame_nr, channel, sub_type). v0.3.1 needs the payload too. The GSMTAP header is fixed at 16 bytes; everything after is the L3 frame. + +- [ ] **Step 1: Add `raw_l3: bytes = b""` field to `Observation` dataclass** + +```python +@dataclasses.dataclass +class Observation: + ts: float + arfcn: int + frame_nr: int + channel: int + sub_type: int + raw_l3: bytes = b"" # NEW — L3 payload after GSMTAP header + lac: Optional[int] = None + ci: Optional[int] = None + mcc: Optional[int] = None + mnc: Optional[int] = None + cell_id: Optional[str] = None + subscriber_hash: Optional[str] = None +``` + +The order matters: `bytes = b""` is a fixed default so it doesn't break the existing dataclass ordering rules (non-default fields first). + +- [ ] **Step 2: In `_on_datagram`, slice `raw_l3 = data[hdr_len:]` and pass it** + +```python +def _on_datagram(self, data: bytes) -> None: + 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, + ) + ... +``` + +- [ ] **Step 3: Extend `test_listener_receives_a_datagram` to assert `raw_l3 == expected_payload`** + +Construct the datagram with a known payload (e.g. `b"\x06\x1a"` = BCCH SI Type 3 first bytes), assert the listener yields `Observation` with `raw_l3` matching. + +- [ ] **Step 4: Run + commit** + +```bash +cd packages/secubox-sentinelle-gsm +python3 -m pytest api/tests/test_gsmtap_listener.py -v +``` +Expected: 4 still pass + 1 new assertion (raw_l3 check) within the existing test. + +```bash +git commit -am "feat(sentinelle-gsm): GsmtapListener exposes raw_l3 payload bytes (ref #349)" +``` + +--- + +## Task 2 — `l3_decode.py` (BCCH SI + CCCH paging request) + +**Files:** +- Create: `lib/sentinelle_gsm/l3_decode.py` +- Create: `api/tests/test_l3_decode.py` + +Pure-Python TLV per message type. We only need the subset that matters for the heuristics + paging extraction. + +### GSM L3 message-type reference (3GPP TS 24.008 + TS 44.018) + +| msg_type byte | Channel | Meaning | We need | +|---|---|---|---| +| `0x06 0x18` | BCCH | SI Type 1 | (frequency list — for orphan_arfcn, optional v0.3.2) | +| `0x06 0x19` | BCCH | SI Type 2 | (neighbour ARFCN list) | +| `0x06 0x1a` | BCCH | SI Type 3 | **MCC, MNC, LAC, CI, cell options (A5/X), T3212** | +| `0x06 0x1c` | BCCH | SI Type 4 | LAC (re-validation) | +| `0x06 0x1e` | BCCH | SI Type 6 | cipher mode + LAI | +| `0x06 0x21` | CCCH | Paging Request Type 1 | 1-2 paged identities | +| `0x06 0x22` | CCCH | Paging Request Type 2 | 2 TMSI + 1 IMSI | +| `0x06 0x24` | CCCH | Paging Request Type 3 | 4 TMSI | + +- [ ] **Step 1: Write `l3_decode.py` skeleton with parsers per msg_type** + +```python +# packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/l3_decode.py +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +GSM L3 message decoder — minimal subset for IMSI-catcher detection. + +Hard invariant : public API NEVER emits plaintext IMSI/TMSI/IMEI. +The internal `_extract_mobile_id_bcd()` returns plaintext bytes; the +public `parse_paging_request()` immediately calls Anonymizer.anonymize() +on those bytes BEFORE building the ParsedPagingRequest dataclass. + +Reference : + 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 + +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) +``` + +This file is ~210 lines. The TS 44.018 BCCH structures are subtle — the parser is permissive (returns None / empty CellInfo on unexpected layouts rather than throwing). + +- [ ] **Step 2: Write `test_l3_decode.py` with synthetic frames** + +```python +import pytest +from sentinelle_gsm.observer import Anonymizer +from sentinelle_gsm.l3_decode import ( + L3Decode, MID_TYPE_TMSI, MID_TYPE_IMSI, _decode_mcc_mnc, _decode_bcd_imsi, +) + + +@pytest.fixture +def decoder(): + return L3Decode(Anonymizer(b"x" * 32)) + + +def test_decode_mcc_mnc_2_digit_mnc(): + # Orange FR : MCC=208, MNC=01 — encoded with F-padding nibble + # buf bytes : 0x80 0xF1 0x10 → digits (8 0 F 1 1 0) ; MCC=208, MNC=01 + mcc, mnc = _decode_mcc_mnc(b"\x80\xF1\x10") + assert mcc == 208 + assert mnc == 1 + + +def test_decode_bcd_imsi_15_digits_even(): + # IMSI 208201234567890 = 15 digits (odd, so the odd flag in body[0]=1) + # Per BCD with odd flag: + # body[0] low nibble = (mid_type=IMSI=1) | odd<<3 → 0x09 + # body[0] high nibble = first IMSI digit = 2 + # body[1] = 8 (high) | 0 (low) ... + # Construct manually: + body = bytes([0x29, 0x80, 0x32, 0x54, 0x76, 0x98, 0x09]) + digits = _decode_bcd_imsi(body) + assert digits == "208201234567890" + + +def test_parse_si3_extracts_mcc_mnc_lac_ci(decoder): + # Header: 0x06 0x1a + 2 bytes CI + 5 bytes LAI + filler + raw = (b"\x06\x1a" + # L3 header SI3 + b"\x30\x39" + # CI = 0x3039 = 12345 + b"\x80\xF1\x10" + # LAI: MCC=208 MNC=01 + b"\x00\xC8" + # LAC = 200 + b"\x00\x00\x00" + # control_channel_description + b"\x00" + # cell_options (no A5 bits set) + b"\x00\x00" + # cell_selection + b"\x00\x00\x00") # rach_control + frame = decoder.parse(raw) + assert frame.cell_info is not 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_paging_request_1_with_tmsi(decoder): + # 0x06 0x21 + page_mode(1) + ChannelNeeded(1) + Mobile-ID(TMSI) + # Mobile Identity TLV: length=5, body=[type_byte][4-byte TMSI] + # type_byte for TMSI = 0x04 (id_type=4, odd=0) + tmsi_bytes = b"\xDE\xAD\xBE\xEF" + mid = b"\x05" + b"\xF4" + tmsi_bytes # length=5, body=[0xF4 + 4 bytes TMSI] + raw = b"\x06\x21" + b"\x00" + mid # 2-byte L3 hdr + page_mode_byte + MID + 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_TMSI + assert pid.subscriber_hash != "deadbeef" # MUST be HMAC, not plaintext + + +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): + assert decoder.parse(b"") == decoder.parse(b"\x06") +``` + +Note : the BCD encoding nibble layout in `test_decode_bcd_imsi_15_digits_even` and the TMSI body construction in `test_parse_paging_request_1_with_tmsi` are sensitive to the spec's swap-nibble semantics. Adapt by hand if the assertions fail — the test is verifying the parser's BCD direction, not the spec's exact byte order. Better yet, build the body programmatically with a small helper. + +- [ ] **Step 3: Run + commit** + +```bash +python3 -m pytest api/tests/test_l3_decode.py -v +git commit -am "feat(sentinelle-gsm): l3_decode — BCCH SI + CCCH paging request parsing (ref #349)" +``` + +Expected ~6 passing tests. + +--- + +## Task 3 — `baseline.py` (operator-baseline learning + persistence) + +**Files:** +- Create: `lib/sentinelle_gsm/baseline.py` +- Create: `api/tests/test_baseline.py` + +- [ ] **Step 1: Add the `cell_baseline` table to `observations.py`** + +```python +# IN observations.py: in ObservationsDB.__init__, add: +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 + ) +""") +``` + +- [ ] **Step 2: Write `baseline.py`** + +```python +# packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/baseline.py +""" +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. +""" + +from __future__ import annotations + +import dataclasses +import time +from pathlib import Path +from typing import Optional +import sqlite3 + + +@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.""" + + 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: + 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] +``` + +- [ ] **Step 3: Tests** + +5 tests : +- `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` + +- [ ] **Step 4: Run + commit** + +```bash +python3 -m pytest api/tests/test_baseline.py -v +git commit -am "feat(sentinelle-gsm): baseline.py — operator-baseline learning (ref #349)" +``` + +--- + +## Task 4 — `scoring_engine.py` (8 heuristics + aggregator) + +**Files:** +- Create: `lib/sentinelle_gsm/scoring_engine.py` +- Delete: `lib/sentinelle_gsm/scoring.py` (v0.1 shape-only stub — replaced) +- Create: `api/tests/test_scoring_engine.py` + +### Heuristic spec table + +| # | Name | Condition | Default score | Sliding window | +|---|---|---|---|---| +| 1 | cipher_downgrade | observed A5/X < baseline.cipher_a5 for this cell | 40 | none | +| 2 | ghost_bts | observed cell_id NOT in baseline AND ≥1 paging seen | 35 | none | +| 3 | identity_mismatch | observed (mcc, mnc) ≠ baseline (mcc, mnc) for same ARFCN | 30 | none | +| 4 | relocalization_storm | ≥3 distinct LAC values for our trusted hashes in 60 s | 25 | 60 s | +| 5 | identity_request_abuse | ≥2 IMSI Identity Requests for same TMSI in 5 min | 35 | 5 min | +| 6 | anomalous_neighbours | empty neighbour list (SI Type 2/2bis/2ter) where baseline shows ≥1 | 15 | none | +| 7 | t3212_out_of_band | T3212 < 1 or > 60 minutes (carrier norm 6-30) | 15 | none | +| 8 | orphan_arfcn | this ARFCN not in this carrier's known frequency plan | 20 | none | + +Final score = sum (clamped to [0, 100]). Each heuristic returns `(contrib, reason_str)`; collected reasons[] are attached to the Alert. + +- [ ] **Step 1: Write `scoring_engine.py`** + +The file structure : + +```python +@dataclasses.dataclass +class HeuristicResult: + name: str + triggered: bool + score_contrib: int + reason: str + + +@dataclasses.dataclass +class CellScore: + cell_id: str + score: int # 0-100 + reasons: list[str] + triggered: list[str] # names of fired heuristics + + +class ScoringEngine: + DEFAULT_THRESHOLDS = { + "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: dict | None = None): + self._baseline = baseline + self._thresholds = thresholds or dict(self.DEFAULT_THRESHOLDS) + # In-memory rolling windows : + self._lac_window: dict[str, deque] = defaultdict(lambda: deque()) # cell_id → deque[(ts, lac)] + self._idreq_window: dict[str, deque] = defaultdict(lambda: deque()) # tmsi_hash → deque[ts] + self._neighbour_baseline: dict[str, set[int]] = {} # cell_id → set of arfcns observed + + def evaluate(self, cell_info, parsed_paging, raw_arfcn, raw_neighbours=None) -> CellScore: + """Run all 8 heuristics ; sum-clamp.""" + results: list[HeuristicResult] = [] + cell_id = _cell_id(cell_info) + baseline_cell = self._baseline.get(cell_id) if cell_id else None + + results.append(self._h_cipher_downgrade(cell_info, baseline_cell)) + results.append(self._h_ghost_bts(cell_id, baseline_cell, parsed_paging)) + results.append(self._h_identity_mismatch(cell_info, raw_arfcn, baseline_cell)) + results.append(self._h_relocalization_storm(cell_id, cell_info)) + results.append(self._h_identity_request_abuse(parsed_paging)) + results.append(self._h_anomalous_neighbours(cell_id, raw_neighbours)) + results.append(self._h_t3212_out_of_band(cell_info)) + results.append(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)) + return CellScore( + cell_id=cell_id or f"arfcn-{raw_arfcn}", + score=total, + reasons=[r.reason for r in results if r.triggered], + triggered=[r.name for r in results if r.triggered], + ) + + def update_thresholds(self, new: dict) -> dict: + """Merge ; return effective thresholds.""" + for k, v in new.items(): + if k in self._thresholds and isinstance(v, dict): + self._thresholds[k].update(v) + return dict(self._thresholds) + + def thresholds(self) -> dict: + return dict(self._thresholds) + + # ── 8 heuristic implementations ──────────────────────────────────── + # Each returns HeuristicResult. Implementation per the table above. + # (full code in plan body — ~150 lines) +``` + +- [ ] **Step 2: Tests — 1 per heuristic + 1 aggregator + 1 threshold-update = 10 tests** + +Test each heuristic individually with synthetic CellInfo + Baseline state. Mock the baseline DB. + +- [ ] **Step 3: Delete the v0.1 scoring.py stub** + +```bash +git rm packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring.py +``` +Grep for any importers ; should be empty (the v0.1 stub was never imported by api/main.py). + +- [ ] **Step 4: Run + commit** + +```bash +python3 -m pytest api/tests/test_scoring_engine.py -v +git commit -am "feat(sentinelle-gsm): scoring_engine — 8 heuristics + thresholds (ref #349)" +``` + +--- + +## Task 5 — Consume loop : L3 decode + scoring + trusted match + alert emission + +**Files:** +- Modify: `api/main.py` +- Create: `api/tests/test_alert_emission.py` + +The v0.3.0 consume loop just upserts a sighting. v0.3.1 also : +1. Decodes the L3 payload +2. Updates sighting metadata when SI3/SI4/SI6 lands +3. Persists paging events with subscriber_hash +4. Feeds baseline.consider() +5. Runs scoring_engine.evaluate() +6. If score ≥ `alert_threshold` : + - Look up each paged hash in `trusted_registry.lookup_by_hash()` + - If ≥1 match : write Alert with `subscriber_hash` + `trusted_label` + - Else : write Alert with no target (anomaly only) + +- [ ] **Step 1: Wire singletons** (similar to Task 4 of v0.3.0 plan) + +```python +_l3: Optional[L3Decode] = None +_scoring: Optional[ScoringEngine] = None +_baseline: Optional[CellBaseline] = None +ALERT_THRESHOLD = int(os.environ.get("SENTINELLE_ALERT_THRESHOLD", 60)) +``` + +- [ ] **Step 2: Rewrite the consume coroutine** + +```python +async def _consume_observations() -> None: + listener = get_listener() + obs_db = get_obs_db() + l3 = _l3 + scoring = _scoring + baseline = _baseline + trusted = get_trusted_registry() + sink = get_alert_sink() + + async for obs in listener.observations(): + cell_id_fallback = f"arfcn-{obs.arfcn}-ch-{obs.channel}" + + parsed = l3.parse(obs.raw_l3) + cell_info = parsed.cell_info + cell_id = _cell_id(cell_info) or cell_id_fallback + + # 1. Upsert sighting + obs_db.upsert_sighting(Sighting( + cell_id=cell_id, + mcc=cell_info.mcc if cell_info else None, + mnc=cell_info.mnc if cell_info else None, + lac=cell_info.lac if cell_info else None, + ci=cell_info.ci if cell_info else None, + arfcn=obs.arfcn, + )) + + # 2. Persist paging events (hashed) + if parsed.paging: + for pid in parsed.paging.identities: + obs_db.record_paging(PagingEvent( + ts=obs.ts, cell_id=cell_id, + subscriber_hash=pid.subscriber_hash, + request_type=("paging-imsi" if pid.id_type == MID_TYPE_IMSI else "paging-tmsi"), + )) + + # 3. Feed baseline + baseline.consider(cell_id, + mcc=cell_info.mcc if cell_info else None, + mnc=cell_info.mnc if cell_info else None, + lac=cell_info.lac if cell_info else None, + arfcn=obs.arfcn, + cipher_a5=cell_info.a5_advertised if cell_info else None) + + # 4. Score + result = scoring.evaluate(cell_info, parsed.paging, obs.arfcn) + if result.score < ALERT_THRESHOLD: + continue + + # 5. Match against trusted + trusted_hit = None + if parsed.paging: + for pid in parsed.paging.identities: + hit = trusted.lookup_by_hash(pid.subscriber_hash) + if hit: + trusted_hit = hit + break + + sink.write(Alert( + cell_id=cell_id, + arfcn=obs.arfcn, + score=result.score, + reason=" + ".join(result.reasons[:3]), # truncate for UI + subscriber_hash=(trusted_hit.imsi_hash if trusted_hit else None), + trusted_label=(trusted_hit.label if trusted_hit else None), + )) +``` + +- [ ] **Step 3: New endpoints** + +```python +@app.get("/baseline") +def list_baseline(limit: int = 200): ... + +@app.post("/baseline/learn") +def baseline_learn(body: BaselineLearnBody): ... + # body.sweep_seconds: int (default 300) + # if no scan currently running, raises 409 with hint to start one first + # else flips baseline into learn_mode for `sweep_seconds` + +@app.get("/scoring/thresholds") +def scoring_thresholds_get(): ... + +@app.post("/scoring/thresholds") +def scoring_thresholds_set(body: dict): ... + # writes an audit-log line via logger.info (caught by /journal/stream) +``` + +- [ ] **Step 4: Integration tests in `test_alert_emission.py`** + +5 tests covering : +- Score below threshold → no alert +- Score above threshold + no trusted match → anomaly-only alert +- Score above threshold + trusted match → alert with trusted_label +- Paging event persistence with hash +- Audit log entry on POST /scoring/thresholds + +- [ ] **Step 5: Run + commit** + +```bash +python3 -m pytest api/tests/test_alert_emission.py -v +git commit -am "feat(sentinelle-gsm): consume loop wires L3 + scoring + trusted match → Alert (ref #349)" +``` + +--- + +## Task 6 — WebUI Baseline + Scoring panels + +**Files:** +- Modify: `www/sentinelle/index.html`, `sentinelle.css`, `sentinelle.js` + +- [ ] **Step 1: HTML new panels** + +After "Observations", before "Actions" : + +**`#baseline` panel** — table : Cell ID, MCC, MNC, LAC, ARFCN, Cipher A5, Learn count, Last learned. Buttons : "Refresh", "Start baseline learn (5 min)". + +**`#scoring` panel** — 8 rows, one per heuristic. Per row : name + enabled toggle + score input + reset-to-default button. Footer : "Apply changes" + "Reset all to defaults". + +Alert rows in `#alerts` panel — render `trusted_label` chip when present (violet pill). + +- [ ] **Step 2: JS** + +New helpers : `loadBaseline()`, `startBaselineLearn()`, `loadThresholds()`, `saveThresholds()`. Poll baseline every 30 s while a scan is running. + +The alert row formatter is extended to render `${trusted_label}` when present. + +- [ ] **Step 3: CSS — `.trusted-chip` (violet pill), `.threshold-row` (grid 5-col), `.score-input` (small numeric input)** + +- [ ] **Step 4: Commit** + +```bash +git commit -am "feat(sentinelle-gsm): webui baseline + scoring panels + trusted_label chip (ref #349)" +``` + +--- + +## Task 7 — Privacy invariant tests + changelog 0.3.1 + +- [ ] **Step 1: Add 4 new privacy tests** + +```python +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"} + assert names.isdisjoint(forbidden), f"{cls.__name__} has plaintext id fields" + + +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 with a known TMSI + tmsi = b"\xDE\xAD\xBE\xEF" + mid = b"\x05\xF4" + tmsi + raw = b"\x06\x21\x00" + mid + frame = dec.parse(raw) + pid = frame.paging.identities[0] + # Plaintext bytes MUST NOT appear in the hash output + 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"} + assert names.isdisjoint(forbidden) + + +def test_scoring_engine_consumes_hashes_only(): + """The scoring engine's HeuristicResult.reason is a free-text string + but MUST NOT include any 15-digit shape from input.""" + # ... shape check on reason strings produced by each heuristic with + # synthetic CellInfo containing no plaintext anywhere +``` + +- [ ] **Step 2: Run full sweep** — expect 48 + 6 (l3_decode) + 5 (baseline) + 10 (scoring) + 5 (alert_emission) + 4 (privacy) = **78 passing**. + +- [ ] **Step 3: Changelog bump** + +```text +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: 30 new (l3_decode ×6, baseline ×5, scoring_engine ×10, + alert_emission ×5, privacy ×4); full sweep 78/78 passing. + * Closes #349. Refs #237. + + -- Gerald Kerma Fri, 22 May 2026 16:30:00 +0000 +``` + +- [ ] **Step 4: Commit** + +```bash +git commit -am "chore(sentinelle-gsm): bump 0.3.1 + extend privacy invariant tests (closes #349)" +``` + +--- + +## Task 8 — Build .deb + open PR + +```bash +cd packages/secubox-sentinelle-gsm +dpkg-buildpackage -us -uc -b +``` + +Verify .deb ships : `l3_decode.py`, `baseline.py`, `scoring_engine.py` ; does NOT ship the dropped `scoring.py` stub. + +Open PR with the body summarising commits, the new modules, the trusted-label flow, the 8 heuristics, the privacy invariants extension, and the 78/78 test count. Closes #349. + +--- + +## Task 9 — On-board E2E (manual, after PR merge) + +1. Deploy 0.3.1 .deb to gk2. +2. Verify `systemctl is-active secubox-sentinelle-gsm` → active. +3. Add a trusted phone with the operator's own IMSI hash via the webui. +4. POST `/baseline/learn { sweep_seconds: 300 }` AND start a scan ; let it run 5 min. +5. Verify GET `/baseline` populated with at least one cell. +6. Stop scan ; modify the operator's phone IMSI hash slightly (simulate a paging targeting it) ; restart scan ; force an alert via the test injection path (paged hash that matches). +7. Verify : alert with `trusted_label` chip appears in /sentinelle/ live feed, browser desktop notification fires. +8. Adjust scoring thresholds via the UI ; verify journal log line records the change. +9. Comment on #349 with "v0.3.1 validated, closing". diff --git a/packages/secubox-sentinelle-gsm/api/main.py b/packages/secubox-sentinelle-gsm/api/main.py index 13042297..6680995c 100644 --- a/packages/secubox-sentinelle-gsm/api/main.py +++ b/packages/secubox-sentinelle-gsm/api/main.py @@ -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", diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_alert_emission.py b/packages/secubox-sentinelle-gsm/api/tests/test_alert_emission.py new file mode 100644 index 00000000..6fe3cc7f --- /dev/null +++ b/packages/secubox-sentinelle-gsm/api/tests/test_alert_emission.py @@ -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 +# 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 + ) diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_baseline.py b/packages/secubox-sentinelle-gsm/api/tests/test_baseline.py new file mode 100644 index 00000000..e1cc99a0 --- /dev/null +++ b/packages/secubox-sentinelle-gsm/api/tests/test_baseline.py @@ -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 + +""" +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) diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_gsmtap_listener.py b/packages/secubox-sentinelle-gsm/api/tests/test_gsmtap_listener.py index db907671..1257fe9e 100644 --- a/packages/secubox-sentinelle-gsm/api/tests/test_gsmtap_listener.py +++ b/packages/secubox-sentinelle-gsm/api/tests/test_gsmtap_listener.py @@ -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() diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_l3_decode.py b/packages/secubox-sentinelle-gsm/api/tests/test_l3_decode.py new file mode 100644 index 00000000..707f396d --- /dev/null +++ b/packages/secubox-sentinelle-gsm/api/tests/test_l3_decode.py @@ -0,0 +1,213 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma + +"""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 diff --git a/packages/secubox-sentinelle-gsm/api/tests/test_scoring_engine.py b/packages/secubox-sentinelle-gsm/api/tests/test_scoring_engine.py new file mode 100644 index 00000000..7d17982b --- /dev/null +++ b/packages/secubox-sentinelle-gsm/api/tests/test_scoring_engine.py @@ -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 + +""" +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 diff --git a/packages/secubox-sentinelle-gsm/debian/changelog b/packages/secubox-sentinelle-gsm/debian/changelog index b2c47c83..863fec17 100644 --- a/packages/secubox-sentinelle-gsm/debian/changelog +++ b/packages/secubox-sentinelle-gsm/debian/changelog @@ -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 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 diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/baseline.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/baseline.py new file mode 100644 index 00000000..ef325a2b --- /dev/null +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/baseline.py @@ -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 +# 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] diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/gsmtap_listener.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/gsmtap_listener.py index aa33417f..39d4b6b5 100644 --- a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/gsmtap_listener.py +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/gsmtap_listener.py @@ -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: diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/l3_decode.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/l3_decode.py new file mode 100644 index 00000000..efa416e4 --- /dev/null +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/l3_decode.py @@ -0,0 +1,238 @@ +# SPDX-License-Identifier: LicenseRef-CMSD-1.0 +# Copyright (c) 2026 CyberMind — Gerald Kerma +# Source-Disclosed License — All rights reserved except as expressly granted. +# See LICENCE-CMSD-1.0.md for terms. + +""" +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) diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/observations.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/observations.py index 9ba0a103..b63ce529 100644 --- a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/observations.py +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/observations.py @@ -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: diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring.py deleted file mode 100644 index 67cd7123..00000000 --- a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring.py +++ /dev/null @@ -1,73 +0,0 @@ -# SPDX-License-Identifier: LicenseRef-CMSD-1.0 -# Copyright (c) 2026 CyberMind — Gerald Kerma -# Source-Disclosed License — All rights reserved except as expressly granted. -# See LICENCE-CMSD-1.0.md for terms. - -""" -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, - ) diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring_engine.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring_engine.py new file mode 100644 index 00000000..c7ea3982 --- /dev/null +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/scoring_engine.py @@ -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 +# 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) diff --git a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/trusted.py b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/trusted.py index b4ae8fd2..d92ef7f2 100644 --- a/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/trusted.py +++ b/packages/secubox-sentinelle-gsm/lib/sentinelle_gsm/trusted.py @@ -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, diff --git a/packages/secubox-sentinelle-gsm/tests/test_privacy_invariant.py b/packages/secubox-sentinelle-gsm/tests/test_privacy_invariant.py index 28f7086a..45a56b44 100644 --- a/packages/secubox-sentinelle-gsm/tests/test_privacy_invariant.py +++ b/packages/secubox-sentinelle-gsm/tests/test_privacy_invariant.py @@ -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}" diff --git a/packages/secubox-sentinelle-gsm/www/sentinelle/index.html b/packages/secubox-sentinelle-gsm/www/sentinelle/index.html index 02d9f2ce..797b3ddd 100644 --- a/packages/secubox-sentinelle-gsm/www/sentinelle/index.html +++ b/packages/secubox-sentinelle-gsm/www/sentinelle/index.html @@ -38,6 +38,8 @@ Live logs Scan control Observations + Baseline + Scoring Actions
@@ -229,6 +231,50 @@
+ +
+
+

Operator baseline 0

+
+ + +
+
+
+ + + + + + + + + + +
Cell IDMCCMNCLACARFCNCipherLearn countLast learned
no baseline yet — start a scan + click "Start learn"
+
+

+ Cells graduate to baseline after 3 sightings under normal scan. The "Start learn" button accepts every cell observed within the next 5 minutes immediately. +

+
+ + +
+
+

Scoring heuristics

+
+ + +
+
+
+ +
+

+ 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. +

+
+
diff --git a/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.css b/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.css index 83251d39..a1cd8043 100644 --- a/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.css +++ b/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.css @@ -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; } } diff --git a/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.js b/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.js index 7dd75732..e59d2de4 100644 --- a/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.js +++ b/packages/secubox-sentinelle-gsm/www/sentinelle/sentinelle.js @@ -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 - ? `${escapeHtml(alert.trusted_label)}` + ? `${escapeHtml(alert.trusted_label)}` : '—'; tr.innerHTML = `${escapeHtml(fmtTime(alert.ts))}` + @@ -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 = + `${escapeHtml(c.cell_id)}` + + `${escapeHtml(fmt(c.mcc))}` + + `${escapeHtml(fmt(c.mnc))}` + + `${escapeHtml(fmt(c.lac))}` + + `${escapeHtml(fmt(c.arfcn))}` + + `${escapeHtml(fmt(c.cipher_a5))}` + + `${escapeHtml(fmt(c.learn_count))}` + + `${escapeHtml(fmtDateTime(c.last_learned))}`; + return tr; + } + + function renderBaselineTable(cells) { + if (!els.baselineTbody) return; + els.baselineTbody.innerHTML = ""; + if (!cells || cells.length === 0) { + els.baselineTbody.innerHTML = + 'no baseline yet — start a scan + click "Start learn"'; + 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 = '
no scoring heuristics defined
'; + 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 = + `${escapeHtml(name)}` + + `` + + ``; + 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); }