diff --git a/.env.example b/.env.example index 49c53c5..e6d5061 100644 --- a/.env.example +++ b/.env.example @@ -279,6 +279,18 @@ DATA_RETENTION_INTERVAL_HOURS=24 # enabled, so disabling capture lets existing rows drain. # RAW_PACKET_RETENTION_DAYS=7 +# ------------------------- +# Observer Ingestion Filter +# ------------------------- +# Restrict which remote observers may ingest events, keyed on the observer's +# public key (the segment of its LetsMesh upload topic). Both lists +# are comma-separated. The allowlist takes precedence over the denylist, and +# matching is case-insensitive PREFIX matching (a full 64-char key or a shorter +# prefix both work). Blocked packets are dropped before any decode or DB write. +# Leave both empty to accept all observers (default). +# OBSERVER_ALLOWLIST= +# OBSERVER_DENYLIST= + # ------------------- # Spam Detection # ------------------- diff --git a/docker-compose.yml b/docker-compose.yml index d991c1c..3405949 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -239,6 +239,10 @@ 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} + # Observer ingestion filter (allow/deny remote observers by public key/prefix). + # Allowlist overrides denylist; both empty accepts all observers (default). + - OBSERVER_ALLOWLIST=${OBSERVER_ALLOWLIST:-} + - OBSERVER_DENYLIST=${OBSERVER_DENYLIST:-} # Spam detection (derived from FEATURE_SPAM_DETECTION so one var drives the # backend switch and the web toggle together). On by default; opt out with # FEATURE_SPAM_DETECTION=false. diff --git a/docs/configuration.md b/docs/configuration.md index 0ab694e..d08e997 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -72,6 +72,27 @@ The collector subscribes to MQTT events and persists them to the database. For p | --- | --- | --- | | `CHANNEL_REFRESH_INTERVAL_SECONDS` | `300` | Seconds between channel-key refresh from the database (minimum `10`) | +### Observer Ingestion Filters + +Anyone can contribute as a remote [observer](observer.md) by publishing decoded +packets to your MQTT broker. Each observer identifies itself by its public key, +which is the third segment of its LetsMesh upload topic +(`///`). These two variables let an operator +restrict which observers are ingested. + +| Variable | Default | Description | +| --- | --- | --- | +| `OBSERVER_ALLOWLIST` | _(none)_ | Comma-separated observer public keys (or key prefixes) permitted to ingest. If set, **only** matching observers are accepted and `OBSERVER_DENYLIST` is ignored. | +| `OBSERVER_DENYLIST` | _(none)_ | Comma-separated observer public keys (or key prefixes) blocked from ingesting. Applies only when `OBSERVER_ALLOWLIST` is unset/empty. | + +- **Precedence:** the allowlist overrides the denylist. With an allowlist set, + any observer not on it is blocked. +- **Matching:** case-insensitive **prefix** matching, so you can use a full + 64-char key or a shorter prefix (e.g. `OBSERVER_DENYLIST=F4762185`). +- **Default:** both empty accepts all observers (current behaviour). +- **Effect:** a blocked observer's packets are dropped at ingest, **before** any + decode or database write — nothing is persisted or forwarded for them. + ## Webhooks The collector can forward events (advertisements, messages) to external HTTP endpoints via webhooks with configurable URLs, secrets, retries, and timeouts. For URL routing rules, secret handling, retry behaviour, and payload format, see [webhooks.md](webhooks.md). diff --git a/docs/observer.md b/docs/observer.md index 6b93381..829e666 100644 --- a/docs/observer.md +++ b/docs/observer.md @@ -6,6 +6,8 @@ This document covers the local packet-capture observer (the `observer` compose p > **Prerequisite:** Your MQTT broker must be accessible to remote observers. In production, this means exposing the WebSocket listener via a reverse proxy with TLS (e.g., `wss://mqtt.example.com/mqtt`). +> **Restricting which observers are accepted:** because anyone with broker access can publish as an observer, Hub operators can gate ingestion by observer public key using `OBSERVER_ALLOWLIST` / `OBSERVER_DENYLIST`. See [configuration.md → Observer Ingestion Filters](configuration.md#observer-ingestion-filters). + ## Example: Contribute to MeshCore Hub, LetsMesh, and MeshRank A ready-made Docker Compose setup is provided in `contrib/packetcapture/`. Download it and configure: diff --git a/docs/plans/20260625-2005-observer-ingestion-filters/plan.md b/docs/plans/20260625-2005-observer-ingestion-filters/plan.md new file mode 100644 index 0000000..0b2e593 --- /dev/null +++ b/docs/plans/20260625-2005-observer-ingestion-filters/plan.md @@ -0,0 +1,379 @@ +# Observer Ingestion Filters (Allow / Deny List) + +**Branch:** `feat/observer-ingestion-filters` +**Date:** 2026-06-25 + +## Problem + +Anyone can contribute as a remote **Observer** to a MeshCore Hub by publishing +decoded packets to the Hub's MQTT broker using JWT auth. There is currently no +way for a Hub operator to restrict *which* observers are allowed to have their +events ingested. We want an allow / deny list, configured via environment +variables, keyed on the observer's public key. + +## How observers identify themselves (confirmed) + +Remote observers publish to **LetsMesh upload topics**: + +``` +///(packets|status|internal) +``` + +A real production topic: + +``` +meshcore/STN/F4762185BBB684510B2E3D41568300869DB5E75B284448145475F7708C1EF408/packets + │ │ └─ observer public key (64-char hex, UPPERCASE) └─ feed + prefix iata +``` + +The `` segment is the observer's identity. It is parsed by +`TopicBuilder.parse_letsmesh_upload_topic()` +(`src/meshcore_hub/common/mqtt.py:130`) which returns `(public_key, feed_type)`. + +The collector subscriber consumes these in +`Subscriber._handle_mqtt_message()` (`src/meshcore_hub/collector/subscriber.py:203`), +which calls `_normalize_letsmesh_event()` and obtains `public_key` (the observer +key) before performing **raw packet capture** and **event dispatch**. This is the +single choke point through which every observer-sourced event flows, so it is the +correct place to apply the filter. + +In tests the observer key is a full 64-char hex string (e.g. `"a" * 64`); in +production it is **upper-case** 64-char hex (see example above). This is the +direct reason the filter normalises both the list entries and the topic key to +lower-case before comparing — an operator can paste the key in either case, or +use a leading prefix such as `OBSERVER_DENYLIST=F4762185`, and it will still +match. + +## Decisions (confirmed with user) + +1. **Env var naming:** `OBSERVER_ALLOWLIST` / `OBSERVER_DENYLIST` — **no** + `COLLECTOR_` prefix. This matches the existing pydantic-settings convention + where the field name maps directly to the upper-cased env var + (`RAW_PACKET_CAPTURE_ENABLED`, `DATA_RETENTION_ENABLED`, `SPAM_*`). A + `COLLECTOR_`-prefixed name would be the only one in the codebase and would + require an `env_prefix`/alias. +2. **Matching:** **prefix match, case-insensitive.** An observer key matches a + list entry if the (lower-cased) observer key *starts with* the (lower-cased, + trimmed) entry. This lets operators use short prefixes (e.g. `a1b2c3`) as well + as full keys. +3. **Precedence:** If `OBSERVER_ALLOWLIST` is non-empty it takes effect and + `OBSERVER_DENYLIST` is **ignored**. If the allowlist is empty, the denylist + applies. If both are empty, all observers are ingested (current behaviour). + +## Semantics + +Given a normalised observer key `k`: + +| Allowlist | Denylist | Result | +| --- | --- | --- | +| empty | empty | **allow** (default, unchanged behaviour) | +| non-empty | (ignored) | allow **iff** `k` prefix-matches any allowlist entry | +| empty | non-empty | allow **unless** `k` prefix-matches any denylist entry | + +A blocked event is dropped *before* raw-packet capture and before handler +dispatch, and is logged at `DEBUG` (to avoid log spam from a noisy blocked +observer). It is not persisted anywhere. + +### Drop guarantee + +The filter check is the **first** statement in `_handle_mqtt_message`, so a +blocked observer's packet skips every stage below it: + +| Stage | Runs for blocked packet? | +| --- | --- | +| LetsMesh RF decode/decrypt (`_letsmesh_decoder.decode_payload`) | No | +| `raw_packets` insert (`_maybe_capture_raw_packet`) | No | +| Event handlers → `messages` / `advertisements` / `telemetry` / `event_observers` | No | +| Node upsert / `is_observer` flag | No | +| Webhook dispatch | No | + +The **only** processing a blocked packet incurs is the unavoidable +transport-level `json.loads()` of the MQTT envelope in +`MQTTClient._on_message` (`common/mqtt.py:230`) — which happens for every +message before any handler runs and is shared by all subscribers — plus the +cheap topic-string split the filter does to read the observer key. Neither +performs RF packet decoding nor writes to the database. A blocked packet is +therefore dropped entirely: not decoded, not persisted, not forwarded. + +Empty / whitespace-only entries are discarded when parsing the comma-delimited +lists, so `OBSERVER_DENYLIST=` or trailing commas are harmless. + +## Implementation + +### 1. New module: `src/meshcore_hub/collector/observer_filter.py` + +A small, dependency-free, unit-testable helper. + +```python +"""Allow/deny filtering of observer-sourced events by observer public key.""" + +from __future__ import annotations + +from dataclasses import dataclass + + +def _normalise(entries: list[str] | None) -> tuple[str, ...]: + """Lower-case, strip, and drop empty entries.""" + if not entries: + return () + return tuple(e.strip().lower() for e in entries if e and e.strip()) + + +@dataclass(frozen=True) +class ObserverFilter: + """Decides whether an observer's events should be ingested. + + Allowlist takes precedence over denylist. Matching is case-insensitive + prefix matching against the observer public key. + """ + + allowlist: tuple[str, ...] = () + denylist: tuple[str, ...] = () + + @classmethod + def from_lists( + cls, + allowlist: list[str] | None = None, + denylist: list[str] | None = None, + ) -> "ObserverFilter": + return cls(allowlist=_normalise(allowlist), denylist=_normalise(denylist)) + + @property + def active(self) -> bool: + return bool(self.allowlist or self.denylist) + + def is_allowed(self, public_key: str | None) -> bool: + if not self.allowlist and not self.denylist: + return True + key = (public_key or "").strip().lower() + if self.allowlist: + return any(key.startswith(entry) for entry in self.allowlist) + return not any(key.startswith(entry) for entry in self.denylist) +``` + +> Note on empty-string prefix: `_normalise` drops empty entries, so an +> all-empty list yields `()` and `startswith` is never called with `""`. This +> prevents an accidental "match everything" from a stray comma. + +### 2. Config: `src/meshcore_hub/common/config.py` (`CollectorSettings`) + +Add two raw string fields plus parsed-list properties. Strings (not `list[str]`) +because pydantic-settings parses complex types from env as JSON, which breaks +comma-separated input. + +```python +# Observer ingestion filtering (allow/deny by observer public key). +# Allowlist takes precedence over denylist. Empty = no restriction. +observer_allowlist: str = Field( + default="", + description=( + "Comma-separated observer public keys (or prefixes) allowed to ingest. " + "If set, overrides OBSERVER_DENYLIST." + ), +) +observer_denylist: str = Field( + default="", + description=( + "Comma-separated observer public keys (or prefixes) blocked from ingesting. " + "Ignored when OBSERVER_ALLOWLIST is set." + ), +) + +@property +def observer_allowlist_keys(self) -> list[str]: + return [k.strip() for k in self.observer_allowlist.split(",") if k.strip()] + +@property +def observer_denylist_keys(self) -> list[str]: + return [k.strip() for k in self.observer_denylist.split(",") if k.strip()] +``` + +### 3. Subscriber: `src/meshcore_hub/collector/subscriber.py` + +- Add `observer_filter: ObserverFilter | None = None` parameter to + `Subscriber.__init__`; store as `self._observer_filter = observer_filter or ObserverFilter()`. + Log at startup when active (mirroring the existing raw-capture log line), e.g. + `"Observer filter active: %d allow, %d deny"`. +- In `_handle_mqtt_message()`, **before** `_normalize_letsmesh_event()` (to avoid + decode work for blocked observers), cheaply extract the observer key from the + topic and short-circuit: + +```python +def _handle_mqtt_message(self, topic, pattern, payload): + if self._observer_filter.active: + parsed_topic = self.mqtt.topic_builder.parse_letsmesh_upload_topic(topic) + if parsed_topic: + observer_key, _feed = parsed_topic + if not self._observer_filter.is_allowed(observer_key): + logger.debug( + "Dropping event from blocked observer %s...", observer_key[:12] + ) + return + # ... existing normalize + capture + dispatch ... +``` + + Rationale for placing the check here rather than inside + `_normalize_letsmesh_event`: it covers raw-packet capture too (which also keys + off `public_key`), and skips the decode entirely for blocked observers. When + the filter is inactive (`active == False`) there is **zero** added work on the + hot path. + +- Thread the parameter through the two factory functions in the same file: + - `create_subscriber(...)` — add `observer_filter` param, pass to `Subscriber(...)`. + - `run_collector(...)` — add `observer_filter` param, pass to `create_subscriber(...)`. + +### 4. CLI wiring: `src/meshcore_hub/collector/cli.py` + +In `_run_collector_service()` build the filter from settings and pass it into +`run_collector(...)`: + +```python +from meshcore_hub.collector.observer_filter import ObserverFilter +... +run_collector( + ..., + raw_packet_capture_enabled=settings.raw_packet_capture_enabled, + raw_packet_retention_days=settings.effective_raw_packet_retention_days, + observer_filter=ObserverFilter.from_lists( + allowlist=settings.observer_allowlist_keys, + denylist=settings.observer_denylist_keys, + ), +) +``` + +## Tests + +### New: `tests/test_collector/test_observer_filter.py` (unit, pure logic) + +- empty allow + empty deny → allows any key +- allowlist set → allows listed key; blocks unlisted; denylist ignored even if it + would block the same key (precedence) +- denylist set (no allowlist) → blocks listed key; allows others +- case-insensitivity (upper-case env entry vs lower-case key and vice versa) +- prefix matching: short prefix matches longer key; non-matching prefix does not +- whitespace / empty-entry hygiene (`"", " ", "a,"` etc.) — no spurious matches +- `active` property true/false +- `from_lists(None, None)` behaves as empty + +### Extend: `tests/test_collector/test_subscriber.py` + +Following the existing `_handle_mqtt_message` test style (mock +`_normalize_letsmesh_event`, real topic): + +- observer on denylist → handler **not** called, no raw-packet row written +- observer on allowlist → handler called normally +- observer not on allowlist (allowlist active) → handler not called +- filter inactive (default) → existing behaviour unchanged (regression guard) +- blocked observer with `raw_packet_capture_enabled=True` → no raw row written + (confirms the check precedes capture) + +### Extend: `tests/test_common/` config test (if present) + +- `OBSERVER_ALLOWLIST="aaa, bbb ,"` → `observer_allowlist_keys == ["aaa", "bbb"]` +- defaults → empty lists, `active` false + +## Documentation + +### `docs/configuration.md` + +Add a new subsection under **Collector** (after the Collector table, before +Webhooks) titled **Observer Ingestion Filters**: + +| Variable | Default | Description | +| --- | --- | --- | +| `OBSERVER_ALLOWLIST` | _(none)_ | Comma-separated observer public keys (or key prefixes) permitted to ingest. If set, only matching observers are accepted and `OBSERVER_DENYLIST` is ignored. | +| `OBSERVER_DENYLIST` | _(none)_ | Comma-separated observer public keys (or key prefixes) blocked from ingesting. Applies only when `OBSERVER_ALLOWLIST` is unset/empty. | + +Prose: explain observer = LetsMesh upload topic publisher identified by its +public key; precedence (allow overrides deny); case-insensitive prefix matching; +blocked events are dropped before capture/persistence; default (both empty) = +accept all. Cross-link to [observer.md](../observer.md). + +### `docs/observer.md` + +Add a short "Restricting which observers are accepted" note linking back to the +configuration table, so operators running observers know contributions can be +gated. + +### `docs/upgrading.md` + +Add a **new `## v0.16.0` section at the top** of the file (above the current +`## v0.15.0`), since this ships in the next release. Under it add an **Observer +Ingestion Filters** subsection: new optional `OBSERVER_ALLOWLIST` / +`OBSERVER_DENYLIST` vars; **non-breaking**, defaults preserve existing +accept-all behaviour; note allow-over-deny precedence, case-insensitive prefix +matching, and that blocked observer packets are dropped before any decode or +persistence. Cross-link to [configuration.md](configuration.md) and +[observer.md](observer.md). + +### `.env.example` + +Under the `# COLLECTOR SETTINGS` block (near `DATA_RETENTION_*` / +`RAW_PACKET_*`), add commented examples: + +```ini +# Observer ingestion filter (allow/deny remote observers by public key or prefix). +# Allowlist takes precedence over denylist. Matching is case-insensitive prefix +# matching. Leave both empty to accept all observers (default). +# OBSERVER_ALLOWLIST= +# OBSERVER_DENYLIST= +``` + +### Docker Compose files + +Only the **base** `docker-compose.yml` defines the collector's `environment:` +block, so it is the only compose file that needs the new vars. Wire both into +the **collector** service `environment:` block (near `RAW_PACKET_*`) so Compose +users can set them from their `.env`: + +```yaml + # Observer ingestion filter (allow/deny remote observers by public key/prefix) + - OBSERVER_ALLOWLIST=${OBSERVER_ALLOWLIST:-} + - OBSERVER_DENYLIST=${OBSERVER_DENYLIST:-} +``` + +The other compose files are intentionally **not** modified: + +| File | Why no change | +| --- | --- | +| `docker-compose.dev.yml` | Overlay; only overrides `pull_policy` / `depends_on` / `ports`. Compose merges `environment` additively, so it inherits the base collector env block. | +| `docker-compose.prod.yml` | Overlay; only overrides `networks`. Inherits base env block. | +| `docker-compose.traefik.yml` | Overlay; only adds Traefik labels. Inherits base env block. | +| `contrib/packetcapture/docker-compose.yml` | This is the **observer-side** packet-capture publisher, not the Hub collector. The allow/deny lists are consumed by the ingesting Hub collector, so this file gets nothing. | + +## Files touched + +| File | Change | +| --- | --- | +| `src/meshcore_hub/collector/observer_filter.py` | **new** — `ObserverFilter` | +| `src/meshcore_hub/common/config.py` | add fields + parsed-list properties | +| `src/meshcore_hub/collector/subscriber.py` | filter param on `Subscriber`, `create_subscriber`, `run_collector`; early-drop in `_handle_mqtt_message` | +| `src/meshcore_hub/collector/cli.py` | build `ObserverFilter`, pass to `run_collector` | +| `tests/test_collector/test_observer_filter.py` | **new** unit tests | +| `tests/test_collector/test_subscriber.py` | integration tests for drop/allow | +| `tests/test_common/...` (config test) | env-var parsing tests | +| `docs/configuration.md` | new Observer Ingestion Filters table + prose | +| `docs/observer.md` | short cross-reference note | +| `docs/upgrading.md` | release note | +| `.env.example` | commented example vars | +| `docker-compose.yml` | collector env wiring | + +## Out of scope / non-goals + +- No UI surface for managing the lists (env-driven only, like other collector knobs). +- No database-backed dynamic list; restart picks up env changes (consistent with + existing collector settings). +- Native `event` topics (`//event/#`) are not currently + subscribed by this collector path — the filter targets LetsMesh observer + uploads, which is where untrusted JWT contributors publish. If native event + ingestion is added later, the same `is_allowed` check can be applied there. + +## Validation + +- `make test` (or `pytest tests/test_collector/test_observer_filter.py + tests/test_collector/test_subscriber.py`) green. +- `ruff` / `mypy` clean (the new module is fully typed; frozen dataclass). +- Manual sanity: set `OBSERVER_DENYLIST=`, confirm its + events stop appearing while others continue; clear it and set + `OBSERVER_ALLOWLIST` to a different key, confirm only that observer ingests. diff --git a/docs/upgrading.md b/docs/upgrading.md index 7c4f4c5..c137631 100644 --- a/docs/upgrading.md +++ b/docs/upgrading.md @@ -2,6 +2,23 @@ This guide covers upgrading from a previous MeshCore Hub release to the current version. Check the relevant version section below before upgrading. +## v0.16.0 + +### Observer Ingestion Filters (allow/deny remote observers) + +Remote observers contribute to the Hub by publishing decoded packets to your MQTT broker, and anyone with broker access can do so. You can now restrict which observers are ingested by their public key with two new **optional** collector variables: + +| Variable | Default | Description | +| --- | --- | --- | +| `OBSERVER_ALLOWLIST` | _(none)_ | Comma-separated observer public keys (or prefixes) permitted to ingest. If set, only matching observers are accepted and `OBSERVER_DENYLIST` is ignored. | +| `OBSERVER_DENYLIST` | _(none)_ | Comma-separated observer public keys (or prefixes) blocked from ingesting. Applies only when `OBSERVER_ALLOWLIST` is unset. | + +- **Non-breaking:** both default to empty, which preserves the existing accept-all behaviour. No action is required to keep current behaviour. +- The allowlist **takes precedence** over the denylist, and matching is **case-insensitive prefix** matching (a full 64-char key or a shorter prefix both work). +- A blocked observer's packets are dropped at ingest, **before** any decode or database write — nothing is persisted or forwarded for them. + +In Docker Compose, set `OBSERVER_ALLOWLIST` / `OBSERVER_DENYLIST` in your `.env`; they are wired into the collector service. See [configuration.md → Observer Ingestion Filters](configuration.md#observer-ingestion-filters) and [observer.md](observer.md). + ## v0.15.0 ### Spam Detection (score, hide, and toggle likely-spam messages) diff --git a/src/meshcore_hub/collector/cli.py b/src/meshcore_hub/collector/cli.py index 8fc0793..e2baae9 100644 --- a/src/meshcore_hub/collector/cli.py +++ b/src/meshcore_hub/collector/cli.py @@ -236,6 +236,7 @@ def _run_collector_service( else: click.echo("Webhooks: None configured") + from meshcore_hub.collector.observer_filter import ObserverFilter from meshcore_hub.collector.subscriber import run_collector # Show cleanup configuration @@ -258,6 +259,18 @@ def _run_collector_service( if settings.data_retention_enabled or settings.node_cleanup_enabled: click.echo(f" Interval: {settings.data_retention_interval_hours} hours") + click.echo("") + if settings.observer_allowlist_keys: + click.echo( + f"Observer filter: ALLOWLIST ({len(settings.observer_allowlist_keys)} entries)" + ) + elif settings.observer_denylist_keys: + click.echo( + f"Observer filter: DENYLIST ({len(settings.observer_denylist_keys)} entries)" + ) + else: + click.echo("Observer filter: Disabled (accepting all observers)") + click.echo("") builtin_keys = len(LetsMeshPacketDecoder.BUILTIN_CHANNEL_KEYS) click.echo(f"Packet decoder: {builtin_keys} built-in keys, loading from database") @@ -283,6 +296,10 @@ def _run_collector_service( channel_refresh_interval_seconds=settings.channel_refresh_interval_seconds, raw_packet_capture_enabled=settings.raw_packet_capture_enabled, raw_packet_retention_days=settings.effective_raw_packet_retention_days, + observer_filter=ObserverFilter.from_lists( + allowlist=settings.observer_allowlist_keys, + denylist=settings.observer_denylist_keys, + ), ) diff --git a/src/meshcore_hub/collector/observer_filter.py b/src/meshcore_hub/collector/observer_filter.py new file mode 100644 index 0000000..0f23411 --- /dev/null +++ b/src/meshcore_hub/collector/observer_filter.py @@ -0,0 +1,65 @@ +"""Allow/deny filtering of observer-sourced events by observer public key. + +Remote observers identify themselves via the public-key segment of their +LetsMesh upload topic (``///``). This module +decides whether a given observer's events should be ingested, based on operator- +configured allow/deny lists. + +Matching is case-insensitive prefix matching: an observer key matches a list +entry when the (lower-cased) key starts with the (lower-cased, trimmed) entry, +so operators may use full 64-char keys or shorter prefixes. The allowlist takes +precedence over the denylist. +""" + +from __future__ import annotations + +from dataclasses import dataclass + + +def _normalise(entries: list[str] | None) -> tuple[str, ...]: + """Lower-case, strip, and drop empty entries.""" + if not entries: + return () + return tuple(e.strip().lower() for e in entries if e and e.strip()) + + +@dataclass(frozen=True) +class ObserverFilter: + """Decides whether an observer's events should be ingested. + + Allowlist takes precedence over denylist. Matching is case-insensitive + prefix matching against the observer public key. + """ + + allowlist: tuple[str, ...] = () + denylist: tuple[str, ...] = () + + @classmethod + def from_lists( + cls, + allowlist: list[str] | None = None, + denylist: list[str] | None = None, + ) -> "ObserverFilter": + """Build a filter from raw (un-normalised) allow/deny entry lists.""" + return cls(allowlist=_normalise(allowlist), denylist=_normalise(denylist)) + + @property + def active(self) -> bool: + """True when either list is non-empty (i.e. filtering is in effect).""" + return bool(self.allowlist or self.denylist) + + def is_allowed(self, public_key: str | None) -> bool: + """Return whether the given observer public key may ingest events. + + - Both lists empty: allow everything (default behaviour). + - Allowlist non-empty: allow only keys matching an allowlist entry + (the denylist is ignored). + - Allowlist empty, denylist non-empty: allow unless the key matches a + denylist entry. + """ + if not self.allowlist and not self.denylist: + return True + key = (public_key or "").strip().lower() + if self.allowlist: + return any(key.startswith(entry) for entry in self.allowlist) + return not any(key.startswith(entry) for entry in self.denylist) diff --git a/src/meshcore_hub/collector/subscriber.py b/src/meshcore_hub/collector/subscriber.py index ecb6dd7..ce6c678 100644 --- a/src/meshcore_hub/collector/subscriber.py +++ b/src/meshcore_hub/collector/subscriber.py @@ -23,6 +23,7 @@ from meshcore_hub.common.health import HealthReporter from meshcore_hub.common.mqtt import MQTTClient, MQTTConfig from meshcore_hub.collector.letsmesh_decoder import LetsMeshPacketDecoder from meshcore_hub.collector.letsmesh_normalizer import LetsMeshNormalizer +from meshcore_hub.collector.observer_filter import ObserverFilter if TYPE_CHECKING: from meshcore_hub.collector.webhook import WebhookDispatcher @@ -50,6 +51,7 @@ class Subscriber(LetsMeshNormalizer): channel_refresh_interval_seconds: int = 300, raw_packet_capture_enabled: bool = False, raw_packet_retention_days: int = 7, + observer_filter: Optional[ObserverFilter] = None, ): """Initialize subscriber. @@ -65,6 +67,8 @@ class Subscriber(LetsMeshNormalizer): channel_refresh_interval_seconds: Seconds between channel key refresh raw_packet_capture_enabled: Capture every packets-feed packet to raw_packets raw_packet_retention_days: Days to retain raw packets + observer_filter: Allow/deny filter for observer-sourced events + (defaults to an inactive filter that accepts all observers) """ self.mqtt = mqtt_client self.db = db_manager @@ -95,6 +99,14 @@ class Subscriber(LetsMeshNormalizer): "enabled" if raw_packet_capture_enabled else "disabled", raw_packet_retention_days, ) + # Observer ingestion filter (allow/deny by observer public key) + self._observer_filter = observer_filter or ObserverFilter() + if self._observer_filter.active: + logger.info( + "Observer filter active: %d allow, %d deny", + len(self._observer_filter.allowlist), + len(self._observer_filter.denylist), + ) # Channel key refresh self._channel_refresh_interval_seconds = channel_refresh_interval_seconds self._channel_refresh_thread: Optional[threading.Thread] = None @@ -213,6 +225,22 @@ class Subscriber(LetsMeshNormalizer): pattern: Subscription pattern payload: Message payload """ + # Apply the observer allow/deny filter first, before any decode, + # raw-packet capture, or handler dispatch. Blocked observers' packets + # are dropped entirely and never touch the database. The cheap topic + # split below runs only when the filter is active, so the default + # (accept-all) path is unaffected. + if self._observer_filter.active: + parsed_topic = self.mqtt.topic_builder.parse_letsmesh_upload_topic(topic) + if parsed_topic: + observer_key, _feed = parsed_topic + if not self._observer_filter.is_allowed(observer_key): + logger.debug( + "Dropping event from blocked observer %s...", + observer_key[:12], + ) + return + parsed: tuple[str, str, dict[str, Any]] | None parsed = self._normalize_letsmesh_event(topic, payload) @@ -714,6 +742,7 @@ def create_subscriber( channel_refresh_interval_seconds: int = 300, raw_packet_capture_enabled: bool = False, raw_packet_retention_days: int = 7, + observer_filter: Optional[ObserverFilter] = None, ) -> Subscriber: """Create a configured subscriber instance. @@ -769,6 +798,7 @@ def create_subscriber( channel_refresh_interval_seconds=channel_refresh_interval_seconds, raw_packet_capture_enabled=raw_packet_capture_enabled, raw_packet_retention_days=raw_packet_retention_days, + observer_filter=observer_filter, ) # Register handlers @@ -798,6 +828,7 @@ def run_collector( channel_refresh_interval_seconds: int = 300, raw_packet_capture_enabled: bool = False, raw_packet_retention_days: int = 7, + observer_filter: Optional[ObserverFilter] = None, ) -> None: """Run the collector (blocking). @@ -838,6 +869,7 @@ def run_collector( channel_refresh_interval_seconds=channel_refresh_interval_seconds, raw_packet_capture_enabled=raw_packet_capture_enabled, raw_packet_retention_days=raw_packet_retention_days, + observer_filter=observer_filter, ) # Set up signal handlers diff --git a/src/meshcore_hub/common/config.py b/src/meshcore_hub/common/config.py index 43c9884..0b3d55e 100644 --- a/src/meshcore_hub/common/config.py +++ b/src/meshcore_hub/common/config.py @@ -261,6 +261,26 @@ class CollectorSettings(CommonSettings): ge=1, ) + # Observer ingestion filtering (allow/deny by observer public key). + # Stored as raw comma-separated strings (not list[str]) because + # pydantic-settings parses complex field types from env vars as JSON, which + # breaks comma-separated input; the *_keys properties split them instead. + # Allowlist takes precedence over denylist. Empty = no restriction. + observer_allowlist: str = Field( + default="", + description=( + "Comma-separated observer public keys (or key prefixes) allowed to " + "ingest. If set, overrides OBSERVER_DENYLIST." + ), + ) + observer_denylist: str = Field( + default="", + description=( + "Comma-separated observer public keys (or key prefixes) blocked from " + "ingesting. Ignored when OBSERVER_ALLOWLIST is set." + ), + ) + # 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. @@ -313,6 +333,16 @@ class CollectorSettings(CommonSettings): return self.raw_packet_retention_days return self.data_retention_days + @property + def observer_allowlist_keys(self) -> list[str]: + """Parsed OBSERVER_ALLOWLIST entries (empty/blank entries dropped).""" + return [k.strip() for k in self.observer_allowlist.split(",") if k.strip()] + + @property + def observer_denylist_keys(self) -> list[str]: + """Parsed OBSERVER_DENYLIST entries (empty/blank entries dropped).""" + return [k.strip() for k in self.observer_denylist.split(",") if k.strip()] + @property def collector_data_dir(self) -> str: """Get the collector data directory path.""" diff --git a/tests/test_collector/test_observer_filter.py b/tests/test_collector/test_observer_filter.py new file mode 100644 index 0000000..2ce356e --- /dev/null +++ b/tests/test_collector/test_observer_filter.py @@ -0,0 +1,99 @@ +"""Unit tests for the observer allow/deny ingestion filter.""" + +from meshcore_hub.collector.observer_filter import ObserverFilter + +FULL_KEY = "F4762185BBB684510B2E3D41568300869DB5E75B284448145475F7708C1EF408" +OTHER_KEY = "A1B2C3D4E5F60718293A4B5C6D7E8F90112233445566778899AABBCCDDEEFF00" + + +class TestObserverFilterDefaults: + def test_no_lists_allows_everything(self): + f = ObserverFilter() + assert f.active is False + assert f.is_allowed(FULL_KEY) is True + assert f.is_allowed(OTHER_KEY) is True + + def test_from_lists_with_none_is_inactive(self): + f = ObserverFilter.from_lists(None, None) + assert f.active is False + assert f.is_allowed(FULL_KEY) is True + + def test_from_lists_with_empty_lists_is_inactive(self): + f = ObserverFilter.from_lists([], []) + assert f.active is False + assert f.is_allowed(FULL_KEY) is True + + +class TestAllowlist: + def test_allows_listed_key_blocks_others(self): + f = ObserverFilter.from_lists(allowlist=[FULL_KEY]) + assert f.active is True + assert f.is_allowed(FULL_KEY) is True + assert f.is_allowed(OTHER_KEY) is False + + def test_allowlist_takes_precedence_over_denylist(self): + # Same key is both allowed and denied -> allowlist wins (key allowed). + f = ObserverFilter.from_lists(allowlist=[FULL_KEY], denylist=[FULL_KEY]) + assert f.is_allowed(FULL_KEY) is True + + def test_allowlist_present_means_unlisted_denied_regardless_of_denylist(self): + f = ObserverFilter.from_lists(allowlist=[FULL_KEY], denylist=[OTHER_KEY]) + # OTHER_KEY is not on the allowlist, so it is blocked even though the + # denylist is effectively ignored. + assert f.is_allowed(OTHER_KEY) is False + + +class TestDenylist: + def test_blocks_listed_key_allows_others(self): + f = ObserverFilter.from_lists(denylist=[FULL_KEY]) + assert f.active is True + assert f.is_allowed(FULL_KEY) is False + assert f.is_allowed(OTHER_KEY) is True + + +class TestCaseInsensitivity: + def test_lowercase_entry_matches_uppercase_key(self): + f = ObserverFilter.from_lists(denylist=[FULL_KEY.lower()]) + assert f.is_allowed(FULL_KEY) is False + + def test_uppercase_entry_matches_lowercase_key(self): + f = ObserverFilter.from_lists(denylist=[FULL_KEY.upper()]) + assert f.is_allowed(FULL_KEY.lower()) is False + + +class TestPrefixMatching: + def test_short_prefix_matches_full_key(self): + f = ObserverFilter.from_lists(denylist=["F4762185"]) + assert f.is_allowed(FULL_KEY) is False + assert f.is_allowed(OTHER_KEY) is True + + def test_non_matching_prefix_does_not_match(self): + f = ObserverFilter.from_lists(denylist=["DEADBEEF"]) + assert f.is_allowed(FULL_KEY) is True + + def test_allowlist_prefix(self): + f = ObserverFilter.from_lists(allowlist=["f4762185"]) + assert f.is_allowed(FULL_KEY) is True + assert f.is_allowed(OTHER_KEY) is False + + +class TestHygiene: + def test_whitespace_and_empty_entries_dropped(self): + f = ObserverFilter.from_lists(denylist=[" ", "", FULL_KEY + " "]) + # The blank/whitespace entries must not turn into a match-everything + # prefix; only the real (trimmed) key is in effect. + assert f.is_allowed(OTHER_KEY) is True + assert f.is_allowed(FULL_KEY) is False + + def test_all_blank_entries_yield_inactive_filter(self): + f = ObserverFilter.from_lists(allowlist=["", " "], denylist=[" "]) + assert f.active is False + assert f.is_allowed(FULL_KEY) is True + + def test_none_public_key_against_active_filter(self): + # A missing key never matches a real allowlist entry -> denied. + allow = ObserverFilter.from_lists(allowlist=[FULL_KEY]) + assert allow.is_allowed(None) is False + # ... and is not on a denylist -> allowed. + deny = ObserverFilter.from_lists(denylist=[FULL_KEY]) + assert deny.is_allowed(None) is True diff --git a/tests/test_collector/test_subscriber.py b/tests/test_collector/test_subscriber.py index 0516374..b56da06 100644 --- a/tests/test_collector/test_subscriber.py +++ b/tests/test_collector/test_subscriber.py @@ -3,6 +3,7 @@ import pytest from unittest.mock import MagicMock, call, patch +from meshcore_hub.collector.observer_filter import ObserverFilter from meshcore_hub.collector.subscriber import Subscriber, create_subscriber @@ -200,6 +201,139 @@ class TestSubscriber: mock_mqtt_client.subscribe.assert_has_calls(expected_calls, any_order=False) assert mock_mqtt_client.subscribe.call_count == 3 + def test_blocked_observer_is_dropped_no_dispatch( + self, mock_mqtt_client, db_manager + ): + """An observer on the denylist has its event dropped before dispatch.""" + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.return_value = ( + "b" * 64, + "packets", + ) + subscriber = Subscriber( + mock_mqtt_client, + db_manager, + observer_filter=ObserverFilter.from_lists(denylist=["b" * 64]), + ) + handler = MagicMock() + subscriber.register_handler("channel_msg_recv", handler) + # Spy on normalize: it must never be called for a blocked observer. + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("b" * 64, "channel_msg_recv", {"text": "hi"}) + ) + + subscriber._handle_mqtt_message( + topic=f"meshcore/STN/{'b' * 64}/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "00", "hash": "h1"}, + ) + + handler.assert_not_called() + subscriber._normalize_letsmesh_event.assert_not_called() + + def test_blocked_observer_writes_no_raw_packet(self, mock_mqtt_client, db_manager): + """A blocked observer's packet is dropped before raw-packet capture.""" + from sqlalchemy import select + + from meshcore_hub.common.models import RawPacket + + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.return_value = ( + "b" * 64, + "packets", + ) + subscriber = Subscriber( + mock_mqtt_client, + db_manager, + raw_packet_capture_enabled=True, + observer_filter=ObserverFilter.from_lists(denylist=["b" * 64]), + ) + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("b" * 64, "channel_msg_recv", {"text": "hi"}) + ) + + subscriber._handle_mqtt_message( + topic=f"meshcore/STN/{'b' * 64}/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "0011", "hash": "h1"}, + ) + + session = db_manager.get_session() + try: + rows = session.execute(select(RawPacket)).scalars().all() + assert len(rows) == 0 + finally: + session.close() + + def test_allowed_observer_dispatches_normally(self, mock_mqtt_client, db_manager): + """An observer on the allowlist is dispatched as usual.""" + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.return_value = ( + "a" * 64, + "packets", + ) + subscriber = Subscriber( + mock_mqtt_client, + db_manager, + observer_filter=ObserverFilter.from_lists(allowlist=["a" * 64]), + ) + handler = MagicMock() + subscriber.register_handler("channel_msg_recv", handler) + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("a" * 64, "channel_msg_recv", {"text": "hi"}) + ) + + subscriber._handle_mqtt_message( + topic=f"meshcore/STN/{'a' * 64}/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "00", "hash": "h1"}, + ) + + handler.assert_called_once() + + def test_observer_not_on_allowlist_is_dropped(self, mock_mqtt_client, db_manager): + """With an allowlist active, an unlisted observer is dropped.""" + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.return_value = ( + "b" * 64, + "packets", + ) + subscriber = Subscriber( + mock_mqtt_client, + db_manager, + observer_filter=ObserverFilter.from_lists(allowlist=["a" * 64]), + ) + handler = MagicMock() + subscriber.register_handler("channel_msg_recv", handler) + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("b" * 64, "channel_msg_recv", {"text": "hi"}) + ) + + subscriber._handle_mqtt_message( + topic=f"meshcore/STN/{'b' * 64}/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "00", "hash": "h1"}, + ) + + handler.assert_not_called() + + def test_inactive_filter_does_not_parse_topic(self, mock_mqtt_client, db_manager): + """The default (inactive) filter adds no work: the topic is not parsed + for filtering, and dispatch proceeds normally.""" + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.reset_mock() + subscriber = Subscriber(mock_mqtt_client, db_manager) + handler = MagicMock() + subscriber.register_handler("channel_msg_recv", handler) + subscriber._normalize_letsmesh_event = MagicMock( # type: ignore[method-assign] + return_value=("a" * 64, "channel_msg_recv", {"text": "hi"}) + ) + + subscriber._handle_mqtt_message( + topic=f"meshcore/STN/{'a' * 64}/packets", + pattern="meshcore/+/+/packets", + payload={"raw": "00", "hash": "h1"}, + ) + + handler.assert_called_once() + # Inactive filter short-circuits before touching the topic builder. + mock_mqtt_client.topic_builder.parse_letsmesh_upload_topic.assert_not_called() + def test_letsmesh_status_maps_to_letsmesh_status( self, mock_mqtt_client, db_manager ) -> None: diff --git a/tests/test_common/test_config.py b/tests/test_common/test_config.py index e21edd5..d9877d0 100644 --- a/tests/test_common/test_config.py +++ b/tests/test_common/test_config.py @@ -74,6 +74,27 @@ class TestCollectorSettings: assert settings.raw_packet_capture_enabled is False + def test_observer_lists_default_empty(self) -> None: + """Observer allow/deny lists default to empty (accept-all).""" + settings = CollectorSettings(_env_file=None) + + assert settings.observer_allowlist_keys == [] + assert settings.observer_denylist_keys == [] + + def test_observer_allowlist_parsed_and_trimmed(self) -> None: + """Comma-separated allowlist is split and blank entries dropped.""" + settings = CollectorSettings( + _env_file=None, observer_allowlist="aaa, bbb ,, ,ccc" + ) + + assert settings.observer_allowlist_keys == ["aaa", "bbb", "ccc"] + + def test_observer_denylist_parsed_and_trimmed(self) -> None: + """Comma-separated denylist is split and blank entries dropped.""" + settings = CollectorSettings(_env_file=None, observer_denylist=" key1 ,key2") + + assert settings.observer_denylist_keys == ["key1", "key2"] + def test_explicit_seed_home_overrides(self) -> None: """Test that explicit seed_home overrides the default.""" settings = CollectorSettings(_env_file=None, seed_home="/seed/data")