From c48db03afb3b94329ffe208709a53dd50de9bed1 Mon Sep 17 00:00:00 2001 From: Louis King Date: Tue, 23 Jun 2026 00:11:39 +0100 Subject: [PATCH] feat(spam): score messages at ingest and hide likely spam MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add an optional, off-by-default spam-detection feature that scores each message's spam likelihood at ingest, stores the score on the row, and lets the display layer hide likely-spam by default behind a "show potential spam" toggle. Nothing is ever dropped at ingest, so the threshold can be retuned without reprocessing. Scoring (collector/spam.py): windowed COUNT(*) over new (path_prefix, received_at) and (sender_normalized, received_at) indexes — joint path+sender signal plus a sender-name signal (trailing-digit suffix stripped so bob1/bob2 collapse to bob). When the path is short/zero-hop or absent, the name signal stands alone at full weight so local spam is still flaggable. A background sweep re-scores recent rows with hindsight to catch the leading edge of bursts. The collector logs each score (WARNING at/above the threshold). Display: the messages API gains include_spam and a master-switch-aware hide-filter; the SPA shows the toggle + a badge only when the feature is on. Config: FEATURE_SPAM_DETECTION is the single operator switch, bridged in Compose to the backend SPAM_DETECTION_ENABLED for collector + api (mirrors the FEATURE_PACKETS / RAW_PACKET_CAPTURE_ENABLED pattern). Both default off. Works on SQLite and Postgres: DB-agnostic queries, an Alembic batch migration for the three new columns + two indexes, and backend-aware collector test fixtures (lifted db_backend/db_url into the shared conftest). Also: move the meshcore-hub image pull_policy out of the base compose file. It lived in docker-compose.yml as pull_policy: daily and made `make up` pull the published image over a freshly built local one. Base is now policy-neutral (default missing); dev sets pull_policy: build on the hub services so it only ever uses local builds. Prod refreshes images via a manual `docker compose ... pull`. Co-Authored-By: Claude Opus 4.8 --- .env.example | 36 ++ SCHEMAS.md | 5 + ...fc_add_spam_scoring_columns_to_messages.py | 52 +++ docker-compose.dev.yml | 12 + docker-compose.yml | 24 +- docs/configuration.md | 20 + .../20260622-2243-spam-detection/plan.md | 432 ++++++++++++++++++ src/meshcore_hub/api/app.py | 6 + src/meshcore_hub/api/routes/messages.py | 18 + .../collector/handlers/message.py | 66 ++- .../collector/letsmesh_normalizer.py | 27 ++ src/meshcore_hub/collector/spam.py | 315 +++++++++++++ src/meshcore_hub/collector/subscriber.py | 62 +++ src/meshcore_hub/common/config.py | 72 +++ src/meshcore_hub/common/models/message.py | 26 ++ src/meshcore_hub/common/schemas/messages.py | 4 + .../web/static/js/spa/pages/messages.js | 47 +- src/meshcore_hub/web/static/locales/en.json | 5 + tests/conftest.py | 101 +++- tests/test_api/conftest.py | 126 ++--- tests/test_api/test_messages.py | 70 +++ tests/test_collector/conftest.py | 42 +- .../test_handlers/test_message_spam.py | 97 ++++ tests/test_collector/test_spam.py | 298 ++++++++++++ 24 files changed, 1838 insertions(+), 125 deletions(-) create mode 100644 alembic/versions/20260622_2243_38abdf4651fc_add_spam_scoring_columns_to_messages.py create mode 100644 docs/plans/20260622-2243-spam-detection/plan.md create mode 100644 src/meshcore_hub/collector/spam.py create mode 100644 tests/test_collector/test_handlers/test_message_spam.py create mode 100644 tests/test_collector/test_spam.py diff --git a/.env.example b/.env.example index 194b184..06b2c27 100644 --- a/.env.example +++ b/.env.example @@ -279,6 +279,37 @@ DATA_RETENTION_INTERVAL_HOURS=24 # enabled, so disabling capture lets existing rows drain. # RAW_PACKET_RETENTION_DAYS=7 +# ------------------- +# Spam Detection +# ------------------- +# Score each message's spam likelihood at ingest and hide likely-spam by default +# (with a "show potential spam" toggle on the Messages page). Nothing is ever +# dropped; the score is stored and the display layer filters on it. +# +# Off by default. FEATURE_SPAM_DETECTION is the single switch operators set: in +# Docker Compose it drives the backend SPAM_DETECTION_ENABLED for the collector +# (scoring + sweep) and the api (hide-filter), and exposes the UI toggle. +# FEATURE_SPAM_DETECTION=false + +# Backend operational switch, read by the collector + api. Compose derives it from +# FEATURE_SPAM_DETECTION (SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION}); set it +# directly only when running the services without Compose. +# SPAM_DETECTION_ENABLED=false + +# Score at/above which a message is treated as likely spam (hidden by default in +# the API, logged at WARNING by the collector). Read by the collector + api. +# SPAM_SCORE_THRESHOLD=0.6 + +# Scoring tuning (collector only; only consulted when detection is enabled). +# SPAM_WINDOW_SECONDS=300 # sliding window for frequency counts +# SPAM_PATH_HOPS=3 # leading origin-side hops that form the prefix +# SPAM_MIN_PATH_HOPS=5 # min path_len before the path signal applies +# SPAM_PATH_THRESHOLD=5 # joint path+sender count that saturates the path signal +# SPAM_NAME_THRESHOLD=5 # sender count that saturates the name signal +# SPAM_WEIGHT_PATH=0.7 # weight of the path signal +# SPAM_WEIGHT_NAME=0.3 # weight of the name signal +# SPAM_RESCORE_INTERVAL_SECONDS=120 # background re-scoring sweep cadence (0 disables) + # ------------------- # Node Cleanup Settings # ------------------- @@ -554,6 +585,11 @@ SYSTEM_MAINTENANCE=false # Packets page is ON by default. This var also drives raw-packet capture # on the collector via Compose (RAW_PACKET_CAPTURE_ENABLED=${FEATURE_PACKETS}). # FEATURE_PACKETS=true +# Spam detection is OFF by default. This var also drives the backend scoring + +# hide switch on the collector/api via Compose +# (SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION}). See the Spam Detection +# section above for the scoring tuning vars. +# FEATURE_SPAM_DETECTION=false # ------------------- # Contact Information diff --git a/SCHEMAS.md b/SCHEMAS.md index a5b9d06..1c0306d 100644 --- a/SCHEMAS.md +++ b/SCHEMAS.md @@ -189,6 +189,11 @@ Group/broadcast messages on specific channels. - Additional channel labels are loaded from the `channels` database table via the collector's periodic refresh. - When decoder output includes a human sender (`payload.decoded.decrypted.sender`), message text is normalized to `Name: Message`; sender identity remains unknown when only hash/prefix metadata is available. +**Spam scoring columns** (stored on `messages`, populated only when `SPAM_DETECTION_ENABLED` is set; otherwise null): +- `path_prefix` (`string(48)`): first `SPAM_PATH_HOPS` origin-side hop hashes joined (e.g. `"16,69,23"`), derived from the decoded `path`. Stored null when `path_len` is below `SPAM_MIN_PATH_HOPS` so short local-mesh paths don't pollute path counts. Indexed with `received_at`. +- `sender_normalized` (`string(255)`): the self-declared channel sender name (the part before the first `": "` in the body) lower-cased with the trailing digit/space suffix stripped (`bob17` → `bob`). Indexed with `received_at`. +- `spam_score` (`float`): likely-spam score `0.0`–`1.0` computed at ingest from windowed joint `(path_prefix, sender_normalized)` and `sender_normalized` frequency counts, refined by a background re-scoring sweep. Messages at/above `SPAM_SCORE_THRESHOLD` are hidden from `GET /api/v1/messages` unless `include_spam=true`. + **Compatibility ingest note (advertisements)**: - In LetsMesh upload compatibility mode, `status` feed payloads are persisted as informational `letsmesh_status` events and are not normalized to `ADVERTISEMENT`. - In LetsMesh upload compatibility mode, decoded payload type `4` is normalized to `ADVERTISEMENT` when node identity metadata is present. diff --git a/alembic/versions/20260622_2243_38abdf4651fc_add_spam_scoring_columns_to_messages.py b/alembic/versions/20260622_2243_38abdf4651fc_add_spam_scoring_columns_to_messages.py new file mode 100644 index 0000000..4d997d2 --- /dev/null +++ b/alembic/versions/20260622_2243_38abdf4651fc_add_spam_scoring_columns_to_messages.py @@ -0,0 +1,52 @@ +"""add spam scoring columns to messages + +Revision ID: 38abdf4651fc +Revises: d4e5f6a7b8c9 +Create Date: 2026-06-22 22:43:00.000000+00:00 + +Adds the three nullable spam-scoring columns (path_prefix, sender_normalized, +spam_score) and their two composite ``(*, received_at)`` indexes used by the +windowed COUNT(*) scorer. Uses batch mode so it applies on SQLite (which lacks +full ALTER TABLE) as well as Postgres. + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "38abdf4651fc" +down_revision: Union[str, None] = "d4e5f6a7b8c9" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + with op.batch_alter_table("messages", schema=None) as batch_op: + batch_op.add_column( + sa.Column("path_prefix", sa.String(length=48), nullable=True) + ) + batch_op.add_column( + sa.Column("sender_normalized", sa.String(length=255), nullable=True) + ) + batch_op.add_column(sa.Column("spam_score", sa.Float(), nullable=True)) + batch_op.create_index( + "ix_messages_path_prefix_received_at", + ["path_prefix", "received_at"], + unique=False, + ) + batch_op.create_index( + "ix_messages_sender_normalized_received_at", + ["sender_normalized", "received_at"], + unique=False, + ) + + +def downgrade() -> None: + with op.batch_alter_table("messages", schema=None) as batch_op: + batch_op.drop_index("ix_messages_sender_normalized_received_at") + batch_op.drop_index("ix_messages_path_prefix_received_at") + batch_op.drop_column("spam_score") + batch_op.drop_column("sender_normalized") + batch_op.drop_column("path_prefix") diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index ff217a5..9c2194b 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -7,7 +7,11 @@ # docker compose -f docker-compose.yml -f docker-compose.dev.yml up -d services: + # Hub services (collector/api/web/migrate/seed) build from local source and + # never pull the published image in development (pull_policy: build), so + # `make build` + `make up` cannot clobber a local build with a registry pull. collector: + pull_policy: build depends_on: migrate: condition: service_completed_successfully @@ -19,13 +23,21 @@ services: - "${MQTT_PORT:-1883}:${MQTT_PORT:-1883}" api: + pull_policy: build ports: - "${API_PORT:-8000}:8000" web: + pull_policy: build ports: - "${WEB_PORT:-8080}:8080" + migrate: + pull_policy: build + + seed: + pull_policy: build + redis: ports: - "${REDIS_PORT:-6379}:6379" diff --git a/docker-compose.yml b/docker-compose.yml index f5c9120..4ad4cd7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -182,7 +182,6 @@ services: # ========================================================================== collector: image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest} - pull_policy: daily build: context: . dockerfile: Dockerfile @@ -240,6 +239,18 @@ services: # capture and the web Packets page). Retention defaults to 7 days. - RAW_PACKET_CAPTURE_ENABLED=${FEATURE_PACKETS:-true} - RAW_PACKET_RETENTION_DAYS=${RAW_PACKET_RETENTION_DAYS:-7} + # Spam detection (derived from FEATURE_SPAM_DETECTION so one var drives the + # backend switch and the web toggle together). Off by default. + - SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION:-false} + - SPAM_SCORE_THRESHOLD=${SPAM_SCORE_THRESHOLD:-0.6} + - SPAM_WINDOW_SECONDS=${SPAM_WINDOW_SECONDS:-300} + - SPAM_PATH_HOPS=${SPAM_PATH_HOPS:-3} + - SPAM_MIN_PATH_HOPS=${SPAM_MIN_PATH_HOPS:-5} + - SPAM_PATH_THRESHOLD=${SPAM_PATH_THRESHOLD:-5} + - SPAM_NAME_THRESHOLD=${SPAM_NAME_THRESHOLD:-5} + - SPAM_WEIGHT_PATH=${SPAM_WEIGHT_PATH:-0.7} + - SPAM_WEIGHT_NAME=${SPAM_WEIGHT_NAME:-0.3} + - SPAM_RESCORE_INTERVAL_SECONDS=${SPAM_RESCORE_INTERVAL_SECONDS:-120} command: ["collector"] healthcheck: test: ["CMD", "meshcore-hub", "health", "collector"] @@ -253,7 +264,6 @@ services: # ========================================================================== api: image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest} - pull_policy: daily build: context: . dockerfile: Dockerfile @@ -304,6 +314,10 @@ services: - REDIS_KEY_PREFIX=${REDIS_KEY_PREFIX:-hub} - REDIS_CACHE_TTL=${REDIS_CACHE_TTL:-30} - REDIS_CACHE_TTL_DASHBOARD=${REDIS_CACHE_TTL_DASHBOARD:-30} + # Spam detection: hide-filter switch + threshold (bridged from + # FEATURE_SPAM_DETECTION so it tracks the collector and web toggle). + - SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION:-false} + - SPAM_SCORE_THRESHOLD=${SPAM_SCORE_THRESHOLD:-0.6} command: ["api"] healthcheck: test: @@ -323,7 +337,6 @@ services: # ========================================================================== web: image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest} - pull_policy: daily build: context: . dockerfile: Dockerfile @@ -397,6 +410,9 @@ services: - FEATURE_RADIO_CONFIG=${FEATURE_RADIO_CONFIG:-true} # Packets page is on by default; this also drives collector capture. - FEATURE_PACKETS=${FEATURE_PACKETS:-true} + # Spam detection toggle is off by default; this also drives the backend + # scoring/hide switch on the collector and api. + - FEATURE_SPAM_DETECTION=${FEATURE_SPAM_DETECTION:-false} command: ["web"] healthcheck: test: @@ -416,7 +432,6 @@ services: # ========================================================================== migrate: image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest} - pull_policy: daily build: context: . dockerfile: Dockerfile @@ -453,7 +468,6 @@ services: # ========================================================================== seed: image: ghcr.io/ipnet-mesh/meshcore-hub:${IMAGE_VERSION:-latest} - pull_policy: daily build: context: . dockerfile: Dockerfile diff --git a/docs/configuration.md b/docs/configuration.md index 7c173b2..6f11e79 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -126,6 +126,25 @@ The collector automatically cleans up old event data and inactive nodes. Retenti | `RAW_PACKET_CAPTURE_ENABLED` | `false` | Capture raw packets into `raw_packets`. In Compose, derived from `FEATURE_PACKETS` — see [letsmesh.md → Raw Packet Capture](letsmesh.md#raw-packet-capture) | | `RAW_PACKET_RETENTION_DAYS` | `7` | Days to retain raw packets (independent of `DATA_RETENTION_DAYS`) | +## Spam Detection + +Scores each message's spam likelihood at ingest, stores the score on the message row, and hides likely-spam by default in the API (with a "show potential spam" toggle on the Messages page). Nothing is ever dropped — the design is reversible and the threshold can be retuned without reprocessing. + +Off by default. `FEATURE_SPAM_DETECTION` is the single switch operators set: in Compose it drives the backend `SPAM_DETECTION_ENABLED` for the **collector** (scoring + the background re-scoring sweep) and the **api** (the hide-filter), and exposes the UI toggle (see [Feature Flags](#feature-flags)). Set `SPAM_DETECTION_ENABLED` directly only when running services without Compose. + +| Variable | Default | Read by | Description | +| --- | --- | --- | --- | +| `SPAM_DETECTION_ENABLED` | `false` | collector, api | Operational switch for scoring + hiding. In Compose, derived from `FEATURE_SPAM_DETECTION` (`SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION}`) | +| `SPAM_SCORE_THRESHOLD` | `0.6` | collector, api | Score at/above which a message is treated as likely spam (hidden by default; logged at `WARNING`) | +| `SPAM_WINDOW_SECONDS` | `300` | collector | Sliding window (seconds) for frequency counts | +| `SPAM_PATH_HOPS` | `3` | collector | Leading origin-side hops that form the `path_prefix` | +| `SPAM_MIN_PATH_HOPS` | `5` | collector | Minimum `path_len` before the path signal applies (short local-mesh paths share prefixes) | +| `SPAM_PATH_THRESHOLD` | `5` | collector | Joint `(path_prefix, sender)` count that saturates the path signal | +| `SPAM_NAME_THRESHOLD` | `5` | collector | Sender count that saturates the name signal | +| `SPAM_WEIGHT_PATH` | `0.7` | collector | Weight of the path signal in the combined score | +| `SPAM_WEIGHT_NAME` | `0.3` | collector | Weight of the name signal in the combined score | +| `SPAM_RESCORE_INTERVAL_SECONDS` | `120` | collector | Background re-scoring sweep cadence in seconds (`0` disables the sweep) | + ## API REST API server. For multi-worker scaling guidance (`API_WORKERS`), see [deployment.md → Scaling the API](deployment.md#scaling-the-api). @@ -192,6 +211,7 @@ Control which pages are visible in the web dashboard. Disabled features are full | `FEATURE_CHANNELS` | `true` | Enable the `/channels` page | | `FEATURE_RADIO_CONFIG` | `true` | Show radio config panel on home page | | `FEATURE_PACKETS` | `true` | Enable the `/packets` raw-packet browser. In Compose this also drives `RAW_PACKET_CAPTURE_ENABLED` on the collector | +| `FEATURE_SPAM_DETECTION` | `false` | Show the "show potential spam" toggle on `/messages`. In Compose this also drives `SPAM_DETECTION_ENABLED` on the collector + api — see [Spam Detection](#spam-detection) | **Dependencies:** Dashboard auto-disables when all of Nodes/Advertisements/Messages are disabled. Map auto-disables when Nodes is disabled. Members auto-disables when OIDC is disabled (set via `OIDC_ENABLED`). diff --git a/docs/plans/20260622-2243-spam-detection/plan.md b/docs/plans/20260622-2243-spam-detection/plan.md new file mode 100644 index 0000000..1d97207 --- /dev/null +++ b/docs/plans/20260622-2243-spam-detection/plan.md @@ -0,0 +1,432 @@ +# Spam Detection / Message Scoring at Ingest + +**Date**: 2026-06-22 +**Status**: Design settled, ready for implementation +**Branch**: `feat/spam-detection` + +## Problem Statement + +A user is flooding the mesh network with near-identical messages. The sender name +varies slightly each time (incrementing suffixes, e.g. `bob1`, `bob2`), so naive +name-blocking does not work. The message content is currently trivial (`"test"`) +but is trivially changeable, so content-blocking is also unreliable. + +The goal is to **flag likely-spam messages with a score at ingest time, store that +score on the message row, and let the display layer filter on it** (hide by +default, with a "show potential spam" toggle). We never drop or block at ingest — +the design stays reversible, so the threshold can be retuned later without +reprocessing anything. + +## Research Findings + +The original design assumed messages carried a sender *name* and a *path prefix* +(first few route hops). Exploration of the codebase shows the real data model +differs in ways that shape the implementation: + +### Sender identity + +- Messages do **not** store a human sender name. They store `pubkey_prefix` + (12-char public-key prefix) — `models/message.py:42`. +- A human `sender_name` *is* resolved during normalization + (`letsmesh_normalizer.py:172`, via `_extract_letsmesh_decoder_sender` + + `_normalize_sender_name`) but is currently only used to prefix the message text, + not persisted. +- **On public channels (where the spam lives) there is no cryptographic sender + identity.** Channel messages are encrypted with a shared channel key, so + `pubkey_prefix` is typically null and the sender name is *self-declared* — it is + literally part of the decrypted body, formatted `": "` + (confirmed: name then a colon-space `": "` delimiter). That is exactly why the + spammer can trivially rotate `bob1`/`bob2`. The normalizer already extracts that + name into the payload's `sender_name`, so `sender_normalized = + normalize_sender(payload["sender_name"])`; if ever absent it can be recovered by + splitting the stored `text` on the first `": "`. A future content-fingerprint + signal must strip everything up to and including that `": "` before comparing. + +### Route / path + +- Messages store only `path_len` — a hop **count** integer (`models/message.py:54`), + **not** the route itself. +- However, the ordered route **is available at ingest** in the decoded packet as a + top-level `decoded.path` list of hop hashes. It is simply discarded for message + packets. The trace handler already keeps it as `path_hashes` + (`letsmesh_normalizer.py:287`), and `api/routes/packet_groups.py:36` + `_extract_path_hashes()` documents the exact location (`decoded.get("path")`, + with `decoded.payload.decoded.pathHashes` as the trace-style fallback). +- The path is the **strongest signal**: it reflects the real relay topology and is + far harder for a spammer to fake than a name or message body. +- **Two ordering/determinism caveats** that affect how the prefix is taken: + 1. *Direction.* We must take the **origin-side** hops. A spammer at a fixed + location reaches the same repeaters that can physically hear it, so the first + hop(s) are effectively a **location fingerprint** the spammer can't change by + editing the name or body — this is why path is the strongest signal. + Origin-side selection is also what makes the stored prefix **deterministic + across observers**: the receiver-side hops diverge per observer (and dedup + keeps only the first-arriving observer's row), but the origin-side hops are + shared by all observers, so the key is stable regardless of which observer + wins. The hop list is append-order as the packet is relayed (each repeater + appends its hash), so index 0 is closest to the origin and `path_hashes[:N]` + is the origin-side prefix. **Confirmed origin-first** — the Packets page + renders these paths from origin toward observer using the same `decoded.path` + ordering. + 2. *Cross-observer dedup.* One logical message received by several observers is + deduped to a **single** `messages` row written by whichever observer inserted + first, and each observer received the message via a *different* route. So the + stored `path_prefix` is nondeterministic at the receiver end. Taking only the + first 2–3 *origin-side* hops keeps it as stable as possible, but the prefix is + inherently a little noisy — which is why `sender_normalized` is kept as an + independent secondary signal rather than folding everything into the path. + +### Infrastructure + +- The **collector has no Redis**. Redis exists only in the API service for + response caching (`common/redis.py`, used via `@cached` in the API routes). +- The `messages` table is already indexed on `received_at` + (`ix_messages_received_at`). Adding a composite `(path_prefix, received_at)` + index makes a windowed `COUNT(*)` trivially cheap at mesh volumes. +- Migrations use **Alembic** with autogenerate; an example add-column migration is + `alembic/versions/20260515_1920_add_route_type_advert_timestamp.py`. CLI: + `meshcore-hub db revision --autogenerate -m "…"` then `meshcore-hub db upgrade`. +- The display path is: API `api/routes/messages.py::list_messages` (cached) → + schema `common/schemas/messages.py::MessageRead` → SPA + `web/static/js/spa/pages/messages.js` (lit-html), which renders filters via + `renderFilterCard()` in `components.js`. + +## Design Decisions + +| Decision | Choice | Rationale | +|----------|--------|-----------| +| Sliding-window counter store | **Postgres-direct windowed `COUNT(*)`** | Collector has no Redis; mesh volume is low; the `(path_prefix, received_at)` index makes counts cheap. One source of truth, no rebuild-on-restart logic. | +| Primary signal | **Joint `(path_prefix, sender_normalized)`** frequency in window | The real spam signature is "same route **and** same normalized sender, repeatedly." A joint count means a busy *legitimate* path carrying many *different* senders does not score — only concentrated repetition does. | +| Secondary signal | **`sender_normalized`** alone (name with trailing digits stripped) frequency | Robust fallback that catches `bob1`/`bob2` rotation even when the stored `path_prefix` is noisy/null (see cross-observer path note). Weighted lower. | +| Content signal | **Out of scope (v1)** | Content is trivial/spoofable now; shingling/fingerprinting deferred to a future iteration. | +| Block vs. score | **Score, never block** | Reversible; threshold retunable without reprocessing. Filtering happens at display time. | +| Threshold location | **Env var `SPAM_SCORE_THRESHOLD`** (default 0.6) | Retune without code change/redeploy. | +| Short-path handling | **Minimum-path-length gate `SPAM_MIN_PATH_HOPS`** (default 5) | Short paths (`<5` hops) in local meshes legitimately share the same prefix; below the gate the path signal is disabled and `path_prefix` is stored null so these rows don't pollute path counts. | +| Historical backfill | **None (v1)** | Backfill would require re-decoding `raw_packets`. Window is minutes, so counts self-heal as new messages arrive. | +| Database support | **Both SQLite and Postgres** | Project still ships SQLite as the default. All work stays DB-agnostic: SQLAlchemy Core/ORM for queries, Alembic batch mode for the migration (no Postgres-only DDL), and a **Python-computed cutoff datetime** for the window — never SQL `NOW() - INTERVAL` (which differs between engines). | +| Master on/off switch | **`FEATURE_SPAM_DETECTION` (UI) bridged to `SPAM_DETECTION_ENABLED` (backend), both default `false`** | Mirrors the existing **raw-packets** pattern exactly. `FEATURE_SPAM_DETECTION` (config `feature_spam_detection`) feeds the `features` dict the frontend reads (like `FEATURE_PACKETS`) and gates the UI toggle. `SPAM_DETECTION_ENABLED` (config `spam_detection_enabled`) is the operational switch the **collector + API** read; when off the collector does no scoring (null `spam_score`), the sweep doesn't run, and the API applies no hide-filter. Compose bridges them — `SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION:-false}` (cf. `RAW_PACKET_CAPTURE_ENABLED=${FEATURE_PACKETS}`) — so operators set **one** var. Keeping two avoids making backend services depend on a UI "feature" flag. | + +## Combined Design + +### 1. Database model (`models/message.py`) + +Add three nullable columns (backward compatible with existing rows) and two +composite indexes: + +- `path_prefix: Mapped[str | None]` — `String(48)`, first N hop hashes joined + (e.g. `"16,69,23"`); null below the path-length gate. +- `sender_normalized: Mapped[str | None]` — `String(255)`, lower-cased name with + trailing digits stripped. +- `spam_score: Mapped[float | None]` — `Float`, 0.0–1.0. +- `Index("ix_messages_path_prefix_received_at", "path_prefix", "received_at")` +- `Index("ix_messages_sender_normalized_received_at", "sender_normalized", "received_at")` + +The Alembic migration must use **batch mode** for the `add_column` operations so it +works on SQLite (which lacks full `ALTER TABLE`) as well as Postgres — the project's +`alembic/env.py` already drives batch mode for SQLite. Follow the existing +add-column migration `alembic/versions/20260515_1920_add_route_type_advert_timestamp.py`. + +### 2. Capture the route on message packets (`letsmesh_normalizer.py`) + +In `_build_letsmesh_message_payload` (near line 148 where `path_len` is set), also +extract the ordered hop list and add `normalized_payload["path_hashes"]`. Reuse +the same source as `api/routes/packet_groups.py::_extract_path_hashes` +(`decoded.get("path")`), normalized through the existing `_normalize_hash_list` +helper (line 820). `sender_name` is already populated (line 172) — no change there. + +### 3. Scoring module (new — `collector/spam.py`) + +Pure, unit-testable helpers plus one DB-querying scorer and a config object: + +- `normalize_sender(name: str | None) -> str | None` — lower-case, strip trailing + digits/whitespace (`bob17` → `bob`); None if empty. +- `compute_path_prefix(path_hashes, hops) -> str | None` — join first `hops` hop + hashes; None if no path. +- `score_message(session, *, path_prefix, sender_normalized, path_len, received_at, cfg) -> float` + — two windowed `COUNT(*)` queries over `messages` (using the new indexes). The + cutoff is computed in Python (`cutoff = received_at - timedelta(seconds=cfg.window_seconds)`) + and passed as a bound parameter — `where(Message.received_at >= cutoff)` — so + identical code runs on SQLite and Postgres (no SQL `NOW()`/`INTERVAL`): + - `path_count` = **joint** count: rows with the same `path_prefix` **and** the + same `sender_normalized` where `received_at >= cutoff` (uses the + `(path_prefix, received_at)` index, then filters on `sender_normalized`). Only + runs when both `path_prefix` and `sender_normalized` are present. + - `name_count` = rows with the same `sender_normalized` in the same window + - combine: a weighted average over the signals **actually available**, + normalised by the active weight. With both signals present (weights summing + to 1.0): `min(1.0, w_path*min(path_count/path_thr,1) + w_name*min(name_count/name_thr,1))`. + - counts exclude the current row (scored before insert), so a first-ever message + scores 0. + - **Minimum-path-length gate:** if `path_len < cfg.min_path_hops` (or there is no + path at all — e.g. an observer **zero hops** from the sender), the path signal + is disabled and `path_prefix` is stored null. In that case the **name signal + stands alone at full weight** (`score = min(1.0, name_count/name_thr)`), *not* + capped at `w_name`. Without this, name-only spam (`bob1`/`bob2` rotation) on a + short/zero-hop path could never exceed `w_name` (0.3) and would be permanently + undetectable below a 0.6 threshold. +- `SpamConfig` dataclass, read once from env (mirrors the `envvar=` pattern in + `collector/cli.py`). The operational switch lives on the shared `Settings` + (`common/config.py`) as `spam_detection_enabled` (env `SPAM_DETECTION_ENABLED`) so + the API reads it too; the UI-facing `feature_spam_detection` (`FEATURE_SPAM_DETECTION`) + is bridged to it in Compose (see §7): + + | Env var | Default | Meaning | + |---------|---------|---------| + | `SPAM_DETECTION_ENABLED` | **false** | Operational switch read by collector + API (bridged from `FEATURE_SPAM_DETECTION`) | + | `SPAM_WINDOW_SECONDS` | 300 | Sliding window for counts | + | `SPAM_PATH_HOPS` | 3 | Leading hops that form the prefix | + | `SPAM_MIN_PATH_HOPS` | 5 | Minimum `path_len` before the path signal applies | + | `SPAM_PATH_THRESHOLD` | 5 | path_count that saturates the path signal | + | `SPAM_NAME_THRESHOLD` | 5 | name_count that saturates the name signal | + | `SPAM_WEIGHT_PATH` | 0.7 | Weight of the path signal | + | `SPAM_WEIGHT_NAME` | 0.3 | Weight of the name signal | + | `SPAM_RESCORE_INTERVAL_SECONDS` | 120 | Background re-scoring sweep cadence (`0` disables) | + +### 4. Wire scoring into the handler (`collector/handlers/message.py`) + +**Guarded by the master switch:** when `spam_detection_enabled` is false, skip all +of the below — `Message` is created with `spam_score`/`path_prefix`/`sender_normalized` +left null, exactly as today (one early `if not cfg.enabled: ...` bypass). + +In `_handle_message`, on the **new-insert branch only** (~line 143; not the +duplicate/observer-merge path), before constructing `Message(...)`: + +- `path_prefix = compute_path_prefix(payload.get("path_hashes"), cfg.path_hops)` + (left null when `path_len < cfg.min_path_hops`) +- `sender_normalized = normalize_sender(payload.get("sender_name"))` +- `spam_score = score_message(session, path_prefix=…, sender_normalized=…, path_len=path_len, received_at=now, cfg=cfg)` + +Set all three on the `Message` record. This is the **online** score: it counts +only rows that already exist (priors), so the leading edge of a burst scores low +(see the re-scoring sweep below and the limitation note). + +**Log the score at ingest.** Extend the existing "Stored … message" log lines +(`handlers/message.py:195-204`) to include `spam_score` and the signals that drove +it, so the score is visible in the collector log without querying the DB, e.g.: + +``` +Stored channel 0 message [spam=0.84 path=aaa1,aab2 path_n=6 sender=bob]: test... +``` + +Log at `INFO` when `spam_score >= SPAM_SCORE_THRESHOLD` (so likely-spam stands out) +and keep the normal line otherwise; include the score either way. `score_message` +should return the component counts (path_count/name_count) alongside the score so +they can be logged, and the §4b sweep should log at `DEBUG` when it changes a row's +score (old → new). + +### 4b. Background re-scoring sweep (`collector/spam.py` + `subscriber.py`) + +A periodic task that recomputes `spam_score` for recent rows **with hindsight** — +counting matching peers across the *whole* window (including messages that arrived +*after* a given row), not just priors. This flags the leading edge of bursts that +the online score necessarily misses, and closes the bursty-evasion hole. + +This is a **small** addition because the collector already runs periodic async +tasks in the same pattern: the cleanup scheduler (`subscriber.py:363`) and the +channel-refresh loop (`subscriber.py:483`). Add a third task following that +template: + +- `rescore_recent(db, cfg)` in `collector/spam.py` — select rows where + `received_at >= now - lookback` (a couple of windows), recompute each score via + the same combine logic but with **symmetric** window counts + (`abs(other.received_at - row.received_at) <= window`, both directions), and + `UPDATE` only rows whose score changed. Idempotent. +- Schedule it from `Subscriber` like the existing loops, gated by + `SPAM_RESCORE_INTERVAL_SECONDS` (default e.g. 120; `0` disables) **and** the + master switch — the task is not scheduled at all when `spam_detection_enabled` + is false. + +Config-gated so it can ship disabled and be enabled once tuned. + +### 5. Display layer — API (`api/routes/messages.py`, `schemas/messages.py`) + +`list_messages`: +- Add `include_spam: bool = Query(False)`. +- **Master switch:** when `spam_detection_enabled` is false, apply **no** hide-filter + at all — every message is returned regardless of stored score (so toggling the + feature off instantly un-hides everything, including rows scored while it was on). +- When enabled and `include_spam` is False, filter both the count query and the + results query: + `where(or_(Message.spam_score < threshold, Message.spam_score.is_(None)))`, + `threshold = SPAM_SCORE_THRESHOLD` (env, default 0.6). +- Add `spam_score` to the `msg_dict` built ~line 183. +- Update the `@cached` key builder (`_messages_key_builder`) to include + `include_spam` so cached responses don't leak across toggle states. + +`MessageRead`: add `spam_score: Optional[float] = Field(default=None, …)`. + +### 6. Display layer — frontend (`web/static/js/spa/pages/messages.js`) + +- **Only render the toggle/badge when the feature is enabled.** Add + `feature_spam_detection` (`FEATURE_SPAM_DETECTION`, default `False`) to + `common/config.py` and a `"spam"` entry to the `features` property — exactly like + `feature_packets`/`"packets"`. It flows to the SPA via the existing + `app.state.features` mechanism; gate the toggle on `features.spam`. Because Compose + bridges `SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION}`, the UI flag and the + API's hide behaviour move together, so the toggle is never shown while the API is + returning everything (or vice-versa). +- Add a "Show potential spam" checkbox to the existing filter card + (`renderFilterCard` in `components.js`), bound to URL param `?include_spam=true`, + default off, following the existing select-filter pattern. +- Optionally tag rows with `spam_score >= threshold` with a small badge when shown. +- Add the i18n label(s) to `web/static/locales/en.json`. + +### 7. Config & docs + +Config plumbing (vars are passed explicitly per service, not via `env_file`): + +- `common/config.py` — add `feature_spam_detection` (default `False`) next to the + other `feature_*` fields and map `"spam": self.feature_spam_detection` into the + `features` property; add `spam_detection_enabled` (default `False`) next to the + other `*_enabled` ops flags, plus the scoring fields. +- **`.env.example`** — add a new commented `# Spam Detection` block led by + `FEATURE_SPAM_DETECTION=false` (the one switch operators set), with the bridged + `SPAM_DETECTION_ENABLED` documented as Compose-derived, followed by the `SPAM_*` + scoring vars and `SPAM_SCORE_THRESHOLD`. Mirror the Data-Retention / raw-packet + blocks (which document `FEATURE_PACKETS` driving `RAW_PACKET_CAPTURE_ENABLED`). +- **`docker-compose.yml`** — bridge and thread vars the same way as raw-packets + (`- VAR=${VAR:-default}`): collector service gets + `SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION:-false}` + all scoring/sweep vars; + the **api** service gets the same bridged `SPAM_DETECTION_ENABLED` + + `SPAM_SCORE_THRESHOLD`. Apply to `docker-compose.prod.yml` if it enumerates these. + +Docs: + +- **`docs/configuration.md`** — add a new `## Spam Detection` section (var table in + the same format as `## Data Retention`) listing every `SPAM_*` var, its default, + and which service reads it (collector vs API), and documenting the + `SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION}` bridge; add a `FEATURE_SPAM_DETECTION` + row to the `## Feature Flags` table (noting, like `FEATURE_PACKETS`, that it also + drives the backend switch in Compose). +- Note in user-facing docs that the feature is **off by default** and describe the + scoring behaviour + "show potential spam" toggle. +- Update `SCHEMAS.md` with the new `messages` columns. + +### 8. Tests — run against **both** SQLite and Postgres + +The scoring queries (windowed `COUNT(*)`, the joint `path_prefix`+`sender_normalized` +filter, the `or_(... is_(None))` display filter) can pass on SQLite and break on +Postgres (or vice-versa), so the DB-touching tests must exercise **both** backends. + +The project **already** has a backend switch — no new mechanism is needed, just +correct fixture reuse. The whole suite runs against Postgres with: + +``` +TEST_DATABASE_BACKEND=postgres \ +TEST_POSTGRES_URL="postgresql+psycopg2://postgres:postgres@localhost:55432/test" \ +make test +``` + +Default (no env) runs SQLite. `TEST_DATABASE_BACKEND` defaults to `sqlite`; setting +`postgres` requires `TEST_POSTGRES_URL` (else the fixture **skips**). The +`tests/test_api/conftest.py` fixtures (`db_backend`/`db_url`/`api_db_engine`, +schema via `create_all`, per-xdist-worker Postgres DBs) already honor this. + +Implications for the new tests: + +- **API spam tests** go in `tests/test_api/test_messages.py` using the existing + `api_db_session`/test-client fixtures → they run on **both** backends for free + under the command above. No fixture work. +- **Collector spam tests** are the gap: `tests/test_collector/conftest.py` hardcodes + `sqlite:///:memory:` (`db_manager`, `async_db_session`). Make these backend-aware + so `score_message`/`rescore_recent`/handler tests also run on Postgres — extract + the `db_backend`/`db_url` helper from `tests/test_api/conftest.py` into the shared + `tests/conftest.py` and have the collector fixtures consume it (falling back to + in-memory SQLite when `TEST_DATABASE_BACKEND` is unset). +- **Pure-function tests** (`normalize_sender`, `compute_path_prefix`) need no DB. +- **Migration on Postgres:** the suite builds schema via `create_all`, so it does + **not** exercise the Alembic batch migration. Verifying the migration on Postgres + stays a manual step (see Verification) unless we add a dedicated migration test. + +Coverage: +- Seed N identical-path/identical-sender rows → assert online `score_message` rises + past threshold; assert a diverse busy path (many senders) does **not**. +- Handler test: feed duplicate messages → assert `spam_score`, `path_prefix`, + `sender_normalized` populated; assert short-path (`< SPAM_MIN_PATH_HOPS`) rows get + null `path_prefix` and rely on the name signal. +- `rescore_recent`: insert a burst out of order → assert the leading-edge rows get + re-scored above threshold with hindsight. +- API (`tests/test_api/test_messages.py`): spammy rows hidden by default, visible + with `include_spam=true`; cache key differs per toggle state. +- **Master switch (both collector and API):** with `spam_detection_enabled=false` + (the backend flag the services read), the handler stores null `spam_score` (no + scoring) and the API returns **all** rows including ones with a high stored score; + with it true, scoring + hiding apply. (Tests set the backend flag directly; the + `FEATURE_SPAM_DETECTION` bridge is a Compose concern, verified in step 1.) + +## Affected Files + +| File | Change | +|------|--------| +| `src/meshcore_hub/common/config.py` | Add `feature_spam_detection` (+ `"spam"` in `features`) and `spam_detection_enabled` ops switch (both default `False`) + scoring fields | +| `src/meshcore_hub/common/models/message.py` | Add `path_prefix`, `sender_normalized`, `spam_score` columns + 2 composite indexes | +| `src/meshcore_hub/collector/letsmesh_normalizer.py` | Add `path_hashes` to the normalized message payload | +| `src/meshcore_hub/collector/spam.py` | **New** — `normalize_sender`, `compute_path_prefix`, `score_message`, `rescore_recent`, `SpamConfig` | +| `src/meshcore_hub/collector/handlers/message.py` | Compute + store `path_prefix`/`sender_normalized`/`spam_score` on insert | +| `src/meshcore_hub/collector/subscriber.py` | Add periodic re-scoring task alongside the existing cleanup/channel-refresh loops | +| `src/meshcore_hub/collector/cli.py` | Wire `SPAM_*` env/config options (incl. `SPAM_RESCORE_INTERVAL_SECONDS`) | +| `src/meshcore_hub/api/routes/messages.py` | `include_spam` param, default-hide filter, `spam_score` in response, cache key | +| `src/meshcore_hub/common/schemas/messages.py` | Add `spam_score` to `MessageRead` | +| `src/meshcore_hub/web/static/js/spa/pages/messages.js` | "Show potential spam" toggle + optional badge | +| `src/meshcore_hub/web/static/locales/en.json` | Toggle/badge labels | +| `alembic/versions/` | Migration for the 3 new columns + 2 indexes | +| `.env.example` | New `# Spam Detection` block led by `FEATURE_SPAM_DETECTION` | +| `docker-compose.yml` (+ `docker-compose.prod.yml` if it lists these) | Bridge `SPAM_DETECTION_ENABLED=${FEATURE_SPAM_DETECTION:-false}` into collector (+ scoring vars) and api (+ `SPAM_SCORE_THRESHOLD`) env blocks | +| `docs/configuration.md` | New `## Spam Detection` var table + `FEATURE_SPAM_DETECTION` row in `## Feature Flags` | +| `docs/` (user-facing) | Document scoring, toggle, off-by-default behaviour | +| `SCHEMAS.md` | Update `messages` schema | +| `tests/conftest.py` + `tests/test_collector/conftest.py` | Make collector DB fixtures backend-aware: lift the `db_backend`/`db_url` helper out of `tests/test_api/conftest.py` into the shared conftest so `db_manager`/`async_db_session` honor `TEST_DATABASE_BACKEND`/`TEST_POSTGRES_URL` (default SQLite) | +| `tests/test_collector/` | Unit tests for `normalize_sender`, `compute_path_prefix`; DB tests for `score_message`, `rescore_recent`, and the handler (run on both backends via the env switch) | +| `tests/test_api/test_messages.py` | Default-hide vs `include_spam=true` behavior (uses existing `api_db_*` fixtures → both backends) | + +## Verification + +1. `meshcore-hub db upgrade` succeeds; `messages` shows the 3 columns + 2 indexes. + With `FEATURE_SPAM_DETECTION` unset (default → bridged `SPAM_DETECTION_ENABLED=false`), + confirm ingest writes null `spam_score`, the API hides nothing, and the UI shows + no spam toggle — i.e. the feature is fully dark out of the box. Then set + `FEATURE_SPAM_DETECTION=true` (or `SPAM_DETECTION_ENABLED=true` when running + services directly without Compose) for the remaining steps. +2. Run the collector against seed/sample MQTT data (`seed/`, + `docker-compose.dev.yml`); confirm new rows get `path_prefix`, + `sender_normalized`, and a `spam_score`. +3. Replay a burst of near-identical messages sharing a path prefix with + `path_len >= SPAM_MIN_PATH_HOPS`; later rows score above + `SPAM_SCORE_THRESHOLD`. Confirm a short-path burst (`path_len <` gate) is **not** + flagged on the path signal alone. +4. `GET /api/v1/messages` hides flagged rows by default; `?include_spam=true` + returns them with `spam_score` populated; the UI toggle reflects this. +5. `pytest` (unit + handler + API) green; `ruff`/`mypy` clean per + `.pre-commit-config.yaml`. +6. Run the full suite on **both** backends and confirm green: + - SQLite (default): `make test` + - Postgres: `TEST_DATABASE_BACKEND=postgres TEST_POSTGRES_URL="postgresql+psycopg2://postgres:postgres@localhost:55432/test" make test` + This confirms the windowed/joint counts and the display filter behave identically + on both. Separately, apply the **Alembic migration** to a Postgres DB + (`meshcore-hub db upgrade` against the test cluster) to prove the batch + `add_column` + composite indexes apply there — the suite builds schema via + `create_all`, so the migration itself is not covered by tests. +7. **Cross-observer stability:** confirm the same logical message seen by multiple + observers stores the same origin-side `path_prefix` (`decoded.path` is + origin-first — already relied on by the Packets page, so `path_hashes[:N]` is + correct). +8. **Channel sender check:** confirm `sender_name` is populated from the + `": "` body prefix for public-channel messages, and that + `normalize_sender` strips the rotating suffix (`bob1`/`bob2` → `bob`). + +## Notes / Non-goals + +- **Backfill:** existing rows have null `path_prefix`/`sender_normalized`, so they + don't contribute to counts until re-seen. Because the window is minutes, counts + self-heal; no historical backfill (which would require re-decoding `raw_packets`) + in v1. +- **Content fingerprinting** (text shingling, after stripping the `": "` + prefix) is a future tertiary signal, not in scope now. +- **Frozen score / leading edge of bursts.** The *online* score (set at insert) + counts only priors, so the first `~path_thr` messages of a burst score low and, + without re-scoring, stay shown forever — and **bursty** spam (a few messages, a + pause longer than the window, repeat) could evade entirely because the online + count restarts at zero each burst. This is exactly what the §4b background + re-scoring sweep addresses by counting peers symmetrically (with hindsight). + `SPAM_PATH_THRESHOLD` is the precision/recall knob: higher = fewer false + positives but more of each burst's head leaks until the sweep catches it. diff --git a/src/meshcore_hub/api/app.py b/src/meshcore_hub/api/app.py index fb203e0..c9f3eb5 100644 --- a/src/meshcore_hub/api/app.py +++ b/src/meshcore_hub/api/app.py @@ -91,6 +91,8 @@ def create_app( redis_key_prefix: str = "hub", redis_cache_ttl: int = 30, redis_cache_ttl_dashboard: int = 30, + spam_detection_enabled: bool = False, + spam_score_threshold: float = 0.6, ) -> FastAPI: """Create and configure the FastAPI application. @@ -152,6 +154,8 @@ def create_app( app.state.redis_key_prefix = redis_key_prefix app.state.redis_cache_ttl = redis_cache_ttl app.state.redis_cache_ttl_dashboard = redis_cache_ttl_dashboard + app.state.spam_detection_enabled = spam_detection_enabled + app.state.spam_score_threshold = spam_score_threshold # Configure CORS if cors_origins is None: @@ -268,4 +272,6 @@ def create_app_from_env() -> FastAPI: redis_key_prefix=settings.redis_key_prefix, redis_cache_ttl=settings.redis_cache_ttl, redis_cache_ttl_dashboard=settings.redis_cache_ttl_dashboard, + spam_detection_enabled=settings.spam_detection_enabled, + spam_score_threshold=settings.spam_score_threshold, ) diff --git a/src/meshcore_hub/api/routes/messages.py b/src/meshcore_hub/api/routes/messages.py index 67b6de0..f3cc090 100644 --- a/src/meshcore_hub/api/routes/messages.py +++ b/src/meshcore_hub/api/routes/messages.py @@ -58,6 +58,9 @@ def list_messages( since: Optional[datetime] = Query(None, description="Start timestamp"), until: Optional[datetime] = Query(None, description="End timestamp"), search: Optional[str] = Query(None, description="Search in message text"), + include_spam: bool = Query( + False, description="Include messages scored as likely spam" + ), sort: Optional[str] = Query(None, description="Sort column"), order: Optional[str] = Query(None, description="Sort direction: asc or desc"), limit: int = Query(50, ge=1, le=100, description="Page size"), @@ -98,6 +101,19 @@ def list_messages( if search: query = query.where(Message.text.ilike(f"%{search}%")) + # Spam hide-filter. Master switch: when spam detection is disabled, never + # filter — every message is returned regardless of any stored score, so + # toggling the feature off instantly un-hides rows scored while it was on. + spam_enabled = getattr(request.app.state, "spam_detection_enabled", False) + spam_threshold = getattr(request.app.state, "spam_score_threshold", 0.6) + if spam_enabled and not include_spam: + query = query.where( + or_( + Message.spam_score < spam_threshold, + Message.spam_score.is_(None), + ) + ) + # Apply channel visibility filtering role = resolve_user_role(request) max_level = get_max_visibility_level(role) @@ -204,6 +220,7 @@ def list_messages( "received_at": m.received_at, "created_at": m.created_at, "packet_hash": m.packet_hash, + "spam_score": m.spam_score, "observers": ( observers_by_hash.get(m.event_hash, []) if m.event_hash else [] ), @@ -271,6 +288,7 @@ def get_message( "received_at": message.received_at, "created_at": message.created_at, "packet_hash": message.packet_hash, + "spam_score": message.spam_score, "observers": observers, } return MessageRead(**data) diff --git a/src/meshcore_hub/collector/handlers/message.py b/src/meshcore_hub/collector/handlers/message.py index 20b25c0..9c027b4 100644 --- a/src/meshcore_hub/collector/handlers/message.py +++ b/src/meshcore_hub/collector/handlers/message.py @@ -10,6 +10,12 @@ from sqlalchemy.exc import IntegrityError from meshcore_hub.common.database import DatabaseManager from meshcore_hub.common.hash_utils import compute_message_hash from meshcore_hub.common.models import Message, Node, add_event_observer +from meshcore_hub.collector.spam import ( + compute_path_prefix, + get_spam_config, + normalize_sender, + score_message, +) logger = logging.getLogger(__name__) @@ -140,6 +146,34 @@ def _handle_message( ) return + # Spam scoring (online): only on the first insert of an event_hash, and + # only when the feature is switched on. When off, the columns stay null + # exactly as before. Counts existing rows (priors), so the leading edge + # of a burst scores low until the background re-scoring sweep catches it. + cfg = get_spam_config() + spam_score: float | None = None + path_prefix: str | None = None + sender_normalized: str | None = None + spam_path_count = 0 + spam_name_count = 0 + if cfg.enabled: + sender_normalized = normalize_sender(payload.get("sender_name")) + if path_len is not None and path_len >= cfg.min_path_hops: + path_prefix = compute_path_prefix( + payload.get("path_hashes"), cfg.path_hops + ) + result = score_message( + session, + path_prefix=path_prefix, + sender_normalized=sender_normalized, + path_len=path_len, + received_at=now, + cfg=cfg, + ) + spam_score = result.score + spam_path_count = result.path_count + spam_name_count = result.name_count + # Create message record message = Message( observer_node_id=receiver_node.id if receiver_node else None, @@ -155,6 +189,9 @@ def _handle_message( received_at=now, event_hash=event_hash, packet_hash=packet_hash, + path_prefix=path_prefix, + sender_normalized=sender_normalized, + spam_score=spam_score, ) session.add(message) @@ -192,13 +229,26 @@ def _handle_message( ) return + # Surface the spam score (and the signals that drove it) in the log so it is + # observable without querying the DB. Likely-spam (>= threshold) is logged at + # WARNING so it stands out; everything else stays at INFO. + spam_suffix = "" + is_spam = False + if cfg.enabled and spam_score is not None: + spam_suffix = ( + f" [spam={spam_score:.2f} path={path_prefix or '-'} " + f"path_n={spam_path_count} sender={sender_normalized or '-'} " + f"name_n={spam_name_count}]" + ) + is_spam = spam_score >= cfg.score_threshold + + text_preview = f"{text[:30]}{'...' if len(text) > 30 else ''}" if message_type == "contact": - logger.info( - f"Stored contact message from {pubkey_prefix!r}: " - f"{text[:30]}{'...' if len(text) > 30 else ''}" - ) + line = f"Stored contact message from {pubkey_prefix!r}{spam_suffix}: {text_preview}" else: - logger.info( - f"Stored channel {channel_idx} message: " - f"{text[:30]}{'...' if len(text) > 30 else ''}" - ) + line = f"Stored channel {channel_idx} message{spam_suffix}: {text_preview}" + + if is_spam: + logger.warning(line) + else: + logger.info(line) diff --git a/src/meshcore_hub/collector/letsmesh_normalizer.py b/src/meshcore_hub/collector/letsmesh_normalizer.py index f91e909..2464ddc 100644 --- a/src/meshcore_hub/collector/letsmesh_normalizer.py +++ b/src/meshcore_hub/collector/letsmesh_normalizer.py @@ -150,6 +150,13 @@ class LetsMeshNormalizer: path_len = self._extract_letsmesh_decoder_path_length(decoded_packet) normalized_payload["path_len"] = path_len + # Ordered origin-side hop hashes, used as a spam-scoring signal. The path + # is discarded for messages today; the trace handler already keeps it + # (see _build_letsmesh_trace_payload) using the same decoded.path source. + path_hashes = self._extract_message_path_hashes(decoded_packet) + if path_hashes: + normalized_payload["path_hashes"] = path_hashes + sender_timestamp = self._parse_sender_timestamp(payload) if sender_timestamp is None: sender_timestamp = self._extract_letsmesh_decoder_sender_timestamp( @@ -816,6 +823,26 @@ class LetsMeshNormalizer: return None + def _extract_message_path_hashes( + self, decoded_packet: dict[str, Any] | None + ) -> list[str] | None: + """Extract the ordered hop hashes from a decoded message packet. + + Mirrors ``api/routes/packet_groups.py::_extract_path_hashes``: the path + lives at the top level as ``decoded.path`` for normal packets, with + ``decoded.payload.decoded.pathHashes`` as the trace-style fallback. The + hashes are normalised (upper-cased, validated hex) so the stored + ``path_prefix`` is stable across observers. + """ + if not decoded_packet: + return None + hashes = self._normalize_hash_list(decoded_packet.get("path")) + if hashes: + return hashes + payload = decoded_packet.get("payload") or {} + inner = payload.get("decoded") or {} + return self._normalize_hash_list(inner.get("pathHashes")) + @staticmethod def _normalize_hash_list(value: Any) -> list[str] | None: """Normalize a list of variable-length hex hash strings. diff --git a/src/meshcore_hub/collector/spam.py b/src/meshcore_hub/collector/spam.py new file mode 100644 index 0000000..bb4fb5c --- /dev/null +++ b/src/meshcore_hub/collector/spam.py @@ -0,0 +1,315 @@ +"""Spam scoring for ingested messages. + +Pure helpers (`normalize_sender`, `compute_path_prefix`) plus two DB-querying +scorers (`score_message`, `rescore_recent`) and a config object (`SpamConfig`). + +The design is described in ``docs/plans/20260622-2243-spam-detection/plan.md``. +Counts are computed directly from Postgres/SQLite over the +``(path_prefix, received_at)`` and ``(sender_normalized, received_at)`` indexes; +the window cutoff is always a Python-computed datetime passed as a bound +parameter, so identical code runs on both backends (no SQL ``NOW()``/``INTERVAL``). +""" + +from __future__ import annotations + +import logging +import re +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import TYPE_CHECKING, Optional + +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from meshcore_hub.common.models import Message + +if TYPE_CHECKING: + from meshcore_hub.common.config import CollectorSettings + +logger = logging.getLogger(__name__) + +# Trailing run of digits and/or whitespace, e.g. "bob 17" / "bob17" -> "bob". +_TRAILING_DIGITS = re.compile(r"[\s\d]+$") + + +@dataclass(frozen=True) +class SpamConfig: + """Resolved spam-scoring configuration. + + Built from ``CollectorSettings`` (which inherits the shared + ``spam_detection_enabled`` / ``spam_score_threshold`` from CommonSettings). + """ + + enabled: bool = False + window_seconds: int = 300 + path_hops: int = 3 + min_path_hops: int = 5 + path_threshold: int = 5 + name_threshold: int = 5 + weight_path: float = 0.7 + weight_name: float = 0.3 + score_threshold: float = 0.6 + rescore_interval_seconds: int = 120 + + @classmethod + def from_settings(cls, settings: "CollectorSettings") -> "SpamConfig": + """Build a SpamConfig from a CollectorSettings instance.""" + return cls( + enabled=settings.spam_detection_enabled, + window_seconds=settings.spam_window_seconds, + path_hops=settings.spam_path_hops, + min_path_hops=settings.spam_min_path_hops, + path_threshold=settings.spam_path_threshold, + name_threshold=settings.spam_name_threshold, + weight_path=settings.spam_weight_path, + weight_name=settings.spam_weight_name, + score_threshold=settings.spam_score_threshold, + rescore_interval_seconds=settings.spam_rescore_interval_seconds, + ) + + +@dataclass(frozen=True) +class SpamScore: + """Result of scoring a single message, with the driving component counts.""" + + score: float + path_count: int + name_count: int + + +_cached_config: Optional[SpamConfig] = None + + +def get_spam_config() -> SpamConfig: + """Return the process-wide SpamConfig, built once from collector settings.""" + global _cached_config + if _cached_config is None: + from meshcore_hub.common.config import get_collector_settings + + _cached_config = SpamConfig.from_settings(get_collector_settings()) + return _cached_config + + +def reset_spam_config() -> None: + """Clear the cached SpamConfig (used by tests after changing env).""" + global _cached_config + _cached_config = None + + +def normalize_sender(name: Optional[str]) -> Optional[str]: + """Lower-case a sender name and strip the trailing digit/space suffix. + + ``bob17`` -> ``bob``; ``Bob 2`` -> ``bob``. Returns None for empty input or + a name that is entirely digits/whitespace. + """ + if not name: + return None + stripped = _TRAILING_DIGITS.sub("", name.strip()) + stripped = stripped.strip().lower() + return stripped or None + + +def compute_path_prefix(path_hashes: Optional[list[str]], hops: int) -> Optional[str]: + """Join the first ``hops`` origin-side hop hashes into a stable prefix key. + + Returns None when there is no path or ``hops`` is non-positive. + """ + if not path_hashes or hops <= 0: + return None + prefix = ",".join(path_hashes[:hops]) + return prefix or None + + +def _count( + session: Session, + *, + sender_normalized: str, + path_prefix: Optional[str], + lo: datetime, + hi: Optional[datetime], + exclude_id: Optional[str], +) -> int: + """Count messages in ``[lo, hi]`` matching sender (and path when given).""" + stmt = ( + select(func.count()) + .select_from(Message) + .where(Message.sender_normalized == sender_normalized) + .where(Message.received_at >= lo) + ) + if path_prefix is not None: + stmt = stmt.where(Message.path_prefix == path_prefix) + if hi is not None: + stmt = stmt.where(Message.received_at <= hi) + if exclude_id is not None: + stmt = stmt.where(Message.id != exclude_id) + return int(session.execute(stmt).scalar() or 0) + + +def _combine( + *, path_count: int, name_count: int, path_eligible: bool, cfg: SpamConfig +) -> float: + """Combine the windowed counts into a 0.0-1.0 score. + + The score is a weighted average over the signals that are actually + *available*, normalised by the active weight. When both signals are present + and the weights sum to 1.0 (the default), this is the plain + ``w_path*path + w_name*name`` blend. + + The key case: when the path signal is gated off — a short/zero-hop path or no + path data at all, e.g. an observer sitting right next to the sender — the name + signal stands on its own at *full* weight. Otherwise name-only spam (rotating + ``bob1``/``bob2``) would be capped at ``weight_name`` (0.3 by default) and + could never cross the threshold, leaving local/zero-hop spam undetectable. + """ + name_thr = max(cfg.name_threshold, 1) + name_term = min(name_count / name_thr, 1.0) + + if not path_eligible: + # Name signal is the only evidence available; trust it fully. + return min(1.0, name_term) + + path_thr = max(cfg.path_threshold, 1) + path_term = min(path_count / path_thr, 1.0) + total_weight = cfg.weight_path + cfg.weight_name + if total_weight <= 0: + return 0.0 + score = (cfg.weight_path * path_term + cfg.weight_name * name_term) / total_weight + return min(1.0, score) + + +def score_message( + session: Session, + *, + path_prefix: Optional[str], + sender_normalized: Optional[str], + path_len: Optional[int], + received_at: datetime, + cfg: SpamConfig, +) -> SpamScore: + """Online spam score for a message about to be inserted. + + Counts only prior rows (those already in the window before ``received_at``), + so a first-ever message scores 0. The path signal is the joint + ``(path_prefix, sender_normalized)`` count and is only applied when the path + is at/above the ``min_path_hops`` gate; otherwise scoring relies on the name + signal alone. + """ + if sender_normalized is None: + return SpamScore(score=0.0, path_count=0, name_count=0) + + cutoff = received_at - timedelta(seconds=cfg.window_seconds) + path_eligible = ( + path_prefix is not None + and path_len is not None + and path_len >= cfg.min_path_hops + ) + + path_count = 0 + if path_eligible: + path_count = _count( + session, + sender_normalized=sender_normalized, + path_prefix=path_prefix, + lo=cutoff, + hi=None, + exclude_id=None, + ) + name_count = _count( + session, + sender_normalized=sender_normalized, + path_prefix=None, + lo=cutoff, + hi=None, + exclude_id=None, + ) + + score = _combine( + path_count=path_count, + name_count=name_count, + path_eligible=path_eligible, + cfg=cfg, + ) + return SpamScore(score=score, path_count=path_count, name_count=name_count) + + +def rescore_recent( + session: Session, + cfg: SpamConfig, + now: Optional[datetime] = None, +) -> int: + """Recompute scores for recent rows with hindsight (symmetric window). + + Unlike the online score, this counts peers on *both* sides of each row + (``|other.received_at - row.received_at| <= window``), so it catches the + leading edge of a burst that the online score necessarily missed. Only rows + whose score actually changes are written. Returns the number of rows updated. + Idempotent. + """ + if now is None: + now = datetime.now(timezone.utc) + + window = timedelta(seconds=cfg.window_seconds) + # Look back a couple of windows so we re-touch rows whose hindsight peers + # have since arrived, without rescanning the whole table. + lookback = now - timedelta(seconds=cfg.window_seconds * 2) + + rows = ( + session.execute(select(Message).where(Message.received_at >= lookback)) + .scalars() + .all() + ) + + updated = 0 + for row in rows: + if row.sender_normalized is None: + new_score = 0.0 + path_count = 0 + name_count = 0 + else: + center = row.received_at + lo = center - window + hi = center + window + path_eligible = ( + row.path_prefix is not None + and row.path_len is not None + and row.path_len >= cfg.min_path_hops + ) + path_count = 0 + if path_eligible: + path_count = _count( + session, + sender_normalized=row.sender_normalized, + path_prefix=row.path_prefix, + lo=lo, + hi=hi, + exclude_id=row.id, + ) + name_count = _count( + session, + sender_normalized=row.sender_normalized, + path_prefix=None, + lo=lo, + hi=hi, + exclude_id=row.id, + ) + new_score = _combine( + path_count=path_count, + name_count=name_count, + path_eligible=path_eligible, + cfg=cfg, + ) + + old_score = row.spam_score + if old_score is None or abs(old_score - new_score) > 1e-9: + logger.debug( + "Re-scored message %s: %s -> %.2f (path_n=%d name_n=%d)", + row.id, + "None" if old_score is None else f"{old_score:.2f}", + new_score, + path_count, + name_count, + ) + row.spam_score = new_score + updated += 1 + + return updated diff --git a/src/meshcore_hub/collector/subscriber.py b/src/meshcore_hub/collector/subscriber.py index 67d4373..ecb6dd7 100644 --- a/src/meshcore_hub/collector/subscriber.py +++ b/src/meshcore_hub/collector/subscriber.py @@ -98,6 +98,8 @@ class Subscriber(LetsMeshNormalizer): # Channel key refresh self._channel_refresh_interval_seconds = channel_refresh_interval_seconds self._channel_refresh_thread: Optional[threading.Thread] = None + # Background spam re-scoring sweep + self._spam_rescore_thread: Optional[threading.Thread] = None # Load initial channel keys from database self._include_test_channel = self._load_channel_keys_from_db() self._letsmesh_decoder = LetsMeshPacketDecoder( @@ -512,6 +514,60 @@ class Subscriber(LetsMeshNormalizer): if self._channel_refresh_thread.is_alive(): logger.warning("Channel refresh thread did not stop cleanly") + def _start_spam_rescore_scheduler(self) -> None: + """Start background thread that re-scores recent messages with hindsight. + + Disabled (not scheduled at all) when spam detection is off or the + interval is 0. Follows the same loop template as the channel-refresh + scheduler and uses synchronous sessions. + """ + from meshcore_hub.collector.spam import get_spam_config + + cfg = get_spam_config() + if not cfg.enabled or cfg.rescore_interval_seconds <= 0: + logger.info( + "Spam re-scoring sweep disabled (enabled=%s, interval=%ds)", + cfg.enabled, + cfg.rescore_interval_seconds, + ) + return + + interval = cfg.rescore_interval_seconds + logger.info("Starting spam re-scoring sweep (interval=%ds)", interval) + + def run_rescore_loop() -> None: + """Periodically re-score recent messages with symmetric-window counts.""" + from meshcore_hub.collector.spam import get_spam_config, rescore_recent + + while self._running: + for _ in range(interval): + if not self._running: + break + time.sleep(1) + if self._running: + try: + sweep_cfg = get_spam_config() + with self.db.session_scope() as session: + updated = rescore_recent(session, sweep_cfg) + if updated: + logger.info( + "Spam re-scoring sweep updated %d rows", updated + ) + except Exception as e: + logger.error("Spam re-scoring error: %s", e, exc_info=True) + + self._spam_rescore_thread = threading.Thread( + target=run_rescore_loop, daemon=True, name="spam-rescore" + ) + self._spam_rescore_thread.start() + + def _stop_spam_rescore_scheduler(self) -> None: + """Stop the spam re-scoring sweep thread.""" + if self._spam_rescore_thread and self._spam_rescore_thread.is_alive(): + self._spam_rescore_thread.join(timeout=5.0) + if self._spam_rescore_thread.is_alive(): + logger.warning("Spam re-scoring thread did not stop cleanly") + def start(self) -> None: """Start the subscriber.""" logger.info("Starting collector subscriber") @@ -579,6 +635,9 @@ class Subscriber(LetsMeshNormalizer): # Start channel key refresh scheduler self._start_channel_refresh_scheduler() + # Start background spam re-scoring sweep (no-op when disabled) + self._start_spam_rescore_scheduler() + # Start health reporter for Docker health checks self._health_reporter = HealthReporter( component="collector", @@ -617,6 +676,9 @@ class Subscriber(LetsMeshNormalizer): # Stop channel refresh scheduler self._stop_channel_refresh_scheduler() + # Stop spam re-scoring sweep + self._stop_spam_rescore_scheduler() + # Stop webhook processor self._stop_webhook_processor() diff --git a/src/meshcore_hub/common/config.py b/src/meshcore_hub/common/config.py index 3536f35..a5d9544 100644 --- a/src/meshcore_hub/common/config.py +++ b/src/meshcore_hub/common/config.py @@ -153,6 +153,25 @@ class CommonSettings(BaseSettings): description="WebSocket path for MQTT transport (used when MQTT_TRANSPORT=websockets)", ) + # Spam detection (shared operational switch + display threshold). + # The collector reads it to decide whether to score; the API reads it to + # decide whether to hide flagged messages. Bridged from FEATURE_SPAM_DETECTION + # in Compose so operators set a single var. Both default off (feature dark). + spam_detection_enabled: bool = Field( + default=False, + description=( + "Operational switch for spam scoring/hiding, read by collector + API " + "(bridged from FEATURE_SPAM_DETECTION in Compose)" + ), + ) + spam_score_threshold: float = Field( + default=0.6, + description=( + "Score at/above which a message is treated as likely spam (hidden by " + "default in the API; logged at the collector)" + ), + ) + class CollectorSettings(CommonSettings): """Settings for the Collector component.""" @@ -242,6 +261,51 @@ class CollectorSettings(CommonSettings): ge=1, ) + # Spam scoring tuning (only consulted when SPAM_DETECTION_ENABLED is true). + # The shared SPAM_DETECTION_ENABLED / SPAM_SCORE_THRESHOLD live on + # CommonSettings; these collector-only knobs tune the scorer + sweep. + spam_window_seconds: int = Field( + default=300, + description="Sliding window (seconds) for spam frequency counts", + ge=1, + ) + spam_path_hops: int = Field( + default=3, + description="Number of leading origin-side hops that form the path prefix", + ge=1, + ) + spam_min_path_hops: int = Field( + default=5, + description=( + "Minimum path_len before the path signal is applied; below this the " + "path_prefix is stored null (short local-mesh paths share prefixes)" + ), + ge=0, + ) + spam_path_threshold: int = Field( + default=5, + description="Joint path+sender count that saturates the path signal", + ge=1, + ) + spam_name_threshold: int = Field( + default=5, + description="Sender count that saturates the name signal", + ge=1, + ) + spam_weight_path: float = Field( + default=0.7, description="Weight of the path signal in the combined score" + ) + spam_weight_name: float = Field( + default=0.3, description="Weight of the name signal in the combined score" + ) + spam_rescore_interval_seconds: int = Field( + default=120, + description=( + "Background re-scoring sweep cadence in seconds (0 disables the sweep)" + ), + ge=0, + ) + @property def effective_raw_packet_retention_days(self) -> int: """Resolve raw-packet retention, falling back to the global default.""" @@ -498,6 +562,13 @@ class WebSettings(CommonSettings): feature_radio_config: bool = Field( default=True, description="Enable radio config panel on home page" ) + feature_spam_detection: bool = Field( + default=False, + description=( + "Expose the 'show potential spam' toggle on the messages page; in " + "Compose this also drives the backend SPAM_DETECTION_ENABLED switch" + ), + ) # Content directory (contains pages/ and media/ subdirectories) content_home: Optional[str] = Field( @@ -528,6 +599,7 @@ class WebSettings(CommonSettings): "packets": self.feature_packets, "pages": self.feature_pages, "radio_config": self.feature_radio_config, + "spam": self.feature_spam_detection, } @property diff --git a/src/meshcore_hub/common/models/message.py b/src/meshcore_hub/common/models/message.py index 938bee6..7dc965e 100644 --- a/src/meshcore_hub/common/models/message.py +++ b/src/meshcore_hub/common/models/message.py @@ -26,6 +26,9 @@ class Message(Base, UUIDMixin, TimestampMixin): sender_timestamp: Sender's timestamp received_at: When received by interface created_at: Record creation timestamp + path_prefix: First N origin-side hop hashes joined (spam scoring signal) + sender_normalized: Lower-cased sender name with trailing digits stripped + spam_score: Likely-spam score 0.0-1.0 (null when scoring disabled) """ __tablename__ = "messages" @@ -87,12 +90,35 @@ class Message(Base, UUIDMixin, TimestampMixin): nullable=True, index=True, ) + # Spam scoring: first N origin-side hop hashes joined (e.g. "16,69,23"); + # null below the path-length gate so short paths don't pollute path counts. + path_prefix: Mapped[Optional[str]] = mapped_column( + String(48), + nullable=True, + ) + # Spam scoring: lower-cased sender name with trailing digits stripped + # (e.g. "bob17" -> "bob"), so suffix rotation collapses to one identity. + sender_normalized: Mapped[Optional[str]] = mapped_column( + String(255), + nullable=True, + ) + # Spam scoring: likely-spam score 0.0-1.0; null when scoring is disabled. + spam_score: Mapped[Optional[float]] = mapped_column( + Float, + nullable=True, + ) __table_args__ = ( Index("ix_messages_message_type", "message_type"), Index("ix_messages_pubkey_prefix", "pubkey_prefix"), Index("ix_messages_channel_idx", "channel_idx"), Index("ix_messages_received_at", "received_at"), + Index("ix_messages_path_prefix_received_at", "path_prefix", "received_at"), + Index( + "ix_messages_sender_normalized_received_at", + "sender_normalized", + "received_at", + ), ) def __repr__(self) -> str: diff --git a/src/meshcore_hub/common/schemas/messages.py b/src/meshcore_hub/common/schemas/messages.py index 13f57ce..fc11095 100644 --- a/src/meshcore_hub/common/schemas/messages.py +++ b/src/meshcore_hub/common/schemas/messages.py @@ -61,6 +61,10 @@ class MessageRead(BaseModel): packet_hash: Optional[str] = Field( default=None, description="LetsMesh wire packet hash, links to raw packets" ) + spam_score: Optional[float] = Field( + default=None, + description="Likely-spam score 0.0-1.0 (null when scoring disabled)", + ) observers: list[ObserverInfo] = Field( default_factory=list, description="All observers that captured this message" ) diff --git a/src/meshcore_hub/web/static/js/spa/pages/messages.js b/src/meshcore_hub/web/static/js/spa/pages/messages.js index e7c7ca9..40aa8d0 100644 --- a/src/meshcore_hub/web/static/js/spa/pages/messages.js +++ b/src/meshcore_hub/web/static/js/spa/pages/messages.js @@ -15,6 +15,7 @@ export async function render(container, params, router) { const query = params.query || {}; const message_type = query.message_type || ''; const channel_idx = query.channel_idx || ''; + const includeSpamParam = query.include_spam === 'true' || query.include_spam === true; const page = parseInt(query.page, 10) || 1; const limit = parseInt(query.limit, 10) || 50; const offset = (page - 1) * limit; @@ -27,6 +28,10 @@ export async function render(container, params, router) { const config = getConfig(); const features = config.features || {}; const packetsEnabled = features.packets !== false; + // Spam toggle is only shown when the feature is enabled; when off the API + // returns everything anyway, so the include_spam param is a no-op. + const spamEnabled = features.spam === true; + const includeSpam = spamEnabled && includeSpamParam; let channelLabels = new Map(); const tz = config.timezone || ''; const tzBadge = tz && tz !== 'UTC' ? html`${tz}` : nothing; @@ -124,6 +129,16 @@ export async function render(container, params, router) { return `${sender}: ${body}`; } + // Small badge for rows the scorer flagged as likely spam. Only meaningful + // when the spam feature is on (otherwise spam_score is null on every row). + function spamBadge(msg) { + if (!spamEnabled || msg.spam_score == null || msg.spam_score < 0.6) { + return nothing; + } + return html`${t('messages.spam.badge')}`; + } + function dedupeBySignature(items) { const deduped = []; const bySignature = new Map(); @@ -261,6 +276,7 @@ ${displayContent}`, container); // Phase 2: fetch the messages with the resolved observer filter. const apiParams = { limit, offset, message_type, channel_idx, sort, order }; if (observerFilterActive) apiParams.observed_by = enabledObserverKeys; + if (includeSpam) apiParams.include_spam = true; const data = await apiGet('/api/v1/messages', apiParams, { signal }); const messages = dedupeBySignature(data.items || []); const total = data.total || 0; @@ -301,8 +317,9 @@ ${displayContent}`, container);
${fromPrimary}
-
+
${formatDateTimeShort(msg.received_at)} + ${spamBadge(msg)}
@@ -343,13 +360,19 @@ ${displayContent}`, container);
${fromPrimary}
- ${displayMessage} + +
+ ${displayMessage} + ${spamBadge(msg)} +
+ ${receiversBlock} `; }); + const spamParam = includeSpam ? { include_spam: 'true' } : {}; const paginationBlock = pagination(page, totalPages, '/messages', { - message_type, channel_idx, limit, sort, order, + message_type, channel_idx, limit, sort, order, ...spamParam, }); const filterFields = [ @@ -380,7 +403,21 @@ ${displayContent}`, container); `, ]; - const hasActiveFilters = message_type !== '' || channel_idx !== ''; + if (spamEnabled) { + filterFields.push(() => html` +
+ + +
`); + } + const hasActiveFilters = message_type !== '' || channel_idx !== '' || includeSpam; const existingDetails = container.querySelector('details.collapse'); const isFilterOpen = existingDetails ? existingDetails.open : hasActiveFilters; @@ -392,7 +429,7 @@ ${displayContent}`, container); defaultOpen: isFilterOpen, }); - const headerParams = { message_type, channel_idx, limit }; + const headerParams = { message_type, channel_idx, limit, ...spamParam }; const sortable = (label, sortKey) => sortableTableHeader(label, { sortKey, currentSort: sort, currentOrder: order, navigate, basePath: '/messages', params: headerParams, diff --git a/src/meshcore_hub/web/static/locales/en.json b/src/meshcore_hub/web/static/locales/en.json index bc5e71c..8395e11 100644 --- a/src/meshcore_hub/web/static/locales/en.json +++ b/src/meshcore_hub/web/static/locales/en.json @@ -200,6 +200,11 @@ "type_channel": "Channel", "type_contact": "Contact", "type_public": "Public", + "spam": { + "filter_label": "Spam", + "show": "Show potential spam", + "badge": "Spam" + }, "sort": { "newest": "Time (newest)", "oldest": "Time (oldest)", diff --git a/tests/conftest.py b/tests/conftest.py index 34fbc9a..011d1be 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,13 @@ """Shared pytest fixtures for all tests.""" +import os +import tempfile +from typing import Generator + import dotenv import pytest -from sqlalchemy import create_engine +from sqlalchemy import create_engine, text +from sqlalchemy.engine.url import make_url from sqlalchemy.orm import sessionmaker from meshcore_hub.common import config as config_module @@ -120,3 +125,97 @@ def db_session(db_engine): session = Session() yield session session.close() + + +@pytest.fixture(scope="session") +def test_db_path(): + """Session-scoped temporary SQLite database file path. + + One file per pytest session; engines below build schema on it once. + """ + fd, path = tempfile.mkstemp(suffix=".db") + os.close(fd) + yield path + if os.path.exists(path): + os.unlink(path) + + +@pytest.fixture(scope="session") +def db_backend() -> str: + """Active test database backend (``sqlite`` or ``postgres``). + + Controlled by ``TEST_DATABASE_BACKEND`` env var (default: ``sqlite``). + When ``postgres``, ``TEST_POSTGRES_URL`` must also be set. + """ + backend = os.environ.get("TEST_DATABASE_BACKEND", "sqlite").lower() + if backend not in ("sqlite", "postgres"): + raise ValueError( + f"TEST_DATABASE_BACKEND must be 'sqlite' or 'postgres', got: {backend}" + ) + return backend + + +@pytest.fixture(scope="session") +def db_url(db_backend: str, test_db_path: str, request) -> Generator[str, None, None]: + """Database URL for the active backend. + + For Postgres, each pytest-xdist worker gets its own database (e.g. + ``test_gw0``) to avoid truncation races between parallel workers. Shared by + the API and collector suites so both exercise the same backend. + """ + if db_backend == "postgres": + env_url = os.environ.get("TEST_POSTGRES_URL") + if not env_url: + pytest.skip( + "TEST_DATABASE_BACKEND=postgres but TEST_POSTGRES_URL is not set; " + "e.g. TEST_POSTGRES_URL=postgresql+psycopg2://postgres:postgres@localhost:55432/test" + ) + assert env_url is not None + + worker_id = "master" + if hasattr(request.config, "workerinput"): + worker_id = request.config.workerinput["workerid"] + + base_url = make_url(env_url) + worker_db = f"{base_url.database}_{worker_id}" + worker_url = base_url.set(database=worker_db).render_as_string( + hide_password=False + ) + + admin_url = base_url.set(database="postgres") + admin_engine = create_engine( + admin_url.render_as_string(hide_password=False), + isolation_level="AUTOCOMMIT", + ) + try: + with admin_engine.connect() as conn: + exists = conn.execute( + text("SELECT 1 FROM pg_database WHERE datname = :name"), + {"name": worker_db}, + ).scalar() + if not exists: + conn.execute(text(f'CREATE DATABASE "{worker_db}"')) + finally: + admin_engine.dispose() + + yield worker_url + + admin_engine = create_engine( + admin_url.render_as_string(hide_password=False), + isolation_level="AUTOCOMMIT", + ) + try: + with admin_engine.connect() as conn: + conn.execute( + text( + "SELECT pg_terminate_backend(pid) " + "FROM pg_stat_activity " + "WHERE datname = :name AND pid <> pg_backend_pid()" + ), + {"name": worker_db}, + ) + conn.execute(text(f'DROP DATABASE IF EXISTS "{worker_db}"')) + finally: + admin_engine.dispose() + else: + yield f"sqlite:///{test_db_path}" diff --git a/tests/test_api/conftest.py b/tests/test_api/conftest.py index 0e2b9e4..1e57db9 100644 --- a/tests/test_api/conftest.py +++ b/tests/test_api/conftest.py @@ -1,15 +1,16 @@ -"""API test fixtures.""" +"""API test fixtures. + +The ``db_backend`` / ``db_url`` / ``test_db_path`` fixtures that drive the +SQLite-vs-Postgres switch live in the shared ``tests/conftest.py`` so the +collector suite can reuse the same backend; they are inherited here. +""" -import os -import tempfile from contextlib import contextmanager from datetime import datetime, timezone -from typing import Generator import pytest from fastapi.testclient import TestClient -from sqlalchemy import create_engine, event as sa_event, text -from sqlalchemy.engine.url import make_url +from sqlalchemy import create_engine, event as sa_event from sqlalchemy.orm import sessionmaker from meshcore_hub.api.app import create_app @@ -34,99 +35,6 @@ from meshcore_hub.common.models import ( ) -@pytest.fixture(scope="session") -def test_db_path(): - """Session-scoped temporary database file path. - - One file per pytest session; the engine below builds schema on it once. - """ - fd, path = tempfile.mkstemp(suffix=".db") - os.close(fd) - yield path - if os.path.exists(path): - os.unlink(path) - - -@pytest.fixture(scope="session") -def db_backend() -> str: - """Active test database backend (``sqlite`` or ``postgres``). - - Controlled by ``TEST_DATABASE_BACKEND`` env var (default: ``sqlite``). - When ``postgres``, ``TEST_POSTGRES_URL`` must also be set. - """ - backend = os.environ.get("TEST_DATABASE_BACKEND", "sqlite").lower() - if backend not in ("sqlite", "postgres"): - raise ValueError( - f"TEST_DATABASE_BACKEND must be 'sqlite' or 'postgres', got: {backend}" - ) - return backend - - -@pytest.fixture(scope="session") -def db_url(db_backend: str, test_db_path: str, request) -> Generator[str, None, None]: - """Database URL for the active backend. - - For Postgres, each pytest-xdist worker gets its own database (e.g. - ``test_gw0``) to avoid truncation races between parallel workers. - """ - if db_backend == "postgres": - env_url = os.environ.get("TEST_POSTGRES_URL") - if not env_url: - pytest.skip( - "TEST_DATABASE_BACKEND=postgres but TEST_POSTGRES_URL is not set; " - "e.g. TEST_POSTGRES_URL=postgresql+psycopg2://postgres:postgres@localhost:55432/test" - ) - assert env_url is not None - - worker_id = "master" - if hasattr(request.config, "workerinput"): - worker_id = request.config.workerinput["workerid"] - - base_url = make_url(env_url) - worker_db = f"{base_url.database}_{worker_id}" - worker_url = base_url.set(database=worker_db).render_as_string( - hide_password=False - ) - - admin_url = base_url.set(database="postgres") - admin_engine = create_engine( - admin_url.render_as_string(hide_password=False), - isolation_level="AUTOCOMMIT", - ) - try: - with admin_engine.connect() as conn: - exists = conn.execute( - text("SELECT 1 FROM pg_database WHERE datname = :name"), - {"name": worker_db}, - ).scalar() - if not exists: - conn.execute(text(f'CREATE DATABASE "{worker_db}"')) - finally: - admin_engine.dispose() - - yield worker_url - - admin_engine = create_engine( - admin_url.render_as_string(hide_password=False), - isolation_level="AUTOCOMMIT", - ) - try: - with admin_engine.connect() as conn: - conn.execute( - text( - "SELECT pg_terminate_backend(pid) " - "FROM pg_stat_activity " - "WHERE datname = :name AND pid <> pg_backend_pid()" - ), - {"name": worker_db}, - ) - conn.execute(text(f'DROP DATABASE IF EXISTS "{worker_db}"')) - finally: - admin_engine.dispose() - else: - yield f"sqlite:///{test_db_path}" - - @pytest.fixture(scope="session") def api_db_engine(db_url: str, db_backend: str): """Session-scoped database engine. Schema is built once per pytest session. @@ -285,6 +193,26 @@ def app_with_auth(db_url, api_db_engine, mock_mqtt, mock_db_manager): yield app +@pytest.fixture(scope="module") +def app_spam(db_url, api_db_engine, mock_mqtt, mock_db_manager): + """Module-scoped app with spam detection enabled (no auth).""" + app = create_app( + database_url=db_url, + read_key=None, + admin_key=None, + spam_detection_enabled=True, + spam_score_threshold=0.6, + ) + _wire_overrides(app, api_db_engine, mock_mqtt, mock_db_manager) + yield app + + +@pytest.fixture +def client_spam(app_spam) -> TestClient: + """Test client with spam detection enabled.""" + return TestClient(app_spam, raise_server_exceptions=True) + + @pytest.fixture def client_no_auth(app_no_auth) -> TestClient: """Test client with no authentication.""" diff --git a/tests/test_api/test_messages.py b/tests/test_api/test_messages.py index d588573..55d8a2c 100644 --- a/tests/test_api/test_messages.py +++ b/tests/test_api/test_messages.py @@ -830,3 +830,73 @@ class TestMessageChannelVisibility: response = client_no_auth.get(f"/api/v1/messages/{msg.id}") assert response.status_code == 200 assert response.json()["observers"] == [] + + +class TestSpamFiltering: + """Tests for spam-score hide/show behaviour (GET /messages).""" + + @staticmethod + def _seed(api_db_session): + """Two clean messages + one likely-spam message.""" + now = datetime.now(timezone.utc) + api_db_session.add_all( + [ + Message( + message_type="direct", + pubkey_prefix="clean1", + text="clean one", + received_at=now, + spam_score=0.0, + ), + Message( + message_type="direct", + pubkey_prefix="unscor1", + text="unscored", + received_at=now, + spam_score=None, + ), + Message( + message_type="direct", + pubkey_prefix="spam1", + text="buy now buy now", + received_at=now, + spam_score=0.95, + ), + ] + ) + api_db_session.commit() + + def test_spam_hidden_by_default(self, client_spam, api_db_session): + """With detection on, high-score rows are hidden by default.""" + self._seed(api_db_session) + data = client_spam.get("/api/v1/messages").json() + texts = {item["text"] for item in data["items"]} + assert "buy now buy now" not in texts + assert "clean one" in texts + assert "unscored" in texts # null score is never hidden + assert data["total"] == 2 + + def test_include_spam_shows_all(self, client_spam, api_db_session): + """include_spam=true returns the flagged row with its score.""" + self._seed(api_db_session) + data = client_spam.get("/api/v1/messages?include_spam=true").json() + texts = {item["text"] for item in data["items"]} + assert "buy now buy now" in texts + assert data["total"] == 3 + spam_item = next(i for i in data["items"] if i["text"] == "buy now buy now") + assert spam_item["spam_score"] == 0.95 + + def test_master_switch_off_returns_all(self, client_no_auth, api_db_session): + """With detection disabled, even high-score rows are returned.""" + self._seed(api_db_session) + data = client_no_auth.get("/api/v1/messages").json() + texts = {item["text"] for item in data["items"]} + assert "buy now buy now" in texts + assert data["total"] == 3 + + def test_spam_score_in_response(self, client_no_auth, api_db_session): + """spam_score is surfaced on the message payload.""" + self._seed(api_db_session) + data = client_no_auth.get("/api/v1/messages").json() + clean = next(i for i in data["items"] if i["text"] == "clean one") + assert clean["spam_score"] == 0.0 diff --git a/tests/test_collector/conftest.py b/tests/test_collector/conftest.py index aa64012..99968de 100644 --- a/tests/test_collector/conftest.py +++ b/tests/test_collector/conftest.py @@ -1,4 +1,13 @@ -"""Fixtures for collector component tests.""" +"""Fixtures for collector component tests. + +The ``db_backend`` / ``db_url`` switch lives in the shared ``tests/conftest.py``. +The synchronous ``db_manager`` / ``db_session`` fixtures honour it so the spam +scorer, the re-scoring sweep, and the message handler are exercised on both +SQLite (default) and Postgres (``TEST_DATABASE_BACKEND=postgres``). On Postgres +they share the per-xdist-worker database with the API suite, so tables are +created idempotently and *truncated* (never dropped) at teardown to coexist with +the API's session-scoped engine. +""" import pytest from sqlalchemy import event as sa_event @@ -9,13 +18,32 @@ from meshcore_hub.common.database import DatabaseManager from meshcore_hub.common.models.base import Base +def _truncate_all(engine) -> None: + """Delete rows from every table in child-first order (FK-safe).""" + with engine.begin() as conn: + for table in reversed(Base.metadata.sorted_tables): + conn.execute(table.delete()) + + @pytest.fixture -def db_manager(): - """Create an in-memory database manager for testing.""" - manager = DatabaseManager("sqlite:///:memory:") - manager.create_tables() - yield manager - manager.dispose() +def db_manager(db_backend, db_url): + """Create a database manager for testing, honouring the active backend. + + Default (SQLite) uses an isolated in-memory database per test, exactly as + before. Postgres reuses the shared worker database with idempotent schema + + truncate-on-teardown. + """ + if db_backend == "postgres": + manager = DatabaseManager(db_url) + manager.create_tables() + yield manager + _truncate_all(manager.engine) + manager.dispose() + else: + manager = DatabaseManager("sqlite:///:memory:") + manager.create_tables() + yield manager + manager.dispose() @pytest.fixture diff --git a/tests/test_collector/test_handlers/test_message_spam.py b/tests/test_collector/test_handlers/test_message_spam.py new file mode 100644 index 0000000..f6cb626 --- /dev/null +++ b/tests/test_collector/test_handlers/test_message_spam.py @@ -0,0 +1,97 @@ +"""Tests for spam scoring wired into the message handler.""" + +from sqlalchemy import select + +import meshcore_hub.collector.handlers.message as handler_module +from meshcore_hub.collector.handlers.message import handle_channel_message +from meshcore_hub.collector.spam import SpamConfig +from meshcore_hub.common.models import Message + +ENABLED_CFG = SpamConfig( + enabled=True, + window_seconds=300, + path_hops=3, + min_path_hops=5, + path_threshold=5, + name_threshold=5, + weight_path=0.7, + weight_name=0.3, + score_threshold=0.6, + rescore_interval_seconds=120, +) + + +def _channel_payload(i, sender, *, path_len=6, path=None): + return { + "channel_idx": 0, + "text": "buy now", + "sender_name": sender, + "path_len": path_len, + "path_hashes": ( + path if path is not None else ["AA", "BB", "CC", "DD", "EE", "FF"] + ), + # Vary sender_timestamp so the event-hash dedup does not collapse the + # burst into a single row (text/channel/timestamp drive the hash). + "sender_timestamp": 1732820000 + i, + } + + +def _enable_spam(monkeypatch, cfg=ENABLED_CFG): + monkeypatch.setattr(handler_module, "get_spam_config", lambda: cfg) + + +class TestHandlerSpamDisabled: + def test_score_columns_null_when_disabled(self, db_manager, db_session): + """Default (feature off) leaves the spam columns null, as before.""" + handle_channel_message( + "a" * 64, "channel_msg_recv", _channel_payload(0, "bob1"), db_manager + ) + msg = db_session.execute(select(Message)).scalar_one() + assert msg.spam_score is None + assert msg.path_prefix is None + assert msg.sender_normalized is None + + +class TestHandlerSpamEnabled: + def test_populates_signals_and_score(self, db_manager, db_session, monkeypatch): + _enable_spam(monkeypatch) + handle_channel_message( + "a" * 64, "channel_msg_recv", _channel_payload(0, "bob1"), db_manager + ) + msg = db_session.execute(select(Message)).scalar_one() + assert msg.path_prefix == "AA,BB,CC" + assert msg.sender_normalized == "bob" + # First-ever message: no priors -> score 0.0 (not null). + assert msg.spam_score == 0.0 + + def test_burst_later_rows_flagged(self, db_manager, db_session, monkeypatch): + _enable_spam(monkeypatch) + # Rotating bob1..bob6 all normalize to "bob" on the same path. + for i in range(6): + handle_channel_message( + "a" * 64, + "channel_msg_recv", + _channel_payload(i, f"bob{i + 1}"), + db_manager, + ) + + msgs = db_session.execute(select(Message)).scalars().all() + assert len(msgs) == 6 + assert all(m.sender_normalized == "bob" for m in msgs) + # The last-inserted row saw 5 priors -> path+name saturate near threshold. + latest = max(msgs, key=lambda m: m.sender_timestamp) + assert latest.spam_score >= ENABLED_CFG.score_threshold + + def test_short_path_stores_null_prefix(self, db_manager, db_session, monkeypatch): + _enable_spam(monkeypatch) + handle_channel_message( + "a" * 64, + "channel_msg_recv", + _channel_payload(0, "bob1", path_len=2, path=["AA", "BB"]), + db_manager, + ) + msg = db_session.execute(select(Message)).scalar_one() + # Below the gate: path_prefix null, but sender still normalized. + assert msg.path_prefix is None + assert msg.sender_normalized == "bob" + assert msg.spam_score == 0.0 diff --git a/tests/test_collector/test_spam.py b/tests/test_collector/test_spam.py new file mode 100644 index 0000000..2085f0a --- /dev/null +++ b/tests/test_collector/test_spam.py @@ -0,0 +1,298 @@ +"""Tests for the spam scoring module (collector/spam.py). + +The DB-touching tests run against whichever backend the shared ``db_manager`` +fixture is configured for (SQLite by default, Postgres when +``TEST_DATABASE_BACKEND=postgres``). +""" + +from datetime import datetime, timedelta, timezone + +import pytest + +from meshcore_hub.collector.spam import ( + SpamConfig, + compute_path_prefix, + normalize_sender, + rescore_recent, + score_message, +) +from meshcore_hub.common.models import Message + +CFG = SpamConfig( + enabled=True, + window_seconds=300, + path_hops=3, + min_path_hops=5, + path_threshold=5, + name_threshold=5, + weight_path=0.7, + weight_name=0.3, + score_threshold=0.6, + rescore_interval_seconds=120, +) + + +# --------------------------------------------------------------------------- # +# Pure helpers (no DB) +# --------------------------------------------------------------------------- # + + +class TestNormalizeSender: + def test_strips_trailing_digits(self): + assert normalize_sender("bob17") == "bob" + + def test_strips_trailing_digit_and_space(self): + assert normalize_sender("Bob 2") == "bob" + + def test_lowercases(self): + assert normalize_sender("ALICE") == "alice" + + def test_none_for_empty(self): + assert normalize_sender("") is None + assert normalize_sender(None) is None + + def test_none_for_all_digits(self): + assert normalize_sender("12345") is None + + def test_keeps_interior_digits(self): + assert normalize_sender("c3po") == "c3po" + + +class TestComputePathPrefix: + def test_joins_first_n_hops(self): + assert compute_path_prefix(["AA", "BB", "CC", "DD"], 3) == "AA,BB,CC" + + def test_shorter_than_hops(self): + assert compute_path_prefix(["AA"], 3) == "AA" + + def test_none_for_empty(self): + assert compute_path_prefix(None, 3) is None + assert compute_path_prefix([], 3) is None + + def test_none_for_zero_hops(self): + assert compute_path_prefix(["AA", "BB"], 0) is None + + +# --------------------------------------------------------------------------- # +# DB-touching scorer +# --------------------------------------------------------------------------- # + + +def _make_message(received_at, *, path_prefix, sender_normalized, path_len=6): + return Message( + message_type="channel", + channel_idx=0, + text="test", + path_len=path_len, + path_prefix=path_prefix, + sender_normalized=sender_normalized, + received_at=received_at, + ) + + +class TestScoreMessage: + def test_first_message_scores_zero(self, db_session): + now = datetime.now(timezone.utc) + result = score_message( + db_session, + path_prefix="AA,BB,CC", + sender_normalized="bob", + path_len=6, + received_at=now, + cfg=CFG, + ) + assert result.score == 0.0 + assert result.path_count == 0 + assert result.name_count == 0 + + def test_identical_burst_crosses_threshold(self, db_session): + now = datetime.now(timezone.utc) + # Seed a burst of identical path+sender priors within the window. + for i in range(6): + db_session.add( + _make_message( + now - timedelta(seconds=10 + i), + path_prefix="AA,BB,CC", + sender_normalized="bob", + ) + ) + db_session.commit() + + result = score_message( + db_session, + path_prefix="AA,BB,CC", + sender_normalized="bob", + path_len=6, + received_at=now, + cfg=CFG, + ) + # path saturates (>=5) -> 0.7, name saturates -> 0.3 => 1.0 + assert result.score >= CFG.score_threshold + assert result.path_count == 6 + assert result.name_count == 6 + + def test_busy_path_diverse_senders_not_flagged(self, db_session): + """Many different senders on the same path should not score high.""" + now = datetime.now(timezone.utc) + for i in range(8): + db_session.add( + _make_message( + now - timedelta(seconds=5 + i), + path_prefix="AA,BB,CC", + sender_normalized=f"user{i}", # already normalized, distinct + ) + ) + db_session.commit() + + result = score_message( + db_session, + path_prefix="AA,BB,CC", + sender_normalized="freshname", + path_len=6, + received_at=now, + cfg=CFG, + ) + # Joint (path, sender) count is 0 for this new sender; name count is 0. + assert result.path_count == 0 + assert result.name_count == 0 + assert result.score == 0.0 + + def test_outside_window_not_counted(self, db_session): + now = datetime.now(timezone.utc) + for i in range(6): + db_session.add( + _make_message( + now - timedelta(seconds=400 + i), # older than 300s window + path_prefix="AA,BB,CC", + sender_normalized="bob", + ) + ) + db_session.commit() + + result = score_message( + db_session, + path_prefix="AA,BB,CC", + sender_normalized="bob", + path_len=6, + received_at=now, + cfg=CFG, + ) + assert result.score == 0.0 + + def test_short_path_relies_on_name_only(self, db_session): + """Below the min-path gate the path signal is off; name stands alone. + + Name-only must be able to cross the threshold, otherwise zero-hop / local + spam (no usable path) could never be flagged. + """ + now = datetime.now(timezone.utc) + # Priors share the sender but were stored with null path_prefix (short). + for i in range(6): + db_session.add( + _make_message( + now - timedelta(seconds=5 + i), + path_prefix=None, + sender_normalized="bob", + path_len=2, + ) + ) + db_session.commit() + + result = score_message( + db_session, + path_prefix=None, # gated to null by the handler for short paths + sender_normalized="bob", + path_len=2, + received_at=now, + cfg=CFG, + ) + # path disabled -> name signal at full weight; saturated count -> 1.0 + assert result.path_count == 0 + assert result.name_count == 6 + assert result.score == pytest.approx(1.0) + assert result.score >= CFG.score_threshold + + def test_short_path_name_scales_below_saturation(self, db_session): + """Name-only score scales with count and crosses threshold at name_thr.""" + now = datetime.now(timezone.utc) + # 3 priors against name_threshold=5 -> 0.6 (== threshold). + for i in range(3): + db_session.add( + _make_message( + now - timedelta(seconds=5 + i), + path_prefix=None, + sender_normalized="bob", + path_len=2, + ) + ) + db_session.commit() + + result = score_message( + db_session, + path_prefix=None, + sender_normalized="bob", + path_len=2, + received_at=now, + cfg=CFG, + ) + assert result.name_count == 3 + assert result.score == pytest.approx(0.6) + + def test_no_sender_scores_zero(self, db_session): + now = datetime.now(timezone.utc) + result = score_message( + db_session, + path_prefix="AA,BB,CC", + sender_normalized=None, + path_len=6, + received_at=now, + cfg=CFG, + ) + assert result.score == 0.0 + + +class TestRescoreRecent: + def test_leading_edge_rescored_with_hindsight(self, db_session): + """Rows that scored low online get raised once peers arrive.""" + now = datetime.now(timezone.utc) + # Insert a burst, all with spam_score=0.0 (as the online path would set + # for the leading edge), out of natural order. + msgs = [] + for i in range(6): + m = _make_message( + now - timedelta(seconds=5 * i), + path_prefix="AA,BB,CC", + sender_normalized="bob", + ) + m.spam_score = 0.0 + msgs.append(m) + db_session.add(m) + db_session.commit() + + updated = rescore_recent(db_session, CFG, now=now) + db_session.commit() + + assert updated >= 1 + # The earliest row now sees its later peers symmetrically -> flagged. + earliest = min(msgs, key=lambda m: m.received_at) + db_session.refresh(earliest) + assert earliest.spam_score >= CFG.score_threshold + + def test_idempotent(self, db_session): + now = datetime.now(timezone.utc) + for i in range(6): + m = _make_message( + now - timedelta(seconds=5 * i), + path_prefix="AA,BB,CC", + sender_normalized="bob", + ) + db_session.add(m) + db_session.commit() + + first = rescore_recent(db_session, CFG, now=now) + db_session.commit() + second = rescore_recent(db_session, CFG, now=now) + db_session.commit() + + assert first >= 1 + assert second == 0 # nothing changes on a second pass