diff --git a/.env.example b/.env.example index 7c22f227..aa75b299 100644 --- a/.env.example +++ b/.env.example @@ -68,6 +68,33 @@ FEDERATION=1 PRIVATE=0 +# ============================================================================= +# TRANSMIT SETTINGS (your ingestor putting traffic on the air) +# ============================================================================= +# +# By default a PotatoMesh ingestor is a LISTENER: it receives, and it transmits +# nothing. Everything below is off unless you deliberately turn it on, because +# airtime is a shared resource and your community's norms about automated +# traffic are yours to set, not ours. + +# Master transmit switch (0=receive only, 1=allow transmitting) +# Leave at 0 and the ingestor never puts anything on the mesh. +# Turning this on enables MeshCore on-air telemetry polling of other nodes, +# which is how their battery/voltage readings reach your dashboard. +# Does NOT by itself enable announcements — see TX_ANNOUNCE. +# Local reads over USB/BLE to your own radio are not transmissions and always +# work, whatever this is set to. +TX_ENABLED=0 + +# Periodic activity announcement (0=off, 1=on). Requires TX_ENABLED=1. +# Broadcasts ONE line on your default channel, at most once every 24 hours, and +# never in the first 24 hours after start: +# "Meshtastic activity in the last 24h: 42 active nodes, 118 packets/hour. https://your.domain" +# This is unsolicited automated traffic on a channel people read. Check that +# your community wants it before enabling. +TX_ANNOUNCE=0 + + # ============================================================================= # ADVANCED SETTINGS # ============================================================================= diff --git a/ACCEPTANCE.md b/ACCEPTANCE.md index 58fbcc5d..3faa9017 100644 --- a/ACCEPTANCE.md +++ b/ACCEPTANCE.md @@ -3172,9 +3172,10 @@ with `protocol="meshcore"`. The poll loop honours only — no broker, Invariant I). Each contact is additionally capped by a fixed 24 h per-node cooldown (stamped at the poll attempt; an all-fresh roster tick transmits nothing, and departed contacts are pruned from the -stamp table). `RX_ONLY=1` forbids every ingestor-initiated transmission: -contact polls stop entirely while the airtime-free companion-link self reads -continue. +stamp table). The transmit policy forbids every ingestor-initiated +transmission unless `TX_ENABLED=1` (**default `0`**), with the legacy +`RX_ONLY=1` vetoing it regardless: contact polls stop entirely while the +airtime-free companion-link self reads continue. ### TI-R1 — Regression: prior acceptance still holds ```bash @@ -4002,9 +4003,10 @@ merged `packets` figure, appends the per-interval delta to its hourly `MAX`-per-protocol 24 h packets/hour moving average (MA4–MA5). Each ingestor then periodically broadcasts a one-line activity summary — numbers **dogfed from the target instance's own API** — on its protocol's default channel (MA6), gated by -the `ANNOUNCE` opt-in (default `0`) and `RX_ONLY` (the global TX kill switch, -which overrides it), the target's `/version` privacy flag -(fail-closed), and a ≥ 24 h post-start delay (MA7–MA8), via an **optional** +`TX_ENABLED` (the master transmit switch, default `0`), `TX_ANNOUNCE` (the +narrower announcement opt-in, default `0`), the target's `/version` privacy flag +(fail-closed), and a ≥ 24 h post-start delay (MA7–MA8), with the legacy +`RX_ONLY` retained as a veto, via an **optional** duck-typed provider send that leaves `MeshProtocol` conformance intact (MA9). Unless a check says otherwise, start the server in **public** mode @@ -4088,19 +4090,23 @@ where `` = the target's `GET /api/stats` `.nodes.day` and `` = line is truncated to the protocol's character limit. `` = the configured `INSTANCE_DOMAIN`. -### MA-A7 — Announcement gates: ANNOUNCE opt-in, RX_ONLY, privacy fail-closed, 24 h — MA7 +### MA-A7 — Announcement gates: TX_ENABLED, TX_ANNOUNCE, privacy fail-closed, 24 h — MA7 ```bash +( . .venv/bin/activate && pytest -q tests/test_tx_policy_unit.py ) ( . .venv/bin/activate && pytest -q tests/test_announce_unit.py \ - -k "gate or private or rx_only or elapsed" ) + -k "gate or private or suppressed or elapsed or monotonic" ) ``` -**Expected:** pass. **No** announcement is sent when any gate fails: `ANNOUNCE` is -unset (the default `0` — the announcement is opt-in, so an ingestor deployed to -feed a map never transmits on its own); `RX_ONLY=1` (the receive-only flag still -forbids every ingestor TX and overrides `ANNOUNCE=1`); the target `/version` -reports `private_mode: true`; the `/version` fetch **errors or is unparseable** -(fail-closed — treated as private/skip); or `< 24 h` have elapsed since ingestor -start. With `ANNOUNCE=1`, `RX_ONLY` unset, `private_mode: false`, and `≥ 24 h` -elapsed, exactly one announcement per 24 h per domain is transmitted. +**Expected:** both pass. **No** announcement is sent when any gate fails: +`TX_ENABLED` is unset (**the default `0`** — an ingestor deployed to feed a map +transmits nothing at all, announcement or telemetry poll); `TX_ANNOUNCE` is unset +(the default `0` — permission to transmit is not permission to broadcast +unsolicited on a shared human channel); the legacy `RX_ONLY=1` is set (it vetoes +`TX_ENABLED=1` and the contradiction is warned about at startup); the target +`/version` reports `private_mode: true`; the `/version` fetch **errors or is +unparseable** (fail-closed — treated as private/skip); or `< 24 h` have elapsed +since ingestor start. With `TX_ENABLED=1`, `TX_ANNOUNCE=1`, `RX_ONLY` unset, +`private_mode: false`, and `≥ 24 h` elapsed, exactly one announcement per 24 h per +domain is transmitted. Each closed gate names itself in the log (`blocked_by`). ### MA-A8 — Default channel/scope + 24 h cadence — MA8 ```bash @@ -5006,3 +5012,117 @@ on the coordinate. Fixes clicks near the crown falling outside the hit area legend / map-layer suites, and **B1–B5** stay green; the changes add one CSS rule, resize the legend swatch, and grow the pin's icon box — nothing touches the API, storage, SSE, or privacy paths. + +--- + +## Bugfix: TX kill switch failed open; transmit policy is now default-off (`TX_*`) + +Maps to SPEC decision **MA7** (amended ×2). Every ingestor-initiated mesh +transmission is now off unless asked for, expressed once in +`data/mesh_ingestor/tx_policy.py` and enforced at each of the four transmit +sites beside `activity.record_tx()`. Two operator flags — `TX_ENABLED` (master) +and `TX_ANNOUNCE` (announcement opt-in), both default `0` — replace `ANNOUNCE`, +and the legacy `RX_ONLY` is retained as an undocumented veto. + +### TX-A1 — The kill switch cannot fail open — MA7 a +```bash +( . .venv/bin/activate && pytest -q tests/test_tx_policy_unit.py \ + -k "kill_switch or fails_safe or spellings" ) +``` +**Expected:** pass. `RX_ONLY` was parsed as an exact `os.environ.get(...) == "1"`, +so `RX_ONLY=true`, `TRUE`, `yes`, `on`, `" 1"` and `"1 "` all resolved to +**False** and the receive-only ingestor **transmitted anyway**. Every one of those +spellings now engages the switch, and an *unrecognized* value resolves to +engaged — a kill switch fails toward killed. `TX_ENABLED` / `TX_ANNOUNCE` fail the +opposite way for the same reason: toward silence. All three **warn** on an +unparseable value rather than raising — deliberately unlike `TRANSPORT`, +`PROTOCOL` and `MESH_UDP_PORT`, which all reject a bad value at import. Those +are selectors with no safe fallback; a transmit flag has one, so a typo costs a +feature rather than the ingestor's ability to keep *receiving*. + +### TX-A2 — Transmission is off by default, at every altitude — MA7 a/b +```bash +( . .venv/bin/activate && pytest -q tests/test_tx_policy_unit.py \ + -k "defaults_off or TransmitPermitted or AnnouncementsPermitted" ) +( . .venv/bin/activate && pytest -q tests/test_provider_unit.py \ + -k "refuses_closed_policy or status_fallback_rechecks" ) +( . .venv/bin/activate && pytest -q tests/test_announce_unit.py \ + -k "refuses_when_transmit_policy_closed or suppressed" ) +``` +**Expected:** all pass. With no TX env set, `transmit_permitted()` is `False`: the +MeshCore on-air contact telemetry/status polls do not run **and** no announcement +is sent. The gate is enforced at **three** altitudes, not one — the daemon entry +point (`maybe_run_announcements`), the exported mid-level send +(`send_announcement_to_instance`), and the transmit primitives themselves (both +providers' `send_channel_announcement`, and each of the two on-air requests in +`_poll_contact_telemetry`, beside that site's `record_tx`). The mid-level check +is load-bearing rather than redundant: `send_channel_announcement` is **not** a +`MeshProtocol` member (`mesh_protocol.py` defines five, and this is not one of +them), so nothing structurally forces a provider to carry the gate — the +`CLAUDE.md` protocol checklist now asks for it, but a checklist is documentation, +not enforcement. Verified by execution: without this check, +`run_announcement_cycle` — which is in `announce.__all__` — transmits through a +gateless provider with every flag off. Companion-link self reads +(USB/BLE to the operator's own radio) are not transmissions and continue. + +### TX-A3 — `TX_ENABLED=1` + `RX_ONLY=1` resolves to silence, loudly — MA7 a +```bash +( . .venv/bin/activate && pytest -q tests/test_tx_policy_unit.py -k "contradict or matrix" ) +``` +**Expected:** pass. The legacy veto wins over the new master switch — a kill +switch an operator deliberately engaged is never silently overridden by a flag +that arrives later in the same `.env` — and the contradictory combination emits a +`warning` at startup naming `RX_ONLY`, because resolving to *silence* is the +surprising direction for whoever just set `TX_ENABLED=1`. + +### TX-A4 — The flags reach every packaged deployment — MA10 +```bash +( . .venv/bin/activate && pytest -q tests/test_tx_policy_unit.py -k "DeploymentSurface" ) +( TX_ENABLED=1 docker compose config | grep -E '^\s+TX_(ENABLED|ANNOUNCE):' ) +``` +**Expected:** both pass. The opt-in previously reached **no** packaged +deployment: `x-ingestor-base.environment:` is a closed allowlist with no +`env_file:`, so `ANNOUNCE=1` in `.env` never entered the container, and +`flake.nix` and `data/Dockerfile` were equally closed. `docker-compose.yml` now +passes both flags through (the `.dev`/`.prod` overlays inherit the mapping), the +image declares both defaults in each stage, the Nix module exposes `txEnabled` / +`txAnnounce`, and `.env.example` documents them under a `TRANSMIT SETTINGS` +heading. A knob nobody can set is a knob that does not exist. + +### TX-A5 — The 24 h anti-spam delay survives an NTP step — MA7 c +```bash +( . .venv/bin/activate && pytest -q tests/test_announce_unit.py -k "monotonic" ) +( . .venv/bin/activate && pytest -q tests/test_daemon_unit.py -k "announcements_delegates" ) +``` +**Expected:** pass. The delay was measured against wall clock +(`ingestors.STATE.start_time`, `time.time()`), so on an RTC-less host — a Pi or +SBC, exactly the hardware these run on — a post-boot NTP step of hours-to-years +cleared it instantly and the first announcement fired on the next 60 s loop. +Scheduling now reads `ingestors.ingestor_start_monotonic()` and `time.monotonic()`, +matching every other daemon timer. `start_time` stays wall clock because it goes +out on the heartbeat wire (D8 unchanged). + +### TX-A6 — Operator-facing docs describe `TX_*` and never `RX_ONLY` — MA7 +```bash +( grep -c 'TX_ENABLED' README.md ) && ( ! grep -q 'RX_ONLY' README.md ) +( . .venv/bin/activate && pytest -q tests/test_tx_policy_unit.py -k "not_reintroduced" ) +``` +**Expected:** pass. `README.md` carries a **Transmitting on the mesh** section — +what each flag turns on, the two-flag truth table, the literal announcement text, +what is never gated, and the startup log line to check — rather than the feature +being visible only in release notes. `RX_ONLY` appears nowhere in it: it remains +supported in code for existing deployments but is retired from documentation. + +### TX-R1 — Regression: prior acceptance still holds +```bash +( . .venv/bin/activate && pytest -q tests/ ) && ( cd web && bundle exec rspec ) && ( cd web && npm test ) +``` +**Expected:** all green. At risk and explicitly required to remain green: +**MA-A1–MA-A9** (the activity feature — counting, heartbeat delta, aggregation and +`/api/stats` are untouched; only the gates moved), **TI-A3** (MeshCore telemetry — +the poll machinery is unchanged, its permission now defaults off, and the +acceptance text is amended to say so), **A4b** (provider conformance — the gate is +inside the optional duck-typed send, so `MeshProtocol` is untouched), and **C2** +(canonical POST shapes — `tests/` fixtures unmodified). The web app is not +touched at all: no Ruby, JS, API, storage, SSE, or privacy path changes. + diff --git a/CHANGELOG.md b/CHANGELOG.md index ce42284a..faf8f7e5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,7 +26,12 @@ table, and reported-only disclosure rows. * Web: nodes table gains grouped headers, curated mobile columns (Battery survives), a per-row disclosure of hidden fields, row hover/click, numeric alignment, captions/scopes ### Fixes -* Ingestor: the periodic activity announcement is now **opt-in** via a new `ANNOUNCE` env var (default `0`) instead of on by default. It previously shipped with `RX_ONLY` as its only gate, so a stock ingestor deployed to feed a map began broadcasting on its community's default channel unprompted, and the sole opt-out also silenced the MeshCore telemetry polls. `RX_ONLY=1` still forbids every ingestor TX and overrides `ANNOUNCE=1`; the README env table now documents both (SPEC MA7, amended) +* **Ingestor: every mesh transmission is now off by default** (`TX_ENABLED`, default `0`). This covers the activity announcement *and* the MeshCore on-air contact telemetry polls — an ingestor deployed to feed a dashboard now transmits nothing until its operator opts in. Announcements need a second, narrower opt-in on top (`TX_ANNOUNCE=1`, default `0`), so permission to transmit is not permission to broadcast unsolicited on a shared human channel. **Breaking for MeshCore operators who want other nodes' telemetry: set `TX_ENABLED=1`.** Companion-link reads over USB/BLE to your own radio are not transmissions and are unaffected (SPEC MA7, amended) +* Ingestor: **`RX_ONLY` no longer fails open.** It was parsed as an exact `== "1"` string comparison, so `RX_ONLY=true`, `TRUE`, `yes`, `on` and `" 1"` all silently resolved to *off* — a receive-only listening post transmitted anyway. All TX flags now accept the common spellings, strip whitespace, and resolve an unrecognized value toward *silence*, warning rather than raising. `RX_ONLY` itself is superseded by `TX_ENABLED` and retired from the docs; it stays supported, and keeps its veto over `TX_ENABLED=1` +* Ingestor: the `TX_*` flags now reach every packaged deployment — `docker-compose.yml`, `data/Dockerfile`, the `flake.nix` NixOS module (`txEnabled`/`txAnnounce`), and `.env.example`. The previous `ANNOUNCE` opt-in reached none of them: the Compose `environment:` mapping is a closed allowlist with no `env_file:`, so setting it in `.env` had no effect at all +* Ingestor: the announcement's 24 h anti-spam delay is measured on a monotonic clock. It used wall clock, so on an RTC-less host (Pi/SBC) a post-boot NTP step cleared the delay instantly and the first announcement fired within a minute of start +* Ingestor: transmit policy is stated in the log at startup without `DEBUG=1`, and each suppressed announcement names the gate that closed (`blocked_by`) +* Docs: README gains a **Transmitting on the mesh** section — what each flag turns on, the truth table, the literal announcement text, and what is never gated. Previously this behavior was visible only in release notes * Web: mesh-activity card — idle card no longer paints an empty pill over the map on phones (≤659px), the mobile strip spans the map as a caption, the sparkline gains headroom and rebases with the protocol toggles, the card is a labelled `role="group"`, and the /charts intro heading drops the single-protocol badge (design-review remediation, SPEC MR1–MR8) * Web: neighbor/trace line legend toggles highlight when their lines are visible (was reversed — highlighted when hidden), matching the role chips (SPEC NT1) * Web/data: `GET /api/stats` exposes the mesh-activity packets/hour rate as `.packets.hour` (folded into the S1 scope→metric→window tree) rather than a top-level `packets_per_hour` map, and `total.packets.hour` now **sums** the per-protocol rates instead of taking a single MAX vantage (distinct protocols never share the air, so they add); the ingestor announcement dogfeed reads the new path (SPEC MA4/MA5, pre-release amendment of the merged-but-unreleased #859) diff --git a/CLAUDE.md b/CLAUDE.md index e0c2a4a5..dc8f4260 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -66,6 +66,7 @@ The `data/mesh_ingestor/mesh_protocol.py` module defines a `@runtime_checkable` 2. Register it in `data/mesh_ingestor/protocols/__init__.py`. 3. Pass an instance via `daemon.main(provider=...)` or make it the default in `main()`. 4. Cover the protocol with unit tests in `tests/test_provider_unit.py` — at minimum an `isinstance(..., MeshProtocol)` conformance check and any retry/error-handling paths. +5. **Gate every transmission.** If the protocol transmits on the mesh at all, each transmit site must call `tx_policy.transmit_permitted()` (or `announcements_permitted()` for the optional `send_channel_announcement`) immediately before the send, beside its `activity.record_tx()`. Transmission is off by default (SPEC MA7); a provider that omits the check would transmit against the operator's configuration. Companion-link reads to the host's own radio are not transmissions and are not gated. Consult `data/mesh_ingestor/CONTRACTS.md` for the canonical event shapes all protocols must emit. diff --git a/README.md b/README.md index 3d0f8d1a..1b17657f 100644 --- a/README.md +++ b/README.md @@ -114,10 +114,10 @@ The web app can be configured with environment variables (defaults shown): | `MESH_UDP_GROUP` | `224.0.0.69` | Multicast group joined in UDP transport. | | `MESH_UDP_PORT` | `4403` | Multicast port joined in UDP transport. | | `INGESTOR_NODE_ID` | _unset_ | `!xxxxxxxx` id used for the ingestor heartbeat in UDP transport (which cannot auto-detect "self"). | -| `MESHCORE_TELEMETRY_POLL_SECONDS` | `300` | Seconds between MeshCore contact telemetry polls (one on-air request per interval, round-robin over the roster; each contact is additionally polled at most once per 24 h — when every contact is fresh the tick sends nothing). Set `0` to disable on-air polling. | +| `MESHCORE_TELEMETRY_POLL_SECONDS` | `300` | **Requires `TX_ENABLED=1`** (polling other nodes is a transmission). Seconds between MeshCore contact telemetry polls (one on-air request per interval, round-robin over the roster; each contact is additionally polled at most once per 24 h — when every contact is fresh the tick sends nothing). Set `0` to disable on-air polling. | | `MESHCORE_SELF_TELEMETRY_SECONDS` | `3600` | Seconds between MeshCore host self-telemetry reads (battery/sensors over the companion link, no airtime). Set `0` to disable. | -| `ANNOUNCE` | `0` | Set to `1` to let the ingestor broadcast a one-line activity summary (`" activity in the last 24h: active nodes, packets/hour. "`) on its default channel — at most once per 24 h, and never in the first 24 h after start. **Off by default**: it is the ingestor's only unsolicited transmission on a shared human channel, and communities differ on whether automated traffic is welcome there. `RX_ONLY=1` overrides it. | -| `RX_ONLY` | `0` | Set to `1` to forbid every ingestor-initiated mesh transmission (receive-only listening post). Disables the MeshCore contact telemetry/status polls and the `ANNOUNCE` activity broadcast; local companion-link reads (self telemetry, roster, channels) continue. | +| `TX_ENABLED` | `0` | Master switch for **everything the ingestor puts on the air**. Off by default — the ingestor is a listener. Set to `1` to allow transmitting; this is what enables MeshCore on-air contact telemetry polling. Does not by itself enable announcements. See [Transmitting on the mesh](#transmitting-on-the-mesh). | +| `TX_ANNOUNCE` | `0` | Set to `1` (**with** `TX_ENABLED=1`) to broadcast a one-line activity summary on the default channel — at most once per 24 h, never in the first 24 h after start. Off by default: it is unsolicited automated traffic on a channel people read. See [Transmitting on the mesh](#transmitting-on-the-mesh). | | `FEDERATION` | `1` | Set to `1` to announce your instance and crawl peers, or `0` to disable federation. Private mode overrides this. | | `PRIVATE` | `0` | Set to `1` to hide the chat UI, disable message APIs, and exclude hidden clients from public listings. | | `EVENTS` | `1` | Set to `0` to disable the live-update SSE stream (`GET /api/events`); clients then fall back to polling at the refresh interval. | @@ -304,6 +304,55 @@ example `ALLOWED_CHANNELS="Chat,Ops"`); packets on other channels are discarded. Use `HIDDEN_CHANNELS` to block specific channels from the web UI even when they appear in the allowlist. +### Transmitting on the mesh + +**By default a PotatoMesh ingestor never transmits.** It listens, and forwards +what it hears to your dashboard. Nothing below happens unless you turn it on. + +That default is deliberate. Airtime on a LoRa mesh is a shared, scarce resource, +and whether automated traffic belongs on a community's channels is that +community's decision — not a default we get to make for you. + +| Variable | Default | What turning it on means | +| --- | --- | --- | +| `TX_ENABLED` | `0` | The ingestor may transmit. On its own this enables **MeshCore on-air telemetry polling**: round-robin requests to other nodes for their battery/sensor readings, at most one request per `MESHCORE_TELEMETRY_POLL_SECONDS` (default 300 s) and at most once per 24 h per node. This is how other nodes' telemetry reaches your dashboard. Meshtastic ingestion needs no transmission at all. | +| `TX_ANNOUNCE` | `0` | **Requires `TX_ENABLED=1`.** Broadcasts one line on your default channel, at most once every 24 h, and never in the first 24 h after the ingestor starts. | + +The announcement looks like this: + +``` +Meshtastic activity in the last 24h: 42 active nodes, 118 packets/hour. https://mesh.example.org +``` + +The numbers are read back from your own instance's public API, so they reflect +the whole mesh your dashboard sees rather than the one radio doing the talking. +It is suppressed automatically when your instance runs with `PRIVATE=1`, and +that check fails closed — if the ingestor cannot reach your instance to ask, it +does not transmit. + +Both flags must be on for an announcement to go out: + +| `TX_ENABLED` | `TX_ANNOUNCE` | Result | +| --- | --- | --- | +| `0` | `0` | Receive only. The default. | +| `0` | `1` | Receive only — `TX_ENABLED=0` wins. | +| `1` | `0` | Telemetry polling on, no announcements. | +| `1` | `1` | Telemetry polling on, one announcement per 24 h. | + +**What is never gated.** Reads over the companion link — USB or BLE to your own +radio, for its own telemetry, contact roster, and channel list — are not mesh +transmissions. They cost no airtime and always work. + +**Checking what your ingestor will do.** It states the resolved policy in its +log at startup, without needing `DEBUG=1`: + +``` +[2026-08-24T09:12:44.117Z] [potato-mesh] [info] context=tx.policy announcements_permitted=False rx_only=False transmit_permitted=False tx_announce=False tx_enabled=False Transmit policy resolved +``` + +If you set a flag and nothing happens, that line tells you what the ingestor +actually resolved — including a flag that never reached the container. + ### MeshCore Set `PROTOCOL=meshcore` to ingest from a MeshCore companion-firmware node diff --git a/SPEC.md b/SPEC.md index 7472951f..49dc7176 100644 --- a/SPEC.md +++ b/SPEC.md @@ -1063,7 +1063,8 @@ on its protocol's default channel, drawing the numbers **back from the target instance's own API** (dogfeeding, so one radio's partial view never under-represents the mesh). Integrates with `data/mesh_ingestor/handlers/_state.py` (RX/TX counting seam), `ingestors.py` -(heartbeat payload), `config.py` (reuse `RX_ONLY`), `daemon.py` +(heartbeat payload), `config.py` + `tx_policy.py` (the `TX_ENABLED` / +`TX_ANNOUNCE` transmit policy), `daemon.py` (announce schedule), `mesh_protocol.py` + `protocols/{meshtastic,meshcore,meshtastic_udp}` (optional send), a new announce + instance-stats-GET module, `data/ingestors.sql` plus a new `data/ingestor_activity.sql` table, @@ -1082,8 +1083,11 @@ on any fetch error; only public aggregates (active nodes, packet rate) are ever announced. *Invariant III (federation) & §5 (no analytics / phone-home)* — **consistent, new authorized behavior**: the announcement publishes an *unsigned public activity line on the community's local mesh* — not the signed federation -wire, not cloud egress, not third-party analytics — and is opt-out via -`RX_ONLY=1`. *Invariant IV (parity & pluggability)* — **extends**: sending is an +wire, not cloud egress, not third-party analytics — and is **opt-in**, requiring +both `TX_ENABLED=1` and `TX_ANNOUNCE=1` (MA7). Unsolicited transmission on a +shared human channel is authorized here only as an operator-chosen behavior, +never as a default: consent to *deploy an ingestor* is not consent to *speak on +the community's channel* (Invariant II). *Invariant IV (parity & pluggability)* — **extends**: sending is an *optional, duck-typed* provider capability, not a new formal `MeshProtocol` member, so the `A4b` conformance contract and every existing provider are untouched; both protocols announce identically. *D8 (stable contract)* — **extends**: the heartbeat @@ -1100,10 +1104,10 @@ contradicted. | **MA4** | **Aggregation = MAX per protocol; `total` = SUM across protocols (dedup-free "in the air").** The mesh-wide packets/hour for a protocol is `MAX` over that protocol's ingestors of *(that ingestor's total `packets` reported in the last 24 h ÷ 24)*. A single radio can only hear ≤ what is actually transmitted, so the busiest single vantage is the best dedup-free estimate of unique air traffic and can never double-count a frame heard by two radios (true per-frame dedup is impossible anyway — ignored/errored frames carry no id). The **`total`** rate is the **SUM** of the per-protocol rates — different protocols ride different frequencies/channels, so their frames never overlap in the air and add rather than dedup. **Amended pre-release:** `total` first shipped as a MAX over *every* ingestor regardless of protocol, which under-counted it to the single busiest protocol (e.g. meshcore 44 + meshtastic 76 rendered as 76, not 120); corrected to the cross-protocol SUM. The fixed `÷ 24` denominator (not ÷ elapsed) keeps the rate stable and avoids divide-by-small spikes; an ingestor with < 24 h of data simply reads lower and ramps up (moot in practice — the announcement fires only ≥ 24 h after start, MA7). *Accepted limitation:* the per-protocol MAX under-estimates the true union when different radios are busiest in different hours. | interview | | **MA5** | **Exposure as an additive per-scope `packets` metric.** `GET /api/stats` exposes the MA4 24 h MAX-aggregated moving average under each scope as `.packets.hour` — a `packets` metric carrying a single `hour` window (it is a rate, not a windowed count, so no day/week/month keys), consistent with the S1 `scope → metric → window` layout. `reticulum.packets.hour` is a forward-looking `0` stub (S6); `total.packets.hour` is the **SUM** of the per-protocol rates (MA4). The rest of the S1 tree is untouched, so this needs **no** version bump. The announcement's active-node figure reuses the existing `.nodes.day` count (already deduped by `node_id`); only packets/hour is new. **Amended pre-release:** the field originally shipped (unreleased, on `main`) as a top-level `packets_per_hour: { total, meshcore, meshtastic, reticulum }` map; it is superseded here by the per-scope `packets.hour` form for consistency with the S1 tree. Because it was merged but **not yet in any tagged release** and is **not** on the signed federation wire, the reshape carries **no** version bump, federation-compat fallback, or signature break — only the in-repo `announce.py` dogfeed reader (MA6) and the specs/docs move with it. | proposed (amended) | | **MA6** | **Announcement content, drawn from the instance's own API (dogfeeding).** Each ingestor broadcasts, for its own configured protocol, a single line formatted to that protocol's character limit: `" activity in the last 24h: active nodes, packets/hour. https://"`. `` and `` are fetched **from the target instance** (`GET /api/stats` → `.nodes.day` and `.packets.hour`), never computed from the ingestor's local view — because one ingestor may not see the whole mesh. `` is the configured `INSTANCE_DOMAIN`. Reporting (MA1–MA4) is independent and continues regardless of announcement state. | interview | -| **MA7** | **Announcement is quadruple-gated (opt-in).** An announcement is transmitted only when **all** hold: (a) `ANNOUNCE=1` — a dedicated boolean, **default `0`**, and `RX_ONLY` is unset; `RX_ONLY=1` still forbids *every* ingestor TX (the MeshCore polls and the announcement alike) and so overrides `ANNOUNCE`. **Amended:** MA7 originally made `RX_ONLY` the *single* transmit gate with **no separate enable env**, which left the announcement **on by default**. In the field that made every stock ingestor start talking on its community's default channel without its operator having chosen to, and the only opt-out — `RX_ONLY` — also silences the MeshCore telemetry polls, so an operator who wants ingestion but not chatter had no setting for it. Reversed to opt-in: unsolicited TX on a shared human channel is a decision each operator's local community norms must make, not a default (Invariant II, consent). `RX_ONLY` keeps its meaning as the global TX kill switch; (b) the target instance reports **non-private** — the ingestor GETs `/version` and honors `config.private_mode`, re-checked every cycle, and **fails closed** (skips) on any fetch/parse error, so a privacy signal is never missed (Invariant II); (c) **≥ 24 h have elapsed since ingestor start** — so the first numbers are accurate over a full window and restarts cannot spam the channel. | interview | +| **MA7** | **Announcement is quadruple-gated, behind a default-off transmit policy.** An announcement is transmitted only when **all** hold: (a) **`TX_ENABLED=1`** — the master switch for *every* ingestor-initiated mesh transmission (the announcement and the MeshCore on-air telemetry polls alike), **default `0`**; (b) **`TX_ANNOUNCE=1`** — a second, narrower opt-in for the announcement specifically, **default `0`**, because unsolicited traffic on a shared human channel is a distinct decision from polling a node for its battery level; (c) **≥ 24 h have elapsed since ingestor start** — so the first numbers are accurate over a full window and restarts cannot spam the channel; (d) the target instance reports **non-private** — the ingestor GETs `/version` and honors `config.private_mode`, re-checked every cycle, and **fails closed** (skips) on any fetch/parse error, so a privacy signal is never missed (Invariant II). The legacy **`RX_ONLY=1`** remains supported as a **veto** over (a) — a kill switch an operator deliberately engaged is never silently overridden — but is retired from operator-facing documentation, since leaving `TX_ENABLED` unset now expresses the same intent by default. **Amended twice.** MA7 originally made `RX_ONLY` the *single* transmit gate with no separate enable env, leaving the announcement **on by default**; in the field that made every stock ingestor start talking on its community's default channel without its operator having chosen to. It was then reversed to an `ANNOUNCE=1` opt-in — but that left the wider hole unaddressed: MeshCore telemetry polling still transmitted by default, and `RX_ONLY`, the only switch that stopped it, was parsed as an exact `== "1"` string comparison, so `RX_ONLY=true` **failed open** and transmitted anyway. This amendment inverts the whole axis: transmission is off unless asked for, the policy has one representation (`tx_policy.py`) consulted at every transmit site, and every flag parses fail-safe toward silence. | interview (amended ×2) | | **MA8** | **Default channel/scope + 24 h cadence.** The announcement is broadcast on the protocol's **default channel and scope** — Meshtastic channel `CHANNEL_INDEX` (default `0`) via the interface text-send; MeshCore its public/default channel — honoring existing `ALLOWED_CHANNELS`/`HIDDEN_CHANNELS` intent. After the initial ≥ 24 h wait it repeats **every 24 h**, per configured instance domain (an ingestor with several `INSTANCE_DOMAIN` targets announces each with its own numbers and link). TX volume is negligible (~1 frame/day/domain). | proposed | | **MA9** | **Optional duck-typed provider send.** Transmission is exposed as a new **optional** provider method (e.g. `send_channel_announcement(iface, text)`) accessed via `getattr(provider, …, None)` — mirroring the existing optional `self_node_item` extension — so the `@runtime_checkable MeshProtocol` interface and its `A4b` isinstance conformance are unchanged, and a provider that cannot transmit (or a receive-only transport) simply omits it. Both `meshtastic` and `meshcore` providers implement it; neither protocol is privileged. | interview | -| **MA10** | **Engineering bar (D9).** Every new/changed unit ships with 100 % unit tests (counter increment across RX families + TX; heartbeat delta + reset; activity-table write + retention; MAX-aggregation math incl. multi-ingestor; the `/api/stats` field; the four announcement gates incl. fail-closed privacy and the 24 h wait; per-protocol send), full PDoc/RDoc, Apache headers, `black`/`rufo` clean; `pytest`/`rspec`/`npm test` stay green; `CONTRACTS.md` documents the additive heartbeat field, the activity schema, and the `/api/stats` addition. | CLAUDE.md | +| **MA10** | **Engineering bar (D9).** Every new/changed unit ships with 100 % unit tests (counter increment across RX families + TX; heartbeat delta + reset; activity-table write + retention; MAX-aggregation math incl. multi-ingestor; the `/api/stats` field; the four announcement gates (`TX_ENABLED`, `TX_ANNOUNCE`, the 24 h wait, fail-closed privacy) plus the `RX_ONLY` veto, the fail-safe env parsing of each flag, and their delivery through every packaged deployment surface; per-protocol send), full PDoc/RDoc, Apache headers, `black`/`rufo` clean; `pytest`/`rspec`/`npm test` stay green; `CONTRACTS.md` documents the additive heartbeat field, the activity schema, and the `/api/stats` addition. | CLAUDE.md | --- diff --git a/data/Dockerfile b/data/Dockerfile index ed976b08..0a6374a8 100644 --- a/data/Dockerfile +++ b/data/Dockerfile @@ -77,6 +77,8 @@ ENV CONNECTION=/dev/ttyACM0 \ MESH_UDP_PORT=4403 \ ALLOWED_CHANNELS="" \ HIDDEN_CHANNELS="" \ + TX_ENABLED=0 \ + TX_ANNOUNCE=0 \ INSTANCE_DOMAIN="" \ API_TOKEN="" @@ -111,6 +113,8 @@ ENV CONNECTION=/dev/ttyACM0 \ MESH_UDP_PORT=4403 \ ALLOWED_CHANNELS="" \ HIDDEN_CHANNELS="" \ + TX_ENABLED=0 \ + TX_ANNOUNCE=0 \ INSTANCE_DOMAIN="" \ API_TOKEN="" diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 4ed543a4..18cf5cd9 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -184,7 +184,7 @@ Single telemetry payload: - Meta: `ingestor`, `lora_freq`, `modem_preset` - `protocol` (optional string; `"meshtastic"` or `"meshcore"`) — explicit per-record protocol stamp; same semantics as on `POST /api/messages`. -**MeshCore telemetry sourcing (TI-A3).** MeshCore exposes other nodes' telemetry only as on-air *pull* requests (there is no unsolicited telemetry broadcast the companion library surfaces), so the MeshCore provider collects it three ways and normalises every reading into this same payload shape with `protocol="meshcore"`: (1) **host self-telemetry** over the local companion link (`get_bat` → battery millivolts as `voltage`; `get_self_telemetry` → the host's CayenneLPP sensor list), no LoRa airtime, cadence `MESHCORE_SELF_TELEMETRY_SECONDS` (default 3600 s, matching the host-telemetry suppression window; `<= 0` disables); (2) **round-robin contact polling** (`req_telemetry_sync`, falling back to `req_status_sync` when a node reports no sensors) at one on-air request per `MESHCORE_TELEMETRY_POLL_SECONDS` (default 300 s; `<= 0` disables) regardless of roster size, with each contact additionally capped at **one poll per 24 h** (a fixed per-node cooldown, stamped at the poll attempt so unreachable nodes are not hammered; when every contact is fresh the tick transmits nothing) — and `RX_ONLY=1` forbids these on-air polls entirely (receive-only ingestors; the local self reads in (1) are unaffected); (3) **unsolicited/tag-matched events** (`TELEMETRY_RESPONSE`, `STATUS_RESPONSE`, `BATTERY`) whenever the radio surfaces them. CayenneLPP types map to canonical keys (`temperature`, `humidity`→`relative_humidity`, `barometer`→`barometric_pressure`, `voltage`, `current` — scaled A→mA to match the Meshtastic column convention, `illuminance`→`lux`, `percentage`→`battery_level`); status `bat`/`level` millivolt gauges map to `voltage` (V). MeshCore assigns no firmware packet id, so the record `id` is the deterministic 53-bit fingerprint of *(node id, receive second, source kind)* — re-reads of the same source in the same second collapse into one row via the `telemetry.id` upsert. +**MeshCore telemetry sourcing (TI-A3).** MeshCore exposes other nodes' telemetry only as on-air *pull* requests (there is no unsolicited telemetry broadcast the companion library surfaces), so the MeshCore provider collects it three ways and normalises every reading into this same payload shape with `protocol="meshcore"`: (1) **host self-telemetry** over the local companion link (`get_bat` → battery millivolts as `voltage`; `get_self_telemetry` → the host's CayenneLPP sensor list), no LoRa airtime, cadence `MESHCORE_SELF_TELEMETRY_SECONDS` (default 3600 s, matching the host-telemetry suppression window; `<= 0` disables); (2) **round-robin contact polling** (`req_telemetry_sync`, falling back to `req_status_sync` when a node reports no sensors) at one on-air request per `MESHCORE_TELEMETRY_POLL_SECONDS` (default 300 s; `<= 0` disables) regardless of roster size, with each contact additionally capped at **one poll per 24 h** (a fixed per-node cooldown, stamped at the poll attempt so unreachable nodes are not hammered; when every contact is fresh the tick transmits nothing) — and the transmit policy gates these on-air polls entirely — they require `TX_ENABLED=1` (default `0`, so an ingestor polls no other node unless its operator opts in), and the legacy `RX_ONLY=1` vetoes them regardless; the local self reads in (1) cost no airtime and are unaffected; (3) **unsolicited/tag-matched events** (`TELEMETRY_RESPONSE`, `STATUS_RESPONSE`, `BATTERY`) whenever the radio surfaces them. CayenneLPP types map to canonical keys (`temperature`, `humidity`→`relative_humidity`, `barometer`→`barometric_pressure`, `voltage`, `current` — scaled A→mA to match the Meshtastic column convention, `illuminance`→`lux`, `percentage`→`battery_level`); status `bat`/`level` millivolt gauges map to `voltage` (V). MeshCore assigns no firmware packet id, so the record `id` is the deterministic 53-bit fingerprint of *(node id, receive second, source kind)* — re-reads of the same source in the same second collapse into one row via the `telemetry.id` upsert. #### `POST /api/neighbors` diff --git a/data/mesh_ingestor/__init__.py b/data/mesh_ingestor/__init__.py index dfea05ee..c7bc53c6 100644 --- a/data/mesh_ingestor/__init__.py +++ b/data/mesh_ingestor/__init__.py @@ -74,6 +74,10 @@ _CONFIG_ATTRS = { "API_TOKEN", "ALLOWED_CHANNELS", "HIDDEN_CHANNELS", + "TX_ENABLED", + "TX_ANNOUNCE", + "RX_ONLY", + "PROTOCOL", "LORA_FREQ", "MODEM_PRESET", "_RECONNECT_INITIAL_DELAY_SECS", diff --git a/data/mesh_ingestor/announce.py b/data/mesh_ingestor/announce.py index b6a1c640..892cf931 100644 --- a/data/mesh_ingestor/announce.py +++ b/data/mesh_ingestor/announce.py @@ -14,17 +14,19 @@ """Activity announcement: dogfeed the instance API and build the broadcast text. -SPEC decision **MA6** — when the operator opts in (``ANNOUNCE=1``, MA7 a), the -ingestor periodically broadcasts a one-line activity summary on its protocol's -default channel, drawing the numbers **back from the target instance's own API** -(dogfeeding: one radio may not see the whole mesh). +SPEC decision **MA6** — when the operator opts in (``TX_ENABLED=1`` **and** +``TX_ANNOUNCE=1``, MA7 a/b), the ingestor periodically broadcasts a one-line +activity summary on its protocol's default channel, drawing the numbers **back +from the target instance's own API** (dogfeeding: one radio may not see the +whole mesh). This module owns the read-only *dogfeed* HTTP client (GET ``/version`` and ``/api/stats`` — both public, no auth) and the character-limited message builder. The transmit primitive lives on each provider (``send_channel_announcement``, -MA9); the scheduling and the ``ANNOUNCE`` / ``RX_ONLY`` / privacy / 24-hour gates -live in the daemon (MA7/MA8). Every fetch fails **soft** — any network or -shape error returns ``None`` so the caller can fail closed. +MA9) and enforces the transmit gates itself; the transmit policy lives in +:mod:`~data.mesh_ingestor.tx_policy`, and the privacy / 24-hour gates and the +cadence live here and in the daemon (MA7/MA8). Every fetch fails **soft** — any +network or shape error returns ``None`` so the caller can fail closed. """ from __future__ import annotations @@ -33,7 +35,7 @@ import json import time import urllib.request -from . import config +from . import config, tx_policy #: Conservative single-frame text limits per protocol (SPEC MA6). ASCII bytes ≈ #: characters; the announcement template is short, so truncation is only a @@ -215,47 +217,39 @@ def build_announcement( #: The first announcement is withheld until the ingestor has been running this #: long, so the dogfed 24-hour numbers are accurate over a full window and a -#: restart storm cannot spam the channel (SPEC MA7 d). +#: restart storm cannot spam the channel (SPEC MA7 c). ANNOUNCE_INITIAL_DELAY_SECS = 24 * 60 * 60 #: Minimum spacing between announcement cycles once eligible (SPEC MA8). ANNOUNCE_INTERVAL_SECS = 24 * 60 * 60 -def announcements_enabled() -> bool: - """Return whether announcements may be transmitted at all (SPEC MA7 a). +def announce_due( + *, start_monotonic: float, last_announce: float | None, now: float +) -> bool: + """Return whether an announcement cycle is due (SPEC MA7 c / MA8). - Two gates, both of which must permit the send: - :data:`~data.mesh_ingestor.config.ANNOUNCE` must be **on** (opt-in, default - off — an unsolicited line on a shared human channel is not something an - operator should get merely by deploying a map ingestor), and - :data:`~data.mesh_ingestor.config.RX_ONLY` must be **off** (the global - receive-only flag forbids every ingestor TX and so overrides ``ANNOUNCE``). - - Returns: - ``True`` when the transmit gates permit an announcement. - """ - - if not getattr(config, "ANNOUNCE", False): - return False - return not getattr(config, "RX_ONLY", False) - - -def announce_due(*, start_time: float, last_announce: float | None, now: float) -> bool: - """Return whether an announcement cycle is due (SPEC MA7 d / MA8). + All three arguments are :func:`time.monotonic` readings, **not** wall clock. + The 24-hour post-start delay exists so a restart storm cannot spam the + channel, and wall clock cannot express that: an RTC-less host (a Pi or SBC — + exactly the hardware these ingestors run on) boots with a stale clock, NTP + then steps it forward by hours or years, and a wall-clock ``now - + start_time`` clears the delay instantly on the next loop. A monotonic clock + is immune to steps, which is why the rest of the daemon's timers already use + one. Parameters: - start_time: Unix time the ingestor started. - last_announce: Unix time of the previous cycle, or ``None`` if none yet. - now: Current unix time. + start_monotonic: Monotonic reading taken when the ingestor started. + last_announce: Monotonic reading of the previous cycle, or ``None``. + now: Current monotonic reading. Returns: ``True`` when at least :data:`ANNOUNCE_INITIAL_DELAY_SECS` have elapsed - since *start_time* **and** at least :data:`ANNOUNCE_INTERVAL_SECS` since - *last_announce*. + since *start_monotonic* **and** at least :data:`ANNOUNCE_INTERVAL_SECS` + since *last_announce*. """ - if now - start_time < ANNOUNCE_INITIAL_DELAY_SECS: + if now - start_monotonic < ANNOUNCE_INITIAL_DELAY_SECS: return False if last_announce is not None and now - last_announce < ANNOUNCE_INTERVAL_SECS: return False @@ -267,7 +261,7 @@ def send_announcement_to_instance( ) -> bool: """Dogfeed one instance and, unless it is private, build and send the line. - The privacy gate is **fail-closed** (SPEC MA7 c / Invariant II): the + The privacy gate is **fail-closed** (SPEC MA7 d / Invariant II): the announcement is sent only when ``/version`` explicitly reports ``private_mode == False``; a ``True`` flag or any fetch/parse error skips it. @@ -281,6 +275,19 @@ def send_announcement_to_instance( ``True`` when an announcement was transmitted, ``False`` otherwise. """ + # Re-checked here even though the daemon entry point and both shipped + # providers also check: this function is exported, and the send target is a + # duck-typed provider method, so a provider written to the documented + # protocol checklist need not carry the gate itself. The transmit policy + # must not depend on every future provider author remembering it. + if not tx_policy.announcements_permitted(): + config._debug_log( + "Activity announcement skipped: transmit policy forbids it", + context="announce.tx", + url=instance_url, + blocked_by=tx_policy.describe_tx_policy()["blocked_by"], + ) + return False if fetch_private_mode(instance_url) is not False: config._debug_log( "Activity announcement skipped: instance private or unreachable", @@ -362,13 +369,13 @@ def maybe_run_announcements( provider: object, iface: object, *, - start_time: float, + start_monotonic: float, last_announce: float | None, now: float | None = None, ) -> float | None: """Daemon entry point: run an announcement cycle when scheduled (SPEC MA7/MA8). - Applies the enable / RX-only gate and the 24-hour schedule, then dogfeeds and + Applies the transmit gates and the 24-hour schedule, then dogfeeds and announces to each configured instance. The cycle timestamp advances whenever the schedule fires (whether or not any instance was actually announced to), so a private or unreachable instance is retried on the next 24-hour tick @@ -377,9 +384,9 @@ def maybe_run_announcements( Parameters: provider: Active mesh provider. iface: Live mesh interface. - start_time: Unix time the ingestor started. - last_announce: Unix time of the previous cycle, or ``None``. - now: Current unix time; defaults to :func:`time.time`. + start_monotonic: Monotonic reading taken when the ingestor started. + last_announce: Monotonic reading of the previous cycle, or ``None``. + now: Current monotonic reading; defaults to :func:`time.monotonic`. Returns: The updated ``last_announce`` timestamp — *now* when a cycle ran, else @@ -387,10 +394,20 @@ def maybe_run_announcements( """ if now is None: - now = time.time() - if not announcements_enabled(): + now = time.monotonic() + if not tx_policy.announcements_permitted(): + # Name the closed gate: an operator who opted in and hears nothing must + # be able to tell "TX_ENABLED unset" from "RX_ONLY still set" from + # "still inside the 24 h delay" without a day of observation per guess. + config._debug_log( + "Activity announcement suppressed by transmit policy", + context="announce.tx", + blocked_by=tx_policy.describe_tx_policy()["blocked_by"], + ) return last_announce - if not announce_due(start_time=start_time, last_announce=last_announce, now=now): + if not announce_due( + start_monotonic=start_monotonic, last_announce=last_announce, now=now + ): return last_announce run_announcement_cycle(provider, iface) return now @@ -401,7 +418,6 @@ __all__ = [ "ANNOUNCE_INITIAL_DELAY_SECS", "ANNOUNCE_INTERVAL_SECS", "announce_due", - "announcements_enabled", "build_announcement", "fetch_activity", "fetch_private_mode", diff --git a/data/mesh_ingestor/config.py b/data/mesh_ingestor/config.py index 16e7c5e9..0b600a23 100644 --- a/data/mesh_ingestor/config.py +++ b/data/mesh_ingestor/config.py @@ -67,6 +67,107 @@ CHANNEL_INDEX = int(os.environ.get("CHANNEL_INDEX", str(DEFAULT_CHANNEL_INDEX))) DEBUG = os.environ.get("DEBUG") == "1" + +def _debug_log( + message: str, + *, + context: str | None = None, + severity: str = "debug", + always: bool = False, + **metadata: Any, +) -> None: + """Print ``message`` with a UTC timestamp when ``DEBUG`` is enabled. + + Parameters: + message: Text to display when debug logging is active. + context: Optional logical component emitting the message. + severity: Log level label to embed in the formatted output. + always: When ``True``, bypasses the :data:`DEBUG` guard. + **metadata: Additional structured log metadata. + """ + + normalized_severity = severity.lower() + + if not DEBUG and not always and normalized_severity == "debug": + return + + timestamp = datetime.now(timezone.utc).isoformat(timespec="milliseconds") + timestamp = timestamp.replace("+00:00", "Z") + parts = [f"[{timestamp}]", "[potato-mesh]", f"[{normalized_severity}]"] + if context: + parts.append(f"context={context}") + for key, value in sorted(metadata.items()): + parts.append(f"{key}={value!r}") + parts.append(message) + print(" ".join(parts)) + + +#: Values accepted as "on" by :func:`_env_flag`, compared case-insensitively +#: after stripping. Deliberately broader than the historic exact-``"1"`` test: +#: a transmit switch spelled ``true`` must not silently mean *transmit*. +_TRUTHY_FLAG_VALUES = frozenset({"1", "true", "yes", "on"}) + +#: Values accepted as "off" by :func:`_env_flag`. +_FALSY_FLAG_VALUES = frozenset({"0", "false", "no", "off"}) + + +def _env_flag(name: str, *, default: bool, on_invalid: bool) -> bool: + """Resolve a boolean environment variable, failing safe on garbage. + + Boolean env vars were historically compared as ``os.environ.get(X) == "1"``, + which silently resolved ``true``/``TRUE``/``yes``/``" 1"`` to :data:`False`. + For a *transmit* switch that is a fail-**open** bug: an operator who wrote + ``RX_ONLY=true`` got an ingestor that transmitted anyway. This parser + accepts the common spellings, strips surrounding whitespace (so a stray + space in a ``.env`` file is inert, matching :data:`MESH_UDP_PORT`), and + resolves anything it cannot understand to *on_invalid* — which each caller + sets to whichever value means "do not transmit" — after warning loudly. + + A blank value is treated as unset so an empty ``.env`` line means "default", + not "off" — the same blank tolerance :data:`MESH_UDP_PORT` has. + + Unparseable values **warn rather than raise**, deliberately unlike + :data:`TRANSPORT` and :data:`PROTOCOL` (which reject an unknown value at + import) and unlike :data:`MESH_UDP_PORT` (whose ``int()`` raises on + anything non-numeric). Those are *selectors*: with no safe fallback, + refusing to start is the only correct answer. A transmit flag has one — it + can resolve toward silence — so a typo costs the operator a feature they + have to notice is missing, not an ingestor that crash-loops and stops + receiving. Since receiving is the ingestor's whole job, staying up while + transmitting nothing strictly dominates. + + Parameters: + name: Environment variable to read. + default: Value used when the variable is unset or blank. + on_invalid: Value used when the variable holds an unrecognized string. + Callers pass the fail-safe side of their own switch. + + Returns: + The resolved boolean. + """ + + raw = os.environ.get(name) + if raw is None: + return default + normalized = raw.strip().lower() + if not normalized: + return default + if normalized in _TRUTHY_FLAG_VALUES: + return True + if normalized in _FALSY_FLAG_VALUES: + return False + _debug_log( + f"Unrecognized {name} value; treating as {'1' if on_invalid else '0'}", + context="config", + severity="warning", + variable=name, + value=raw, + resolved=on_invalid, + accepted=sorted(_TRUTHY_FLAG_VALUES | _FALSY_FLAG_VALUES), + ) + return on_invalid + + _KNOWN_PROTOCOLS = ("meshtastic", "meshcore") _raw_protocol = os.environ.get("PROTOCOL", "meshtastic").strip().lower() @@ -140,28 +241,53 @@ raise ``ValueError`` at import and prevent the service from starting.""" INGESTOR_NODE_ID = os.environ.get("INGESTOR_NODE_ID", "").strip() or None """Optional ``!xxxxxxxx`` host node id used for the ingestor heartbeat in UDP mode.""" -RX_ONLY = os.environ.get("RX_ONLY") == "1" -"""Receive-only mode: forbid every ingestor-initiated mesh transmission. +TX_ENABLED = _env_flag("TX_ENABLED", default=False, on_invalid=False) +"""Master switch for every ingestor-initiated mesh transmission (SPEC MA7 a). -Some operators run listening posts where any TX is undesired. When set, the -ingestor never transmits on the mesh: this disables the MeshCore contact -telemetry/status polls and the periodic activity announcement (SPEC MA7), the -only ingestor-initiated RF traffic. Local companion-link reads (host -self-telemetry, contact roster, channel queries) are not transmissions and -continue to work.""" +**Default off.** An ingestor is a listener first: deploying one to feed a +dashboard does not imply consent to put traffic on the community's air. Until +this is set, the ingestor is a pure receiver — no activity announcement, and no +MeshCore on-air contact telemetry/status polls. -ANNOUNCE = os.environ.get("ANNOUNCE") == "1" -"""Opt in to the periodic activity announcement (SPEC MA7 a). +Local companion-link reads (host self-telemetry, the contact roster, channel +queries) are *not* mesh transmissions: they travel over USB/BLE to the operator's +own radio, cost no airtime, and continue regardless of this flag. -The announcement is the ingestor's only *unsolicited* transmission on a shared -human channel, so it is **off by default**: an operator who deploys an ingestor -to feed a map does not implicitly agree to also speak on their community's -default channel, and local conventions on automated traffic differ widely. Set -``ANNOUNCE=1`` to enable it. +Set ``TX_ENABLED=1`` to allow transmission. This is the only documented transmit +switch; :data:`RX_ONLY` is its retired predecessor and still vetoes it. +Announcements additionally require :data:`TX_ANNOUNCE`. +""" -This is an opt-in on top of :data:`RX_ONLY`, not a replacement for it: -``RX_ONLY=1`` still forbids *every* ingestor-initiated transmission and wins -over ``ANNOUNCE=1``.""" +TX_ANNOUNCE = _env_flag("TX_ANNOUNCE", default=False, on_invalid=False) +"""Opt in to the periodic activity announcement (SPEC MA7 b). + +**Default off**, and subordinate to :data:`TX_ENABLED`: the announcement is the +ingestor's only *unsolicited* transmission on a shared human channel, and local +conventions on automated traffic differ widely, so it takes a second deliberate +opt-in on top of the master switch. ``TX_ENABLED=0`` overrides +``TX_ANNOUNCE=1``; both must be on before a single announcement is sent. +""" + +RX_ONLY = _env_flag("RX_ONLY", default=False, on_invalid=True) +"""Legacy receive-only kill switch — superseded by :data:`TX_ENABLED`. + +Honored wherever it reaches the process environment, so an existing deployment +that sets it keeps working, but removed from operator-facing documentation and +deliberately **not** added to the packaged deployment surfaces (compose, the +image, the Nix module) — it was never settable there, and new configurations +express the same intent by simply leaving :data:`TX_ENABLED` unset, which is now +the default. + +It remains a **veto**. Where it is set, no ingestor-initiated transmission +happens even if ``TX_ENABLED=1`` is also present — a kill switch an operator +deliberately engaged is never silently overridden by a flag that arrives later +in the same ``.env`` file. That combination is contradictory, so +:func:`~data.mesh_ingestor.tx_policy.log_tx_policy` warns about it at startup. + +Note the deliberately asymmetric fail-safe: an unparseable value resolves to +:data:`True` (silence), because for a kill switch the safe reading of garbage is +*engaged*. +""" MESHCORE_TELEMETRY_POLL_SECONDS = int( os.environ.get("MESHCORE_TELEMETRY_POLL_SECONDS", "300").strip() or "300" @@ -380,40 +506,6 @@ _INGESTOR_HEARTBEAT_SECS = DEFAULT_INGESTOR_HEARTBEAT_SECS _SELF_NODE_REPORT_INTERVAL_SECS = DEFAULT_SELF_NODE_REPORT_INTERVAL_SECS -def _debug_log( - message: str, - *, - context: str | None = None, - severity: str = "debug", - always: bool = False, - **metadata: Any, -) -> None: - """Print ``message`` with a UTC timestamp when ``DEBUG`` is enabled. - - Parameters: - message: Text to display when debug logging is active. - context: Optional logical component emitting the message. - severity: Log level label to embed in the formatted output. - always: When ``True``, bypasses the :data:`DEBUG` guard. - **metadata: Additional structured log metadata. - """ - - normalized_severity = severity.lower() - - if not DEBUG and not always and normalized_severity == "debug": - return - - timestamp = datetime.now(timezone.utc).isoformat(timespec="milliseconds") - timestamp = timestamp.replace("+00:00", "Z") - parts = [f"[{timestamp}]", "[potato-mesh]", f"[{normalized_severity}]"] - if context: - parts.append(f"context={context}") - for key, value in sorted(metadata.items()): - parts.append(f"{key}={value!r}") - parts.append(message) - print(" ".join(parts)) - - __all__ = [ "CONNECTION", "SNAPSHOT_SECS", @@ -434,6 +526,10 @@ __all__ = [ "MESH_UDP_GROUP", "MESH_UDP_PORT", "INGESTOR_NODE_ID", + "TX_ENABLED", + "TX_ANNOUNCE", + "RX_ONLY", + "PROTOCOL", "_RECONNECT_INITIAL_DELAY_SECS", "_RECONNECT_MAX_DELAY_SECS", "_CLOSE_TIMEOUT_SECS", diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index 5e6114c0..0f647144 100644 --- a/data/mesh_ingestor/daemon.py +++ b/data/mesh_ingestor/daemon.py @@ -24,7 +24,7 @@ import time from pubsub import pub -from . import announce, config, handlers, ingestors, interfaces, queue +from . import announce, config, handlers, ingestors, interfaces, queue, tx_policy from .mesh_protocol import MeshProtocol from .utils import _retry_dict_snapshot @@ -582,8 +582,8 @@ def _try_send_self_node(state: _DaemonState) -> None: def _process_announcements(state: _DaemonState) -> float | None: """Run the periodic activity-announcement cycle when scheduled (SPEC MA6-MA8). - Delegates the transmit gates (the ``ANNOUNCE`` opt-in and the ``RX_ONLY`` - override), the per-instance fail-closed + Delegates the transmit gates (``TX_ENABLED`` / ``TX_ANNOUNCE``, and the + legacy ``RX_ONLY`` veto), the per-instance fail-closed privacy check, the >=24h post-start delay, and the 24h cadence to :func:`~data.mesh_ingestor.announce.maybe_run_announcements`, which dogfeeds each configured instance's own API for the numbers it announces. A no-op @@ -603,7 +603,7 @@ def _process_announcements(state: _DaemonState) -> float | None: return announce.maybe_run_announcements( state.provider, state.iface, - start_time=ingestors.ingestor_start_time(), + start_monotonic=ingestors.ingestor_start_monotonic(), last_announce=state.last_announce, ) @@ -658,6 +658,11 @@ def _loop_iteration(state: _DaemonState) -> bool: def main(*, provider: MeshProtocol | None = None) -> None: """Run the mesh ingestion daemon until interrupted.""" + # State the resolved transmit policy before anything can transmit, so what + # this ingestor will and will not put on the air is visible in the log from + # the first line rather than inferred from silence (SPEC MA7). + tx_policy.log_tx_policy() + if provider is None: if config.PROTOCOL == "meshcore": from .protocols.meshcore import MeshcoreProvider diff --git a/data/mesh_ingestor/ingestors.py b/data/mesh_ingestor/ingestors.py index e4899d42..eb5c8bef 100644 --- a/data/mesh_ingestor/ingestors.py +++ b/data/mesh_ingestor/ingestors.py @@ -33,6 +33,11 @@ class _IngestorState: """Mutable ingestor identity and heartbeat tracking data.""" start_time: int = field(default_factory=lambda: int(time.time())) + #: Monotonic companion to :attr:`start_time`, used for elapsed-time + #: decisions. ``start_time`` is wall clock because it goes out on the + #: heartbeat wire; wall clock cannot measure "24 h since boot" on a host + #: whose clock NTP steps after start, so scheduling reads this instead. + start_monotonic: float = field(default_factory=time.monotonic) last_heartbeat: int | None = None node_id: str | None = None @@ -45,11 +50,25 @@ INGESTOR_STATE = STATE def ingestor_start_time() -> int: - """Return the unix timestamp representing when the ingestor booted.""" + """Return the unix timestamp representing when the ingestor booted. + + Wall clock, because it is reported on the heartbeat wire. For elapsed-time + decisions use :func:`ingestor_start_monotonic` instead. + """ return STATE.start_time +def ingestor_start_monotonic() -> float: + """Return the monotonic reading taken when the ingestor booted. + + Step-immune, so it is the correct basis for "has N seconds elapsed since + start?" checks such as the announcement's 24-hour initial delay (SPEC MA7 c). + """ + + return STATE.start_monotonic + + def set_ingestor_node_id(node_id: str | None) -> str | None: """Record the canonical host node identifier for the ingestor. @@ -140,6 +159,7 @@ __all__ = [ "HEARTBEAT_INTERVAL_SECS", "INGESTOR_STATE", "ingestor_start_time", + "ingestor_start_monotonic", "queue_ingestor_heartbeat", "set_ingestor_node_id", ] diff --git a/data/mesh_ingestor/protocols/meshcore/provider.py b/data/mesh_ingestor/protocols/meshcore/provider.py index 9cc4b052..0a418542 100644 --- a/data/mesh_ingestor/protocols/meshcore/provider.py +++ b/data/mesh_ingestor/protocols/meshcore/provider.py @@ -20,7 +20,7 @@ import asyncio import sys import threading -from ... import activity, config +from ... import activity, config, tx_policy from ._constants import _ANNOUNCE_SEND_TIMEOUT_SECS, _CONNECT_TIMEOUT_SECS from .decode import _self_info_to_node_dict from .identity import _meshcore_node_id @@ -215,11 +215,19 @@ class MeshcoreProvider: daemon resolves it via ``getattr`` and skips it when absent. It is a no-op when the interface has no live event loop / handle. + Enforces the transmit gates **here**, at the transmit primitive, rather + than relying on the caller: this method is a public, duck-typed provider + capability, so any second caller (a CLI, an operator script, a future + scheduler) must not be able to put traffic on the air by reaching past + the daemon's own check. + Parameters: iface: Active :class:`_MeshcoreInterface`. text: Announcement string to transmit. """ + if not tx_policy.announcements_permitted(): + return if not isinstance(iface, _MeshcoreInterface): return mc = getattr(iface, "_mc", None) diff --git a/data/mesh_ingestor/protocols/meshcore/telemetry.py b/data/mesh_ingestor/protocols/meshcore/telemetry.py index fa2a0b8a..c9543985 100644 --- a/data/mesh_ingestor/protocols/meshcore/telemetry.py +++ b/data/mesh_ingestor/protocols/meshcore/telemetry.py @@ -35,7 +35,7 @@ import asyncio import time from collections.abc import Mapping -from ... import activity, config +from ... import activity, config, tx_policy from .interface import _MeshcoreInterface from .messages import _derive_message_id @@ -419,6 +419,11 @@ async def _poll_contact_telemetry( handlers: The ``data.mesh_ingestor.handlers`` module. state: Mutable poll-loop state (round-robin cursor). """ + # Checked at each transmit site beside its record_tx, as well as at loop + # entry: the loop resolves the poll interval once at startup, so a policy + # read later must still be honoured by the request that is about to go out. + if not tx_policy.transmit_permitted(): + return contact = _next_poll_contact(iface, state) if contact is None: return @@ -448,6 +453,11 @@ async def _poll_contact_telemetry( handlers, node_id, _lpp_to_telemetry_section(lpp), "lpp" ): return + # The status fallback is a second, independent transmission, so it takes its + # own permission check rather than riding the one above: one gate per + # transmit site, beside that site's record_tx. + if not tx_policy.transmit_permitted(): + return try: # The status fallback is a second on-air pull — count it too (MA1). activity.record_tx() @@ -478,10 +488,11 @@ async def _telemetry_poll_loop(mc, iface: _MeshcoreInterface) -> None: ``MESHCORE_SELF_TELEMETRY_SECONDS`` (local, no airtime; ``<= 0`` disables) and ``MESHCORE_TELEMETRY_POLL_SECONDS`` (one on-air request per interval; ``<= 0`` disables), with each contact additionally capped by the fixed - per-node cooldown (:data:`_TELEMETRY_NODE_COOLDOWN_SECONDS`). ``RX_ONLY`` - forbids every ingestor-initiated transmission, so it disables the on-air - contact polls regardless of the poll interval; the self reads are local - companion-link commands and stay active. The loop wakes once per + per-node cooldown (:data:`_TELEMETRY_NODE_COOLDOWN_SECONDS`). The transmit + policy (:func:`~data.mesh_ingestor.tx_policy.transmit_permitted` — off unless + ``TX_ENABLED=1``) disables the on-air contact polls regardless of the poll + interval; the self reads are local companion-link commands, cost no airtime, + and stay active. The loop wakes once per second-granularity deadline rather than busy-polling. Parameters: @@ -491,7 +502,9 @@ async def _telemetry_poll_loop(mc, iface: _MeshcoreInterface) -> None: from ... import handlers as _handlers self_interval = config.MESHCORE_SELF_TELEMETRY_SECONDS - poll_interval = 0 if config.RX_ONLY else config.MESHCORE_TELEMETRY_POLL_SECONDS + poll_interval = ( + config.MESHCORE_TELEMETRY_POLL_SECONDS if tx_policy.transmit_permitted() else 0 + ) if self_interval <= 0 and poll_interval <= 0: return diff --git a/data/mesh_ingestor/protocols/meshtastic.py b/data/mesh_ingestor/protocols/meshtastic.py index d276ed9d..e2cf50d4 100644 --- a/data/mesh_ingestor/protocols/meshtastic.py +++ b/data/mesh_ingestor/protocols/meshtastic.py @@ -18,7 +18,7 @@ from __future__ import annotations from pubsub import pub -from .. import activity, config, daemon as _daemon, handlers, interfaces +from .. import activity, config, daemon as _daemon, handlers, interfaces, tx_policy from ..utils import _retry_dict_snapshot @@ -105,11 +105,19 @@ class MeshtasticProvider: method (not a formal :class:`MeshProtocol` member); the daemon resolves it via ``getattr`` and skips it when absent. + Enforces the transmit gates **here**, at the transmit primitive, rather + than relying on the caller: this method is a public, duck-typed provider + capability, so any second caller (a CLI, an operator script, a future + scheduler) must not be able to put traffic on the air by reaching past + the daemon's own check. + Parameters: iface: Active Meshtastic interface exposing ``sendText``. text: Announcement string to transmit. """ + if not tx_policy.announcements_permitted(): + return send_text = getattr(iface, "sendText", None) if not callable(send_text): return diff --git a/data/mesh_ingestor/tx_policy.py b/data/mesh_ingestor/tx_policy.py new file mode 100644 index 00000000..e80458a6 --- /dev/null +++ b/data/mesh_ingestor/tx_policy.py @@ -0,0 +1,171 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""The ingestor's single transmit-permission predicate (SPEC MA7). + +PotatoMesh ingestors are listeners first. Everything they put on the air is +therefore gated here, in one place, rather than re-derived by each caller. + +**Why this module exists.** Transmit policy used to be expressed twice, at two +different altitudes and in two different idioms: ``announce.py`` composed two +defensive ``getattr`` reads, while the MeshCore telemetry loop wrote +``poll_interval = 0 if config.RX_ONLY else ...`` at loop entry. Two independent +sites meant two independent chances to get the parsing wrong (both did), the +gate sat a layer *above* the transmit primitives rather than beside them, and +adding a knob meant finding every idiom by hand. One predicate, consulted at +each transmit site next to :func:`~data.mesh_ingestor.activity.record_tx`, makes +"may I transmit?" and "count this transmission" one decision per site. + +**The policy.** Two operator flags and one retired predecessor: + +=============== ========= ========================================= +``TX_ENABLED`` ``RX_ONLY`` transmit? +=============== ========= ========================================= +unset unset **no** — the default; a pure listener +unset set no +set unset **yes** +set set no — the legacy veto wins, and warns +=============== ========= ========================================= + +:data:`~data.mesh_ingestor.config.TX_ANNOUNCE` then gates the announcement +specifically, on top of a permitting :func:`transmit_permitted`. + +Only *mesh* transmissions are covered. Companion-link reads over USB/BLE to the +operator's own radio (host self-telemetry, contact roster, channel queries) cost +no airtime, are not transmissions, and are never gated here. +""" + +from __future__ import annotations + +from typing import Any + +from . import config + +#: Context label used for every log line this module emits. +_LOG_CONTEXT = "tx.policy" + + +def transmit_permitted() -> bool: + """Return whether the ingestor may transmit on the mesh at all (SPEC MA7 a). + + The one predicate every transmit site consults. Call it immediately before + the send, beside :func:`~data.mesh_ingestor.activity.record_tx`, so the + decision to transmit and the accounting for it stay together. + + Returns: + ``True`` only when :data:`~data.mesh_ingestor.config.TX_ENABLED` is on + and the legacy :data:`~data.mesh_ingestor.config.RX_ONLY` veto is off. + """ + + return bool(config.TX_ENABLED) and not config.RX_ONLY + + +def announcements_permitted() -> bool: + """Return whether the periodic activity announcement may be sent (SPEC MA7 b). + + Strictly narrower than :func:`transmit_permitted`: announcing is unsolicited + traffic on a shared human channel, so it needs its own opt-in on top of the + master switch. + + Returns: + ``True`` when transmission is permitted **and** + :data:`~data.mesh_ingestor.config.TX_ANNOUNCE` is on. + """ + + return transmit_permitted() and bool(config.TX_ANNOUNCE) + + +def describe_tx_policy() -> dict[str, Any]: + """Return the resolved transmit policy as a structured mapping. + + Used for the startup log line and for the "which gate closed?" diagnostic. + An operator who opts in and hears nothing on air must be able to tell *why* + without waiting out a 24-hour observation window per hypothesis. + + ``blocked_by`` names the **first** flag that forbids announcing, checked + outermost-first (master switch, then legacy veto, then the announcement + opt-in), or ``None`` when every gate permits. + + Returns: + Mapping with the three raw flags, both derived permissions, and + ``blocked_by``. + """ + + tx_enabled = bool(config.TX_ENABLED) + rx_only = bool(config.RX_ONLY) + tx_announce = bool(config.TX_ANNOUNCE) + + if not tx_enabled: + blocked_by: str | None = "TX_ENABLED" + elif rx_only: + blocked_by = "RX_ONLY" + elif not tx_announce: + blocked_by = "TX_ANNOUNCE" + else: + blocked_by = None + + return { + "tx_enabled": tx_enabled, + "tx_announce": tx_announce, + "rx_only": rx_only, + "transmit_permitted": transmit_permitted(), + "announcements_permitted": announcements_permitted(), + "blocked_by": blocked_by, + } + + +def log_tx_policy() -> None: + """State the resolved transmit policy once, at startup. + + Emitted at ``info`` (not ``debug``) so it appears without ``DEBUG=1``: what + an ingestor will and will not put on the air is operator-visible behavior, + not a debugging detail, and silence is otherwise indistinguishable from + misconfiguration. + + Additionally warns when :data:`~data.mesh_ingestor.config.TX_ENABLED` and the + legacy :data:`~data.mesh_ingestor.config.RX_ONLY` are both set. That state + is contradictory but representable — typically a stale ``.env`` line meeting + a newly-added one — and it resolves to *silence*, which is the surprising + direction for whoever just set ``TX_ENABLED=1``. + """ + + policy = describe_tx_policy() + + if policy["tx_enabled"] and policy["rx_only"]: + _debug_log = config._debug_log + _debug_log( + "TX_ENABLED=1 is overridden by the legacy RX_ONLY=1 kill switch; " + "nothing will be transmitted. Unset RX_ONLY to allow transmission.", + context=_LOG_CONTEXT, + severity="warning", + ) + + config._debug_log( + "Transmit policy resolved", + context=_LOG_CONTEXT, + severity="info", + always=True, + tx_enabled=policy["tx_enabled"], + tx_announce=policy["tx_announce"], + rx_only=policy["rx_only"], + transmit_permitted=policy["transmit_permitted"], + announcements_permitted=policy["announcements_permitted"], + ) + + +__all__ = [ + "transmit_permitted", + "announcements_permitted", + "describe_tx_policy", + "log_tx_policy", +] diff --git a/docker-compose.yml b/docker-compose.yml index a03cff57..0279ba4c 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -57,6 +57,8 @@ x-ingestor-base: &ingestor-base DEBUG: ${DEBUG:-0} PROTOCOL: ${PROTOCOL:-meshtastic} ENERGY_SAVING: ${ENERGY_SAVING:-0} + TX_ENABLED: ${TX_ENABLED:-0} + TX_ANNOUNCE: ${TX_ANNOUNCE:-0} FEDERATION: ${FEDERATION:-1} PRIVATE: ${PRIVATE:-0} volumes: diff --git a/flake.nix b/flake.nix index a5f23b3f..6f7a6869 100644 --- a/flake.nix +++ b/flake.nix @@ -290,6 +290,28 @@ default = "/dev/ttyACM0"; description = "Connection target: serial port, IP:port for TCP, or Bluetooth address for BLE"; }; + + txEnabled = lib.mkOption { + type = lib.types.bool; + default = false; + description = '' + Allow the ingestor to transmit on the mesh. Off by default: the + ingestor is a listener unless you opt in. Enables MeshCore on-air + contact telemetry polling. Companion-link reads to your own radio + are not transmissions and are unaffected. + ''; + }; + + txAnnounce = lib.mkOption { + type = lib.types.bool; + default = false; + description = '' + Broadcast a one-line activity summary on the default channel, at + most once per 24h and never in the first 24h after start. Requires + txEnabled. Unsolicited automated traffic on a shared channel, so it + is off by default. + ''; + }; }; }; @@ -360,6 +382,8 @@ CONNECTION = cfg.ingestor.connection; DEBUG = if cfg.debug then "1" else "0"; XDG_DATA_HOME = cfg.dataDir; + TX_ENABLED = if cfg.ingestor.txEnabled then "1" else "0"; + TX_ANNOUNCE = if cfg.ingestor.txAnnounce then "1" else "0"; } // lib.optionalAttrs (cfg.allowedChannels != null) { ALLOWED_CHANNELS = cfg.allowedChannels; } // lib.optionalAttrs (cfg.hiddenChannels != null) { diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 00000000..0b72c065 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,48 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Shared pytest fixtures for the ingestor test suite.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +import data.mesh_ingestor.config as config + + +@pytest.fixture +def permit_tx(monkeypatch): + """Permit mesh transmission for a test that exercises a transmit path. + + Transmission is off by default (SPEC MA7), so a test of transmit *machinery* + would otherwise stop at the policy gate and pass vacuously — asserting "no + frame was sent" while never reaching the code it means to check. Request + this fixture to isolate the machinery from the policy; the policy itself is + covered in ``tests/test_tx_policy_unit.py``. + + Yields: + The patched :mod:`~data.mesh_ingestor.config` module, so a test can + narrow the policy further (e.g. re-engage ``RX_ONLY``). + """ + + monkeypatch.setattr(config, "TX_ENABLED", True) + monkeypatch.setattr(config, "TX_ANNOUNCE", True) + monkeypatch.setattr(config, "RX_ONLY", False) + return config diff --git a/tests/test_announce_unit.py b/tests/test_announce_unit.py index 1ba98236..2a9edea4 100644 --- a/tests/test_announce_unit.py +++ b/tests/test_announce_unit.py @@ -273,40 +273,6 @@ def _stub_dogfeed(monkeypatch, *, private=False, numbers=(12, 50)): monkeypatch.setattr(announce, "fetch_activity", lambda url, proto, **_kw: numbers) -class TestAnnouncementGates: - """Tests for :func:`announce.announcements_enabled` (SPEC MA7 a).""" - - def test_gate_disabled_by_default(self, monkeypatch): - """Off by default: ANNOUNCE unset means no unsolicited TX.""" - monkeypatch.setattr(announce.config, "ANNOUNCE", False) - monkeypatch.setattr(announce.config, "RX_ONLY", False) - assert announce.announcements_enabled() is False - - def test_gate_enabled_when_opted_in(self, monkeypatch): - """Enabled when the operator opts in and RX_ONLY is off.""" - monkeypatch.setattr(announce.config, "ANNOUNCE", True) - monkeypatch.setattr(announce.config, "RX_ONLY", False) - assert announce.announcements_enabled() is True - - def test_gate_disabled_when_rx_only(self, monkeypatch): - """RX_ONLY forbids every ingestor TX and so overrides the opt-in.""" - monkeypatch.setattr(announce.config, "ANNOUNCE", True) - monkeypatch.setattr(announce.config, "RX_ONLY", True) - assert announce.announcements_enabled() is False - - def test_gate_disabled_when_rx_only_and_not_opted_in(self, monkeypatch): - """Both gates closed is still closed (no double-negative surprise).""" - monkeypatch.setattr(announce.config, "ANNOUNCE", False) - monkeypatch.setattr(announce.config, "RX_ONLY", True) - assert announce.announcements_enabled() is False - - def test_gate_defaults_to_off_when_attribute_missing(self, monkeypatch): - """A config module without ``ANNOUNCE`` fails closed, not open.""" - monkeypatch.delattr(announce.config, "ANNOUNCE", raising=False) - monkeypatch.setattr(announce.config, "RX_ONLY", False) - assert announce.announcements_enabled() is False - - class TestAnnounceDue: """Tests for :func:`announce.announce_due` (SPEC MA7 d / MA8).""" @@ -314,7 +280,9 @@ class TestAnnounceDue: """Withheld until 24 h after start.""" now = 1_000_000 assert ( - announce.announce_due(start_time=now - 1000, last_announce=None, now=now) + announce.announce_due( + start_monotonic=now - 1000, last_announce=None, now=now + ) is False ) @@ -322,7 +290,9 @@ class TestAnnounceDue: """Eligible once 24 h have elapsed since start.""" now = 1_000_000 assert ( - announce.announce_due(start_time=now - 90_000, last_announce=None, now=now) + announce.announce_due( + start_monotonic=now - 90_000, last_announce=None, now=now + ) is True ) @@ -331,7 +301,7 @@ class TestAnnounceDue: now = 1_000_000 assert ( announce.announce_due( - start_time=now - 200_000, last_announce=now - 3600, now=now + start_monotonic=now - 200_000, last_announce=now - 3600, now=now ) is False ) @@ -341,7 +311,7 @@ class TestAnnounceDue: now = 1_000_000 assert ( announce.announce_due( - start_time=now - 200_000, last_announce=now - 90_000, now=now + start_monotonic=now - 200_000, last_announce=now - 90_000, now=now ) is True ) @@ -350,6 +320,16 @@ class TestAnnounceDue: class TestSendAnnouncementToInstance: """Tests for :func:`announce.send_announcement_to_instance` (MA6/MA7).""" + @pytest.fixture(autouse=True) + def _permit(self, permit_tx): + """Isolate these tests from the transmit policy. + + Transmission is off by default, so without this every test here would + stop at the gate and pass vacuously. The gate itself is covered by + ``test_refuses_when_transmit_policy_closed`` below and by + ``tests/test_tx_policy_unit.py``. + """ + def test_sends_the_built_line_when_public(self, monkeypatch): """A public instance receives the dogfed announcement line.""" _stub_dogfeed(monkeypatch, private=False, numbers=(12, 50)) @@ -452,10 +432,38 @@ class TestSendAnnouncementToInstance: ) assert any("skipped: instance private" in msg for msg, _meta in logs) + def test_refuses_when_transmit_policy_closed(self, monkeypatch): + """The exported send path re-checks the policy itself. + + ``send_announcement_to_instance`` is in ``__all__`` and dispatches to a + duck-typed provider method, so a provider written without the gate must + still not be able to transmit. + """ + monkeypatch.setattr(announce.config, "TX_ENABLED", False) + _stub_dogfeed(monkeypatch) + provider = _RecordingProvider() + assert ( + announce.send_announcement_to_instance( + provider, "IFACE", "https://x", "meshtastic" + ) + is False + ) + assert provider.sent == [] + class TestRunAnnouncementCycle: """Tests for :func:`announce.run_announcement_cycle` (SPEC MA8 per-domain).""" + @pytest.fixture(autouse=True) + def _permit(self, permit_tx): + """Isolate these tests from the transmit policy. + + Transmission is off by default, so without this every test here would + stop at the gate and pass vacuously. The gate itself is covered by + ``test_refuses_when_transmit_policy_closed`` below and by + ``tests/test_tx_policy_unit.py``. + """ + def test_domains_announces_each_configured_instance(self, monkeypatch): """Every configured instance is announced to, with its own link.""" monkeypatch.setattr( @@ -494,21 +502,23 @@ class TestMaybeRunAnnouncements: def test_gate_noop_when_not_enabled(self, monkeypatch): """No cycle runs and last_announce is unchanged when RX_ONLY forbids TX.""" - monkeypatch.setattr(announce.config, "ANNOUNCE", True) + monkeypatch.setattr(announce.config, "TX_ENABLED", True) + monkeypatch.setattr(announce.config, "TX_ANNOUNCE", True) monkeypatch.setattr(announce.config, "RX_ONLY", True) ran = [] monkeypatch.setattr( announce, "run_announcement_cycle", lambda *a, **k: ran.append(True) ) result = announce.maybe_run_announcements( - "P", "I", start_time=0, last_announce=42.0, now=1_000_000 + "P", "I", start_monotonic=0, last_announce=42.0, now=1_000_000 ) assert result == 42.0 assert ran == [] def test_cadence_noop_when_not_due(self, monkeypatch): """No cycle runs before the 24 h delay; last_announce is unchanged.""" - monkeypatch.setattr(announce.config, "ANNOUNCE", True) + monkeypatch.setattr(announce.config, "TX_ENABLED", True) + monkeypatch.setattr(announce.config, "TX_ANNOUNCE", True) monkeypatch.setattr(announce.config, "RX_ONLY", False) ran = [] monkeypatch.setattr( @@ -516,14 +526,15 @@ class TestMaybeRunAnnouncements: ) now = 1_000_000 result = announce.maybe_run_announcements( - "P", "I", start_time=now - 1000, last_announce=None, now=now + "P", "I", start_monotonic=now - 1000, last_announce=None, now=now ) assert result is None assert ran == [] def test_cadence_runs_cycle_and_advances_timestamp(self, monkeypatch): """When due, the cycle runs and last_announce advances to now.""" - monkeypatch.setattr(announce.config, "ANNOUNCE", True) + monkeypatch.setattr(announce.config, "TX_ENABLED", True) + monkeypatch.setattr(announce.config, "TX_ANNOUNCE", True) monkeypatch.setattr(announce.config, "RX_ONLY", False) ran = [] monkeypatch.setattr( @@ -533,16 +544,64 @@ class TestMaybeRunAnnouncements: ) now = 1_000_000 result = announce.maybe_run_announcements( - "P", "IFACE", start_time=now - 90_000, last_announce=None, now=now + "P", "IFACE", start_monotonic=now - 90_000, last_announce=None, now=now ) assert result == now assert ran == [("P", "IFACE")] - def test_cadence_defaults_now_to_wall_clock(self, monkeypatch): - """With ``now`` omitted the current wall clock is used (gate-off path).""" - monkeypatch.setattr(announce.config, "ANNOUNCE", True) - monkeypatch.setattr(announce.config, "RX_ONLY", True) - result = announce.maybe_run_announcements( - "P", "I", start_time=0, last_announce=7.0 + def test_defaults_now_to_monotonic_not_wall_clock(self, monkeypatch): + """With ``now`` omitted the *monotonic* clock is used, never wall clock. + + Guards the NTP-step defect: a wall-clock basis lets a clock jump on an + RTC-less host clear the 24 h anti-spam delay instantly. + """ + monkeypatch.setattr(announce.config, "TX_ENABLED", True) + monkeypatch.setattr(announce.config, "TX_ANNOUNCE", True) + monkeypatch.setattr(announce.config, "RX_ONLY", False) + monkeypatch.setattr(announce.time, "monotonic", lambda: 500_000.0) + monkeypatch.setattr( + announce.time, "time", lambda: pytest.fail("wall clock consulted") ) - assert result == 7.0 + ran = [] + monkeypatch.setattr( + announce, "run_announcement_cycle", lambda *a, **k: ran.append(True) + ) + # 500_000 s since start comfortably exceeds the 24 h delay. + result = announce.maybe_run_announcements( + "P", "I", start_monotonic=0.0, last_announce=None + ) + assert result == 500_000.0 + assert ran == [True] + + @pytest.mark.parametrize( + ("flags", "expected_gate"), + [ + ({}, "TX_ENABLED"), + ({"TX_ANNOUNCE": True}, "TX_ENABLED"), + ({"TX_ENABLED": True}, "TX_ANNOUNCE"), + ({"TX_ENABLED": True, "TX_ANNOUNCE": True, "RX_ONLY": True}, "RX_ONLY"), + ], + ) + def test_suppressed_paths_run_no_cycle_and_name_the_gate( + self, monkeypatch, capsys, flags, expected_gate + ): + """Every closed-gate path is a silent-on-air no-op that logs *why*. + + The default (no flags set) is the case every deployment now hits, and it + was previously exercised only at the predicate level — never through the + daemon entry point, so a reordering that ran the cycle before the gate + would not have been caught. + """ + monkeypatch.setattr(announce.config, "DEBUG", True) + for name in ("TX_ENABLED", "TX_ANNOUNCE", "RX_ONLY"): + monkeypatch.setattr(announce.config, name, flags.get(name, False)) + ran = [] + monkeypatch.setattr( + announce, "run_announcement_cycle", lambda *a, **k: ran.append(True) + ) + result = announce.maybe_run_announcements( + "P", "I", start_monotonic=0, last_announce=42.0, now=1_000_000 + ) + assert result == 42.0 + assert ran == [] + assert f"blocked_by={expected_gate!r}" in capsys.readouterr().out diff --git a/tests/test_daemon_unit.py b/tests/test_daemon_unit.py index c1674e58..e9e29dee 100644 --- a/tests/test_daemon_unit.py +++ b/tests/test_daemon_unit.py @@ -842,15 +842,17 @@ def test_process_announcements_noop_without_iface(): def test_process_announcements_delegates_when_connected(monkeypatch): """Delegates to announce.maybe_run_announcements with the daemon's provider, - interface, ingestor start time, and last_announce, and returns its result.""" + interface, monotonic ingestor start marker, and last_announce, returning its + result. The marker is monotonic, not wall clock, so an NTP step cannot skip + the 24 h anti-spam delay.""" calls = [] - def _fake(provider, iface, *, start_time, last_announce): - calls.append((provider, iface, start_time, last_announce)) + def _fake(provider, iface, *, start_monotonic, last_announce): + calls.append((provider, iface, start_monotonic, last_announce)) return 999.0 monkeypatch.setattr(daemon.announce, "maybe_run_announcements", _fake) - monkeypatch.setattr(daemon.ingestors, "ingestor_start_time", lambda: 555) + monkeypatch.setattr(daemon.ingestors, "ingestor_start_monotonic", lambda: 555) state = _make_state(iface="IFACE", provider="PROV", last_announce=None) assert daemon._process_announcements(state) == 999.0 assert calls == [("PROV", "IFACE", 555, None)] diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py index 00b92624..65786ff2 100644 --- a/tests/test_provider_unit.py +++ b/tests/test_provider_unit.py @@ -1597,6 +1597,11 @@ def _telemetry_env(monkeypatch, *, contacts=None, frozen_time=1_700_000_000): contacts: Optional contact dicts pre-registered on the interface. frozen_time: Wall-clock second ``time.time`` is pinned to. + Transmission is permitted here (SPEC MA7 defaults it off) so these tests + exercise the poll machinery itself; the policy is covered in + ``tests/test_tx_policy_unit.py``, and a test that needs it closed simply + re-patches the flag afterwards. + Returns: Tuple ``(mc_tel, iface, stub, captured)`` — module under test, the interface, the stubbed handlers module, and the captured packet list. @@ -1608,6 +1613,9 @@ def _telemetry_env(monkeypatch, *, contacts=None, frozen_time=1_700_000_000): stub = _make_stub_handlers_module() stub.store_packet_dict = lambda pkt: captured.append(pkt) monkeypatch.setattr(mc_tel.config, "_debug_log", lambda *_a, **_k: None) + monkeypatch.setattr(mc_tel.config, "TX_ENABLED", True) + monkeypatch.setattr(mc_tel.config, "TX_ANNOUNCE", True) + monkeypatch.setattr(mc_tel.config, "RX_ONLY", False) monkeypatch.setattr(_time, "time", lambda: frozen_time) iface = _MeshcoreInterface(target=None) for contact in contacts or []: @@ -2162,8 +2170,96 @@ def test_telemetry_poll_loop_disabled_and_ticking(monkeypatch): assert calls["contact"] >= 1 # first on-air poll after one interval +@pytest.mark.parametrize( + ("closed", "expected_gate"), + [ + ({"TX_ENABLED": False}, "TX_ENABLED"), + ({"RX_ONLY": True}, "RX_ONLY"), + ], +) +def test_poll_contact_telemetry_refuses_closed_policy( + monkeypatch, closed, expected_gate +): + """A closed transmit policy stops the poll *at the request*, not just the loop. + + Covers the gate inside ``_poll_contact_telemetry`` itself: without this the + check could be deleted and every other test would still pass, because the + loop-entry interval check would mask it. + """ + import types + import data.mesh_ingestor.activity as activity + import data.mesh_ingestor.tx_policy as tx_policy + + mc_tel, iface, stub, _captured = _telemetry_env( + monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}] + ) + iface.host_node_id = "!deadbeef" + for name, value in closed.items(): + monkeypatch.setattr(mc_tel.config, name, value) + assert tx_policy.describe_tx_policy()["blocked_by"] == expected_gate + + calls = {"telemetry": 0, "status": 0} + + class _Commands: + async def req_telemetry_sync(self, contact): # pragma: no cover + calls["telemetry"] += 1 + return None + + async def req_status_sync(self, contact): # pragma: no cover + calls["status"] += 1 + return None + + activity.take_packet_count() # drain + asyncio.run( + mc_tel._poll_contact_telemetry( + types.SimpleNamespace(commands=_Commands()), iface, stub, {} + ) + ) + assert calls == {"telemetry": 0, "status": 0} + assert activity.take_packet_count() == 0 + + +def test_poll_contact_telemetry_status_fallback_rechecks_policy(monkeypatch): + """The status fallback is a second transmission and takes its own check. + + A policy that closes between the telemetry pull and the status fallback must + stop the fallback frame; one check covering both requests would let it out. + """ + import types + import data.mesh_ingestor.activity as activity + + mc_tel, iface, stub, _captured = _telemetry_env( + monkeypatch, contacts=[{"public_key": _TEST_CONTACT_KEY, "adv_name": "Sensor"}] + ) + iface.host_node_id = "!deadbeef" + calls = {"telemetry": 0, "status": 0} + + class _Commands: + async def req_telemetry_sync(self, contact): + calls["telemetry"] += 1 + # The operator revokes permission while this request is in flight. + monkeypatch.setattr(mc_tel.config, "TX_ENABLED", False) + return None + + async def req_status_sync(self, contact): # pragma: no cover + calls["status"] += 1 + return None + + activity.take_packet_count() # drain + asyncio.run( + mc_tel._poll_contact_telemetry( + types.SimpleNamespace(commands=_Commands()), iface, stub, {} + ) + ) + assert calls == {"telemetry": 1, "status": 0} + assert activity.take_packet_count() == 1 # only the first request went out + + def test_telemetry_poll_loop_rx_only_disables_on_air_polls(monkeypatch): - """RX_ONLY forbids ingestor TX: contact polls stop, local self reads stay.""" + """The legacy RX_ONLY veto still stops contact polls even with TX_ENABLED=1. + + Local companion-link self reads cost no airtime and continue regardless. + """ import types mc_tel, iface, stub, _captured = _telemetry_env( @@ -5131,7 +5227,7 @@ def test_run_meshcore_autoadd_set_rejected_logs_warning(monkeypatch): # --------------------------------------------------------------------------- -def test_send_channel_announcement_meshtastic_sends_and_counts(): +def test_send_channel_announcement_meshtastic_sends_and_counts(permit_tx): """Meshtastic sends on CHANNEL_INDEX and counts the transmission (MA1).""" import data.mesh_ingestor.activity as activity import data.mesh_ingestor.config as config @@ -5148,7 +5244,63 @@ def test_send_channel_announcement_meshtastic_sends_and_counts(): assert activity.take_packet_count() == 1 -def test_send_channel_announcement_meshtastic_noop_without_sendtext(): +@pytest.mark.parametrize( + "closed", + [ + {"TX_ENABLED": False}, + {"TX_ANNOUNCE": False}, + {"RX_ONLY": True}, + ], +) +def test_send_channel_announcement_meshtastic_refuses_closed_policy( + permit_tx, monkeypatch, closed +): + """The transmit primitive enforces the gates itself, not just its caller. + + ``send_channel_announcement`` is a public duck-typed capability, so a second + caller reaching past the daemon must not be able to put a frame on the air. + """ + import data.mesh_ingestor.activity as activity + + for name, value in closed.items(): + monkeypatch.setattr(permit_tx, name, value) + calls = [] + + class _Iface: + def sendText(self, text, channelIndex=0): + calls.append(text) + + activity.take_packet_count() # drain + MeshtasticProvider().send_channel_announcement(_Iface(), "hello mesh") + assert calls == [] + assert activity.take_packet_count() == 0 + + +def test_send_channel_announcement_meshcore_refuses_closed_policy( + permit_tx, monkeypatch +): + """The MeshCore primitive refuses before touching the event loop.""" + import types as _types + import data.mesh_ingestor.activity as activity + + monkeypatch.setattr(permit_tx, "TX_ENABLED", False) + iface = _MeshcoreInterface(target=None) + sent = [] + + class _Commands: + async def send_chan_msg(self, chan, msg, timestamp=None): # pragma: no cover + sent.append((chan, msg)) + + iface._mc = _types.SimpleNamespace(commands=_Commands()) + iface._loop = None # never reached: the gate returns first + + activity.take_packet_count() # drain + MeshcoreProvider().send_channel_announcement(iface, "hello mesh") + assert sent == [] + assert activity.take_packet_count() == 0 + + +def test_send_channel_announcement_meshtastic_noop_without_sendtext(permit_tx): """An interface lacking sendText is a no-op that counts no TX.""" import data.mesh_ingestor.activity as activity @@ -5157,7 +5309,7 @@ def test_send_channel_announcement_meshtastic_noop_without_sendtext(): assert activity.take_packet_count() == 0 -def test_send_channel_announcement_meshcore_sends_and_counts(): +def test_send_channel_announcement_meshcore_sends_and_counts(permit_tx): """MeshCore schedules send_chan_msg on its loop and counts the TX (MA1).""" import asyncio as _asyncio import threading as _threading @@ -5190,7 +5342,7 @@ def test_send_channel_announcement_meshcore_sends_and_counts(): loop.close() -def test_send_channel_announcement_meshcore_noop_guards(): +def test_send_channel_announcement_meshcore_noop_guards(permit_tx): """No-op (no TX) for a wrong iface type or a missing/closed loop or handle.""" import asyncio as _asyncio import types as _types diff --git a/tests/test_tx_policy_unit.py b/tests/test_tx_policy_unit.py new file mode 100644 index 00000000..4b4ef2ba --- /dev/null +++ b/tests/test_tx_policy_unit.py @@ -0,0 +1,359 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Unit tests for :mod:`data.mesh_ingestor.tx_policy` (SPEC MA7). + +Guards the single transmit-permission predicate and the fail-safe env parsing +behind it. The defect these tests exist for: ``RX_ONLY`` was parsed as an exact +``== "1"`` comparison, so every near-miss spelling (``true``, ``TRUE``, ``yes``, +``" 1"``) resolved to *false* and the receive-only kill switch **failed open** — +the ingestor transmitted despite the operator having forbidden it. The +companion defect: the opt-in flag reached no packaged deployment surface, so it +could not be set at all from `docker-compose`, the Nix module, or the image. +""" + +from __future__ import annotations + +import importlib +import re +import sys +from pathlib import Path + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: + sys.path.insert(0, str(REPO_ROOT)) + +import data.mesh_ingestor.config as config +import data.mesh_ingestor.tx_policy as tx_policy + +#: Every env var that participates in the transmit policy. +_TX_ENV_VARS = ("TX_ENABLED", "TX_ANNOUNCE", "RX_ONLY") + + +@pytest.fixture +def tx_env(monkeypatch): + """Yield a setter that reloads :mod:`config` under a clean TX environment. + + Restores the ambient (unset) environment and reloads once more on teardown so + a reload-based test cannot leak resolved flags into its neighbours. + """ + + def _apply(**env: str): + for name in _TX_ENV_VARS: + monkeypatch.delenv(name, raising=False) + for name, value in env.items(): + monkeypatch.setenv(name, value) + importlib.reload(config) + return config + + yield _apply + + for name in _TX_ENV_VARS: + monkeypatch.delenv(name, raising=False) + importlib.reload(config) + + +# --------------------------------------------------------------------------- +# config._env_flag — the shared fail-safe boolean parser +# --------------------------------------------------------------------------- + + +class TestEnvFlag: + """Tests for :func:`config._env_flag`.""" + + @pytest.mark.parametrize("raw", ["1", "true", "TRUE", "True", "yes", "on", " 1 "]) + def test_truthy_spellings(self, monkeypatch, raw): + """Every accepted spelling of "on" parses as ``True``, whitespace included.""" + monkeypatch.setenv("SOME_FLAG", raw) + assert config._env_flag("SOME_FLAG", default=False, on_invalid=False) is True + + @pytest.mark.parametrize("raw", ["0", "false", "FALSE", "no", "off", " 0 "]) + def test_falsy_spellings(self, monkeypatch, raw): + """Every accepted spelling of "off" parses as ``False``.""" + monkeypatch.setenv("SOME_FLAG", raw) + assert config._env_flag("SOME_FLAG", default=True, on_invalid=True) is False + + @pytest.mark.parametrize("raw", ["", " "]) + def test_blank_falls_back_to_default(self, monkeypatch, raw): + """A blank value is treated as unset so an empty ``.env`` line is inert.""" + monkeypatch.setenv("SOME_FLAG", raw) + assert config._env_flag("SOME_FLAG", default=True, on_invalid=False) is True + + def test_unset_falls_back_to_default(self, monkeypatch): + """An absent variable resolves to the declared default.""" + monkeypatch.delenv("SOME_FLAG", raising=False) + assert config._env_flag("SOME_FLAG", default=True, on_invalid=False) is True + + def test_unrecognized_uses_on_invalid_and_warns(self, monkeypatch, capsys): + """An unparseable value resolves fail-safe and says so, loudly.""" + monkeypatch.setenv("SOME_FLAG", "maybe") + assert config._env_flag("SOME_FLAG", default=True, on_invalid=False) is False + out = capsys.readouterr().out + assert "SOME_FLAG" in out and "maybe" in out + assert "warning" in out.lower() + + def test_unrecognized_warns_even_without_debug(self, monkeypatch, capsys): + """The warning bypasses the ``DEBUG`` guard — it is not a debug line.""" + monkeypatch.setattr(config, "DEBUG", False) + monkeypatch.setenv("SOME_FLAG", "banana") + config._env_flag("SOME_FLAG", default=False, on_invalid=False) + assert "banana" in capsys.readouterr().out + + +# --------------------------------------------------------------------------- +# Env resolution of the three TX flags +# --------------------------------------------------------------------------- + + +class TestTxEnabledEnv: + """Tests for :data:`config.TX_ENABLED`.""" + + def test_defaults_off(self, tx_env): + """Unset means no transmissions — the whole point of the change.""" + assert tx_env().TX_ENABLED is False + + @pytest.mark.parametrize("raw", ["1", "true", "TRUE", "yes", "on", " 1 "]) + def test_enabled_spellings(self, tx_env, raw): + """Any accepted spelling of "on" enables transmission.""" + assert tx_env(TX_ENABLED=raw).TX_ENABLED is True + + @pytest.mark.parametrize("raw", ["0", "false", "no", "off"]) + def test_disabled_spellings(self, tx_env, raw): + """Any accepted spelling of "off" keeps transmission disabled.""" + assert tx_env(TX_ENABLED=raw).TX_ENABLED is False + + def test_unrecognized_fails_safe_to_off(self, tx_env): + """A typo must not silently grant transmit permission.""" + assert tx_env(TX_ENABLED="maybe").TX_ENABLED is False + + +class TestTxAnnounceEnv: + """Tests for :data:`config.TX_ANNOUNCE`.""" + + def test_defaults_off(self, tx_env): + """Announcements are opt-in.""" + assert tx_env().TX_ANNOUNCE is False + + @pytest.mark.parametrize("raw", ["1", "true", "yes", "on", " 1 "]) + def test_enabled_spellings(self, tx_env, raw): + """Any accepted spelling of "on" opts into announcements.""" + assert tx_env(TX_ANNOUNCE=raw).TX_ANNOUNCE is True + + def test_unrecognized_fails_safe_to_off(self, tx_env): + """A typo must not silently opt an operator into unsolicited TX.""" + assert tx_env(TX_ANNOUNCE="sure").TX_ANNOUNCE is False + + +class TestLegacyRxOnlyEnv: + """Tests for the legacy :data:`config.RX_ONLY` kill switch.""" + + def test_defaults_off(self, tx_env): + """Absent legacy flag asserts nothing; the TX_ENABLED default governs.""" + assert tx_env().RX_ONLY is False + + @pytest.mark.parametrize("raw", ["1", "true", "TRUE", "yes", "on", " 1 ", "1 "]) + def test_kill_switch_spellings_all_engage(self, tx_env, raw): + """The regression: every one of these previously failed **open**.""" + assert tx_env(RX_ONLY=raw).RX_ONLY is True + + @pytest.mark.parametrize("raw", ["0", "false", "no", "off"]) + def test_explicitly_disengaged(self, tx_env, raw): + """An explicit "off" does not engage the kill switch.""" + assert tx_env(RX_ONLY=raw).RX_ONLY is False + + def test_unrecognized_fails_safe_to_engaged(self, tx_env): + """A kill switch fails toward *killed*: a typo silences TX, never grants it.""" + assert tx_env(RX_ONLY="probably").RX_ONLY is True + + +# --------------------------------------------------------------------------- +# The single transmit-permission predicate +# --------------------------------------------------------------------------- + + +def _set_policy(monkeypatch, *, tx_enabled, rx_only=False, tx_announce=False): + """Pin the three resolved policy flags on the config module.""" + monkeypatch.setattr(config, "TX_ENABLED", tx_enabled) + monkeypatch.setattr(config, "RX_ONLY", rx_only) + monkeypatch.setattr(config, "TX_ANNOUNCE", tx_announce) + + +class TestTransmitPermitted: + """Truth table for :func:`tx_policy.transmit_permitted`.""" + + @pytest.mark.parametrize( + ("tx_enabled", "rx_only", "expected"), + [ + (False, False, False), # new default: silent + (False, True, False), # legacy kill switch, nothing to override + (True, False, True), # the only permitting combination + (True, True, False), # RX_ONLY keeps its veto over TX_ENABLED=1 + ], + ) + def test_matrix(self, monkeypatch, tx_enabled, rx_only, expected): + """TX is permitted only when opted in and not vetoed by the legacy flag.""" + _set_policy(monkeypatch, tx_enabled=tx_enabled, rx_only=rx_only) + assert tx_policy.transmit_permitted() is expected + + +class TestAnnouncementsPermitted: + """Truth table for :func:`tx_policy.announcements_permitted`.""" + + @pytest.mark.parametrize( + ("tx_enabled", "tx_announce", "rx_only", "expected"), + [ + (False, False, False, False), + (False, True, False, False), # TX_ENABLED=0 beats TX_ANNOUNCE=1 + (True, False, False, False), # transmit allowed, announcing not asked for + (True, True, False, True), # the only permitting combination + (True, True, True, False), # legacy veto still wins + ], + ) + def test_matrix(self, monkeypatch, tx_enabled, tx_announce, rx_only, expected): + """Announcing requires transmit permission *and* the announcement opt-in.""" + _set_policy( + monkeypatch, + tx_enabled=tx_enabled, + rx_only=rx_only, + tx_announce=tx_announce, + ) + assert tx_policy.announcements_permitted() is expected + + def test_implies_transmit_permitted(self, monkeypatch): + """Announcing can never be permitted where transmitting is not.""" + for tx_enabled in (False, True): + for rx_only in (False, True): + _set_policy( + monkeypatch, + tx_enabled=tx_enabled, + rx_only=rx_only, + tx_announce=True, + ) + if tx_policy.announcements_permitted(): + assert tx_policy.transmit_permitted() is True + + +# --------------------------------------------------------------------------- +# Diagnostics: the operator must be able to tell which gate closed +# --------------------------------------------------------------------------- + + +class TestDescribeTxPolicy: + """Tests for :func:`tx_policy.describe_tx_policy`.""" + + def test_names_the_resolved_state(self, monkeypatch): + """The description carries every flag that fed the decision.""" + _set_policy(monkeypatch, tx_enabled=True, rx_only=False, tx_announce=True) + described = tx_policy.describe_tx_policy() + assert described["tx_enabled"] is True + assert described["tx_announce"] is True + assert described["rx_only"] is False + assert described["transmit_permitted"] is True + assert described["announcements_permitted"] is True + + def test_reports_the_blocking_gate(self, monkeypatch): + """When TX is forbidden the description says which flag forbade it.""" + _set_policy(monkeypatch, tx_enabled=True, rx_only=True, tx_announce=True) + assert tx_policy.describe_tx_policy()["blocked_by"] == "RX_ONLY" + + _set_policy(monkeypatch, tx_enabled=False, tx_announce=True) + assert tx_policy.describe_tx_policy()["blocked_by"] == "TX_ENABLED" + + _set_policy(monkeypatch, tx_enabled=True, tx_announce=False) + assert tx_policy.describe_tx_policy()["blocked_by"] == "TX_ANNOUNCE" + + _set_policy(monkeypatch, tx_enabled=True, tx_announce=True) + assert tx_policy.describe_tx_policy()["blocked_by"] is None + + +class TestLogTxPolicy: + """Tests for :func:`tx_policy.log_tx_policy` (startup diagnostics).""" + + def test_logs_resolved_policy_without_debug(self, monkeypatch, capsys): + """The resolved policy is stated at startup even with DEBUG off.""" + monkeypatch.setattr(config, "DEBUG", False) + _set_policy(monkeypatch, tx_enabled=False) + tx_policy.log_tx_policy() + out = capsys.readouterr().out.lower() + assert "tx_enabled=false" in out + assert "transmit_permitted=false" in out + + def test_warns_on_contradictory_configuration(self, monkeypatch, capsys): + """TX_ENABLED=1 with RX_ONLY=1 is contradictory and must be loud.""" + monkeypatch.setattr(config, "DEBUG", False) + _set_policy(monkeypatch, tx_enabled=True, rx_only=True) + tx_policy.log_tx_policy() + out = capsys.readouterr().out.lower() + assert "warning" in out + assert "rx_only" in out + + def test_no_contradiction_warning_when_consistent(self, monkeypatch, capsys): + """A consistent configuration produces no warning.""" + monkeypatch.setattr(config, "DEBUG", False) + _set_policy(monkeypatch, tx_enabled=True, rx_only=False) + tx_policy.log_tx_policy() + assert "warning" not in capsys.readouterr().out.lower() + + +# --------------------------------------------------------------------------- +# The flags must be reachable from the packaged deployment surfaces +# --------------------------------------------------------------------------- + + +class TestDeploymentSurface: + """A knob nobody can set is a knob that does not exist. + + The opt-in shipped with no way to deliver it: the Compose ``environment:`` + mapping is a closed allowlist with no ``env_file:``, and the Nix module and + image ``ENV`` blocks are equally closed. These checks pin every packaged + path so the next TX flag cannot repeat it. + """ + + @pytest.mark.parametrize("name", ["TX_ENABLED", "TX_ANNOUNCE"]) + def test_compose_passes_the_flag_through(self, name): + """The base compose file maps the flag from the host env into the container. + + Only the base file is checked: ``docker-compose.dev.yml`` and + ``.prod.yml`` are overlays (neither is a valid standalone project) and + Compose merges the base ``environment:`` mapping into them, so the + passthrough is inherited rather than repeated. + """ + text = (REPO_ROOT / "docker-compose.yml").read_text(encoding="utf-8") + assert re.search( + rf"^\s*{name}:\s*\$\{{{name}", text, re.MULTILINE + ), f"docker-compose.yml does not pass {name} through to the ingestor" + + @pytest.mark.parametrize("name", ["TX_ENABLED", "TX_ANNOUNCE"]) + def test_env_example_documents_the_flag(self, name): + """The copy-this-to-.env template mentions the flag.""" + text = (REPO_ROOT / ".env.example").read_text(encoding="utf-8") + assert re.search(rf"^{name}=", text, re.MULTILINE) + + @pytest.mark.parametrize("name", ["TX_ENABLED", "TX_ANNOUNCE"]) + def test_image_declares_the_default(self, name): + """The ingestor image declares the flag so its default is visible.""" + text = (REPO_ROOT / "data" / "Dockerfile").read_text(encoding="utf-8") + assert f"{name}=0" in text + + @pytest.mark.parametrize("name", ["txEnabled", "txAnnounce"]) + def test_nix_module_exposes_the_option(self, name): + """The Nix module exposes the flag as a declarative option.""" + text = (REPO_ROOT / "flake.nix").read_text(encoding="utf-8") + assert name in text + + def test_legacy_flag_is_not_reintroduced_to_docs(self): + """``RX_ONLY`` stays supported in code but out of operator-facing docs.""" + readme = (REPO_ROOT / "README.md").read_text(encoding="utf-8") + assert "RX_ONLY" not in readme