Compare commits

...

26 Commits

Author SHA1 Message Date
BigBodyCobain 79c6b852c3 fix(desktop-build): strip layout.tsx force-dynamic on CRLF checkouts too
build-frontend-export.cjs stages a desktop-only frontend export tree and
strips the ``force-dynamic`` + ``revalidate`` directives from
``frontend/src/app/layout.tsx`` so Next's ``output: "export"`` can
prerender every route.

The strip regexes only matched LF (``\n``). Any Windows checkout without
``core.autocrlf=input`` has CRLF line endings, the strip silently
no-op'd, and the desktop build failed at the static-export step:

    Error: Page with `dynamic = "force-dynamic"` couldn't be exported.
    `output: "export"` requires all pages be renderable statically
    because there is no runtime server to dynamically render routes
    in this output format.
    Export encountered an error on /_not-found/page: /_not-found

Reaches every Windows contributor who hasn't normalized line endings
locally. Replacing each ``\n`` in the strip regexes with ``\r?\n``
makes the strip CRLF-tolerant; LF behavior is unchanged.

Verified by running both regexes against the actual layout.tsx (302
bytes removed, force-dynamic + revalidate both gone) and against a
synthetic LF input (296 bytes removed, same outcome).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 16:02:12 -06:00
Shadowbroker dcea325fba Merge pull request #317 from BigBodyCobain/feat/cumulative-fuel-burn
feat(flights): cumulative fuel burned + CO2 emitted per flight
2026-05-23 08:09:34 -06:00
BigBodyCobain 03b8053617 feat(flights): cumulative fuel burned + CO2 emitted per flight
Pre-fix the emissions tooltip only showed the per-hour *rate* — what most
users actually want is the cumulative *amount* burned. This adds running
totals computed by multiplying the model-based rate by the elapsed
observation time since we first saw the airframe.

New module ``flight_observations.py``:
* Tracks first_seen_at + last_seen_at per icao24 hex.
* Re-opens a fresh session when an aircraft is unseen for > 15 min
  (treated as a new flight — landed and took off, or transited a dead
  zone). Prevents the cumulative counter from resetting mid-flight if
  the trail-rendering cache prunes the trail.
* Clamps elapsed time to 24h max so clock skew can't produce comically
  large numbers.
* Pruned every 5 min via a new scheduler job (mirrors ais_prune cadence).

flights.py + military.py emission enrichment now also attaches:
* observed_seconds — how long we've been tracking this airframe.
* fuel_gallons_burned — rate * elapsed_h.
* co2_kg_emitted — rate * elapsed_h.

The existing per-hour rate fields stay in the dict for backward compat
and are shown as small secondary context in the tooltip.

Frontend EmissionsEstimateBlock (NewsFeed.tsx) now prominently shows
the cumulative totals with the rate as smaller context underneath plus
"Observed in flight for Xh Ym". When observed_seconds is 0 (first refresh)
it renders "Just observed · totals will appear on next refresh" instead
of a misleading "0 gal".

12 backend tests cover record/accumulate/reset, the 24h clamp, prune,
case-insensitive key normalization, and end-to-end emission integration
in _classify_and_publish.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 07:56:23 -06:00
Shadowbroker 20807a2d62 Merge pull request #316 from BigBodyCobain/feat/aishub-fallback
feat(ais): AISHub REST fallback when AISStream is offline (20-min polling)
2026-05-23 07:42:56 -06:00
Shadowbroker 79fbf9741b Merge pull request #314 from BigBodyCobain/feat/ais-upstream-health
feat(ais): surface AISStream upstream outage instead of failing silently
2026-05-23 07:12:37 -06:00
BigBodyCobain a2f5d62926 feat(ais): AISHub REST fallback when AISStream WebSocket is offline
When stream.aisstream.io is unreachable (cert outage, server down — see
2026-05-20 and 2026-05-23 events) the ships layer goes empty. This adds
a slow REST fallback to data.aishub.net so the layer stays populated in
degraded mode.

Behavior:

* Opt-in via AISHUB_USERNAME (free registration at aishub.net/api).
  Without the env var the fetcher is a no-op.
* Default poll cadence 20 min — well inside their free-tier limits, gives
  ships time to move enough to look "alive". Configurable via
  AISHUB_POLL_INTERVAL_MINUTES, clamped to [1, 360].
* Internal gate: skips the poll entirely when the WebSocket primary is
  currently connected. Stomping fresh live data with 20-min-old REST
  data would be worse than leaving it alone.
* Vessels merge into the shared _vessels dict with source="aishub" so
  the existing UI / health tooling can attribute the provider.
* Live data wins races: if a WebSocket update for the same MMSI lands in
  the last 1s, we don't overwrite with the slower REST record.

Scheduler job runs every AISHUB_POLL_INTERVAL_MINUTES minutes alongside
the existing ais_prune job in data_fetcher.py.

24 tests cover gating (no-username, primary-connected), response parsing
(success / error / empty / malformed / unexpected shape), record
normalization (sentinels, missing fields, range checks, AIS @ padding),
poll interval clamping, and end-to-end merge with live-data-wins.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 07:00:32 -06:00
BigBodyCobain 5e0b2c037e feat(ais): surface upstream outage instead of failing silently
On 2026-05-23, stream.aisstream.io went fully offline (TCP timeouts on port
443). The backend kept respawning the node WebSocket proxy every few
seconds with nothing arriving. From the operator's POV the ships layer
silently went empty — no banner, no log surfacing, no way to tell whether
it was their config / network / viewport filter / upstream.

Backend:
* ais_proxy_status() now also returns:
  - connected (bool): true when a vessel message arrived in last 60s
  - last_msg_age_seconds (int | None)
  - proxy_spawn_count (int): proxy respawns — sustained growth without
    connected means upstream is dead
* /api/health escalates top status to "degraded" when AIS_API_KEY is set
  but the proxy is currently disconnected. Existing degraded_tls signal
  preserved.

Frontend:
* useAisUpstreamHealth hook polls /api/health every 30s, derives the
  outage state. Defensively only reports outage once spawn_count > 0 so
  operators who haven't opted in don't see the banner.
* AisUpstreamBanner component renders a dismissible amber notice
  "Ship data temporarily unavailable — AISStream upstream is offline"
  mounted on the main app shell.

7 backend tests pin the status-shape contract and the /api/health
escalation behavior in both with-key and without-key configurations.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 06:38:05 -06:00
Shadowbroker 69ef231e5a Merge pull request #313 from BigBodyCobain/feat/flight-source-attribution
feat(flights): stamp source attribution on every flight record
2026-05-23 06:29:31 -06:00
Shadowbroker 7a5f47ca9e Merge pull request #312 from BigBodyCobain/fix/gps-jamming-thresholds
fix(gps-jamming): count nac_p=0 + lower thresholds so layer actually fires
2026-05-23 06:29:20 -06:00
Shadowbroker 5cd49542bf Merge pull request #311 from BigBodyCobain/fix/uap-fallback-cutoff
fix(uap): stop HF fallback from serving 3-year-old NUFORC sightings
2026-05-23 06:29:08 -06:00
BigBodyCobain f14d4feb6d feat(flights): stamp source attribution on every flight record
Pre-fix, adsb.lol records (the primary source for most flights) carried
no source marker. OpenSky records got is_opensky: True and supplementals
got supplemental_source, so any UI inspecting source labels saw
OpenSky/airplanes.live records as explicitly tagged and adsb.lol records
as "unlabeled" — making it look like adsb.lol wasn't being used at all
even though it's the primary source.

Changes:

* _fetch_adsb_lol_regions stamps source="adsb.lol" on each aircraft
  before returning, so the tag survives the OpenSky dedupe-by-hex merge.
* OpenSky records get source="OpenSky" (alongside is_opensky=True for
  back-compat).
* military fetcher tags source on both adsb.lol and airplanes.live
  records before they're merged, and propagates source into the
  military_flights and uavs output dicts.
* _classify_and_publish promotes the explicit source field into the
  published flight dict. Falls back to legacy supplemental_source if
  source is absent. Final fallback "adsb.lol" preserves prior behavior
  for any caller synthesizing records without going through a fetcher.

8 new tests cover the published-dict propagation, OpenSky tagging,
supplemental fallback, explicit-wins precedence, default behavior, the
adsb.lol regional fetcher tagging, and the military output dict.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-23 06:14:39 -06:00
BigBodyCobain 19a8560a80 fix(gps-jamming): count nac_p=0 + lower thresholds so the layer actually fires
Three stacked filters meant the gps_jamming layer almost never lit up:

1. nac_p == 0 aircraft were dropped on the theory that "0 = old transponder."
   That's only half right — modern Mode-S Enhanced Surveillance transponders
   also fall back to nac_p=0 when they lose GPS lock entirely, which IS the
   jamming signature we want to catch. Discarding them was discarding the
   strongest signal. None (no field at all — typical for OpenSky-sourced
   records) is still skipped because absence-of-data isn't evidence.
2. GPS_JAMMING_MIN_AIRCRAFT was 5 per 1°x1° cell. Jamming hotspots
   (eastern Med, Russia/Ukraine border, Iran/Iraq) tend to have sparser
   traffic because pilots avoid them. Lowered to 3.
3. GPS_JAMMING_MIN_RATIO was 0.30. Combined with the (preserved) -1 noise
   cushion that made the effective bar high. Lowered to 0.20.

The 1-aircraft noise cushion is intact so a single quirky transponder
still can't flag a zone alone.

Also extracted the detector loop into a pure ``detect_gps_jamming_zones()``
function at module scope so it's testable in isolation (was previously
inlined inside ``_classify_and_publish``). The public signature accepts
threshold overrides for ad-hoc re-tuning without code edits.

16 new tests cover nac_p=0 inclusion, None-skip preservation, MIN_AIRCRAFT
lowering, MIN_RATIO lowering, noise cushion preservation, constant pinning,
override behavior, lon/lng key compatibility, and robustness to empty/None
inputs.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 23:40:18 -06:00
BigBodyCobain 0d0e009867 fix(uap): stop HF fallback from serving 3-year-old NUFORC sightings
The UAP sightings layer is sourced from a live scrape of nuforc.org with a
static Hugging Face CSV mirror (kcimc/NUFORC) as a fallback. The fallback
parsed every row, sorted by occurred-desc, and took the top 250 — with no
date cutoff. The HF mirror is a third-party snapshot that hasn't been
refreshed in years, so the "newest 250" rows it returns are from ~2022-23.
When the live path fails (Cloudflare 403, curl disabled on Windows, wdtNonce
regex stale, etc.) users see a map full of sightings from 3 years ago,
labeled as the "last 60 days" layer.

Changes:

* HF fallback now applies the same 60-day cutoff the live path uses. Rows
  outside the window are dropped before take-top-N. If the mirror has
  nothing inside the window the fallback returns [] (don't serve stale).
* When the HF mirror is fully stale a loud ERROR log fires with the count
  of dropped rows so the operator can tell the mirror's the problem, not
  a network issue.
* When BOTH live AND HF fallback produce 0 rows, fetch_uap_sightings now
  trips assert_canary("uap_sightings", 0) so the health registry shows
  the layer as broken instead of "fresh and empty for days."
* Scheduler moved from daily 12:00 UTC to weekly Mondays 12:00 UTC. The
  layer is a rolling 60-day digest; refreshing once a week is enough
  cadence for human-readable map exploration and keeps nuforc.org load
  light.

6 new tests cover the cutoff filter, the doomsday-log path, the mixed-age
path, the both-paths-empty health failure, the positive fallback path, and
the scheduler cadence.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 23:27:12 -06:00
Shadowbroker febcce9125 Merge pull request #310 from BigBodyCobain/fix/infonet-sync-429-backoff
Infonet sync: honor HTTP 429 Retry-After + exponential backoff
2026-05-22 23:11:00 -06:00
BigBodyCobain 31ebcb5cd9 Infonet sync: honor HTTP 429 Retry-After + exponential backoff
Fixes the retry-storm that's been keeping the local node 429'd out of
the seed peer (the diagnosis we ran earlier in the session). Pre-fix:

  1. Sync hits the seed peer, gets HTTP 429 (Too Many Requests)
  2. _peer_sync_response stringifies the status into a ValueError
  3. _sync_from_peer catches it, error becomes the str() of the exc
  4. _run_public_sync_cycle calls finish_sync(error=..., failure_backoff_s=60)
  5. next_sync_due_at = now + 60s
  6. After 60s, sync runs again, hits same upstream that hasn't reset
     its rate-limit bucket, 429 again. Loop indefinitely.

Net effect: a node that hit one transient 429 would hammer the seed
every 60s forever, keeping the bucket full and never recovering. We
saw this in the live status dump: consecutive_failures=49,
last_sync_ok_at=0, retry storm sustained over the entire uptime.

What changed
------------
services/mesh/mesh_infonet_sync_support.py

  * New typed exception PeerSyncRateLimited carries the parsed
    Retry-After value out of the HTTP layer instead of stringifying
    everything into a generic ValueError.

  * New parse_retry_after_header() handles both RFC 7231 §7.1.3
    forms (delay-seconds and HTTP-date). Clamped at 1 hour so a
    hostile peer can't silence us for days.

  * New _failure_backoff_seconds() helper computes the next delay
    as max(exponential, retry_after_s). Schedule with default
    base=60s, cap=1800s:

      failure 1 -> 60s     (preserves pre-fix for transient blips)
      failure 2 -> 120s
      failure 3 -> 240s
      failure 4 -> 480s
      failure 5 -> 960s
      failure 6+ -> 1800s  (capped at 30 min)

    cap_s=0 explicitly disables exponential entirely — operators
    who want pure-Retry-After behavior have that option.

  * finish_sync now accepts retry_after_s and failure_backoff_cap_s
    kwargs. Backward-compatible: existing callers that don't pass
    retry_after_s get the same first-failure delay as before (the
    base value), only repeat failures grow.

main.py

  * _peer_sync_response detects 429 specifically, parses the
    Retry-After header, raises PeerSyncRateLimited(retry_after_s=N).
    Includes the response body prefix in the message so the
    operator's last_error finally shows something useful.

  * _sync_from_peer extended to return (ok, error, forked,
    retry_after_s) — the 4th tuple element is non-zero only when
    the upstream sent a parseable Retry-After. Existing call shape
    preserved: the lone caller in _run_public_sync_cycle was
    updated in the same commit.

  * _run_public_sync_cycle forwards retry_after_s into finish_sync.

Tests
-----
backend/tests/mesh/test_infonet_sync_429_backoff.py — 17 new tests:

  TestParseRetryAfter (7):
    - integer seconds form
    - HTTP-date form (computed as seconds-from-now)
    - HTTP-date in the past returns 0
    - empty / whitespace returns 0
    - malformed returns 0
    - clamps to 1 hour (hostile-peer cap)
    - negative returns 0

  TestFailureBackoffSeconds (5):
    - exponential growth schedule pins each level
    - retry_after wins when larger than exponential
    - exponential wins when larger than retry_after
    - cap_s=0 disables exponential entirely
    - zero inputs return zero

  TestFinishSyncBackoff (5):
    - first failure uses base unchanged (pre-fix back-compat)
    - consecutive_failures actually grow the delay
    - retry_after honored at low failure count
    - success resets consecutive_failures
    - last_error carries the HTTP status / Retry-After detail

All 24 existing sync-support / status-gate tests still pass. Other
failures in tests/mesh/ are pre-existing on origin/main and unrelated
to this change (verified by running the same tests against the
user's main worktree without these edits).

What the operator sees after this lands + a docker rebuild
----------------------------------------------------------
With the live 429 storm we diagnosed:

  Pre-fix: consecutive_failures keeps climbing 1/min forever,
           last_error empty or generic
  Post-fix: consecutive_failures grows, next_sync_due_at backs off
           exponentially (max 30 min), last_error explicitly carries
           "HTTP 429 from <peer> (retry_after=Ns): <body>" so the
           operator can see what's actually wrong. Once the upstream
           bucket drains and a sync succeeds, consecutive_failures
           resets to 0 and the schedule returns to the normal 300s
           interval.
2026-05-22 22:55:05 -06:00
Shadowbroker b3fca3dc18 Merge pull request #309 from BigBodyCobain/feat/cross-node-dm-mailbox-replication
DM mailbox: per-(sender, recipient) anti-spam cap + replication primitives
2026-05-22 22:43:26 -06:00
BigBodyCobain 401f114e4f DM mailbox: outbound replication + receiving endpoint
Second commit on this branch (first added the per-sender cap + accept_replica
primitive). This commit wires the actual cross-node propagation:

Outbound (sender side)
----------------------
* New ``DMRelay._replicate_envelope_to_peers_async()`` — fire-and-forget
  thread that POSTs the envelope to every authenticated relay peer via
  the same per-peer HMAC pattern gate-message replication uses (#256
  ``X-Peer-Url`` + ``X-Peer-HMAC`` headers, ``resolve_peer_key_for_url``).
* ``deposit()`` now calls the replication helper after a successful
  local accept. Per-peer errors are swallowed — slow Tor peers must not
  block the sender's UX, and the recipient polling from a healthy peer
  works fine even if some peers are down.
* Metrics: dm_replication_push_ok / _rejected / _error.

Inbound (receiving side)
------------------------
* New endpoint ``POST /api/mesh/dm/replicate-envelope`` in
  routers/mesh_peer_sync.py.
* Same HMAC auth gate (``_verify_peer_push_hmac``) as the existing
  infonet/gate peer-push endpoints. Unauthenticated requests get 403.
* Body cap of 64 KB (DM envelope is bounded by MESH_DM_MAX_MSG_BYTES).
* Calls DMRelay.accept_replica which enforces the per-sender cap as a
  network rule — hostile sender's relay can hold extras locally but
  honest peers reject them on inbound replication.

End-to-end flow now works
-------------------------
  1. Alice's node accepts a deposit to Bob's mailbox (local cap check).
  2. Alice's node spawns a background thread that POSTs the envelope
     to MESH_RELAY_PEERS with per-peer HMAC.
  3. Each peer's /api/mesh/dm/replicate-envelope verifies the HMAC and
     calls accept_replica, which re-enforces the per-sender cap.
  4. Bob (offline at the time of send) eventually logs into ANY node
     in MESH_RELAY_PEERS, his existing pollDmMailboxes pulls from
     the local mailbox there, finds Alice's envelope, decrypts.

Tests
-----
backend/tests/test_dm_replicate_envelope_endpoint.py — 4 tests:

  TestReplicateEndpointAuth:
    - rejects requests without peer HMAC (403)
    - rejects requests with WRONG peer HMAC (403) — confirms the
      HMAC is actually verified, not just present
    - rejects oversize bodies (>64 KB) with 400/413

  TestReplicateEndpointRegistered:
    - static check that POST /api/mesh/dm/replicate-envelope is
      registered on app.routes — catches future refactor that
      drops the router include

All 38 backend tests touching the new code paths still pass:
  test_dm_relay_per_sender_cap.py (14)
  test_dm_replicate_envelope_endpoint.py (4)
  test_no_new_duplicate_routes.py (1) — new route is unique
  test_per_peer_secret_resolver.py (19) — HMAC primitive unaffected

What's still ahead (PR-3+)
--------------------------
* ack propagation: when recipient pulls a message on node X, peers Y/Z
  should prune their copies to free the sender's quota network-wide.
  Without this, the sender's quota frees only on the node the recipient
  actually polled — other peers still see N pending until TTL expiry.
  Workable but suboptimal. PR-3 will add a /api/mesh/dm/ack endpoint
  with the same HMAC pattern.
* recipient pull-from-peers: today the recipient's poll only hits
  their own node's relay. If they log into a peer they didn't deposit
  with, they need a way to fetch envelopes from other peers in
  MESH_RELAY_PEERS. Today this works as long as the recipient's
  current node is one of the peers Alice's node pushed to — which is
  true in a fully-meshed deployment but not guaranteed for partial
  meshes. PR-4 if telemetry shows this matters.
2026-05-22 19:23:09 -06:00
BigBodyCobain 79b39e8985 DM mailbox: per-(sender, recipient) anti-spam cap + replication primitives
Foundation work for cross-node DM mailbox replication. Adds the network
rule that makes the replication safe to ship next, plus the primitives
the outbound replication PR will call.

The rule
--------
A single sender can have at most N UNACKED messages parked in a single
recipient's mailbox at any one time. Default N=2, tunable via
``MESH_DM_PENDING_PER_SENDER_LIMIT``. Once the recipient pulls (acks) a
message, the sender's quota for that (sender, recipient) pair frees up.

Network rule, not local rule
----------------------------
The cap is enforced TWICE:

  1. ``DMRelay.deposit(...)`` — local check on the sender's own node.
     Refuses to spool the (N+1)th message before it can be replicated.

  2. ``DMRelay.accept_replica(...)`` — replication-acceptance check on
     every receiving peer. Refuses to accept an inbound replica that
     would put the local mailbox over the cap.

The second half is what makes the rule a NETWORK rule. A hostile sender
could patch out the deposit check on their own relay and continue to
spool extras locally — but those extras can never propagate, because
every honest peer enforces the same cap on the way in. A recipient who
polls from honest peers therefore never sees more than N pending from
any one sender, regardless of how many spam attempts the hostile
sender's relay accepted.

New API surface on ``DMRelay``
------------------------------
  _per_sender_pending_limit()       — reads MESH_DM_PENDING_PER_SENDER_LIMIT
  _per_sender_pending_count(...)    — counts unacked from a sender for a mailbox
  accept_replica(envelope=...)      — peer-push receive entry point
  envelope_for_replication(...)     — helper to extract a wire-form envelope

``accept_replica`` is idempotent on duplicate ``msg_id`` (replication
round-trips and multi-path delivery don't double-spool).

``envelope_for_replication`` exposes the exact shape ``accept_replica``
expects, so the follow-up PR (outbound replication wiring) just has to
fetch the envelope and POST it to authenticated peer URLs with the
existing per-peer HMAC pattern from #256.

Why this is PR-1 of two
-----------------------
The full cross-node mailbox replication needs three pieces:

  A. cap enforcement on deposit (in this PR)
  B. cap enforcement on replica acceptance (in this PR)
  C. outbound: push envelope to MESH_RELAY_PEERS after deposit (NEXT PR)

(A) + (B) shipped together close the cap-bypass attack surface BEFORE
(C) introduces the actual cross-node propagation. Shipping them in the
other order would briefly let extras propagate during the window between
"outbound push lands" and "accept_replica cap lands."

Tests
-----
backend/tests/test_dm_relay_per_sender_cap.py — 14 tests:

  TestDepositCap:
    - first 2 deposits succeed (UX baseline)
    - 3rd from same sender rejected with friendly message
    - different senders have independent quotas
    - different recipients have independent quotas
    - ack frees the quota (after recipient pulls, sender can deposit again)
    - cap is env-tunable

  TestAcceptReplicaCap:
    - replica accepted under cap
    - idempotent on duplicate msg_id (no double-spool, no rejection)
    - rejected at cap with structured ``cap_violation`` marker so
      sender's relay can stop retrying
    - per-sender, not per-mailbox: different sender_block_ref passes
      even when another sender at the same mailbox is capped
    - malformed envelope shapes rejected without crash

  TestEnvelopeForReplication:
    - returns the envelope for stored messages
    - returns None for unknown msg_id
    - round-trips through accept_replica end-to-end (proves the wire
      shape matches across the two sides)
2026-05-22 19:18:01 -06:00
Shadowbroker c3e38621fc Merge pull request #308 from BigBodyCobain/fix/296-windows-venv-uvicorn-detection
Fix #296: reject backend venvs missing uvicorn before launch (Windows)
2026-05-22 18:56:08 -06:00
Shadowbroker ba39d3b9aa Merge pull request #307 from BigBodyCobain/fix/302-openclaw-hmac-reveal-hardening
Fix #302: split OpenClaw HMAC reveal into dedicated POST with no-store headers
2026-05-22 18:47:09 -06:00
BigBodyCobain f91ddcf38b Fix #302: split OpenClaw HMAC reveal into dedicated POST with no-store
Reported by @tg12. Pre-fix, two problems lived on the GET endpoint:

  1. `GET /api/ai/connect-info?reveal=true` returned the full HMAC
     secret in the response body on every Connect modal open. Even
     gated to require_local_operator, that put the secret into
     browser history, dev-tools network panels, browser disk caches,
     HAR exports, and screen captures.

  2. The same GET endpoint auto-bootstrapped (generated + persisted)
     the secret on a mere read. Side effects on a GET are a footgun:
     browser prefetchers, mirror tools, and casual curl-from-history
     would all silently mint+persist a fresh secret.

Backend (backend/routers/ai_intel.py)
-------------------------------------
  GET  /api/ai/connect-info             — always returns the MASKED
                                          fingerprint (first6 + bullets
                                          + last4). No `?reveal` param.
                                          NO auto-bootstrap. When the
                                          secret is missing, returns
                                          `hmac_secret_set: false` and
                                          tells the caller to POST to
                                          /bootstrap.
  POST /api/ai/connect-info/bootstrap   — NEW. Mints+persists the secret
                                          if missing. Idempotent. Never
                                          returns the full secret in the
                                          response body.
  POST /api/ai/connect-info/reveal      — NEW. Returns the full secret
                                          with Cache-Control: no-store,
                                          no-cache, must-revalidate +
                                          Pragma: no-cache + Expires: 0.
                                          POST so the body never lands
                                          in URL history. 404 (with a
                                          pointer to /bootstrap) when
                                          the secret isn't set.
  POST /api/ai/connect-info/regenerate  — keeps existing one-time-reveal
                                          behavior (regen IS a deliberate
                                          destructive action triggered
                                          by the operator). Same
                                          no-store/no-cache headers added
                                          so even the regen response
                                          doesn't get cached.

Frontend (AIIntelPanel.tsx, OnboardingModal.tsx)
------------------------------------------------
  * On mount: GET (masked only). If hmac_secret_set: false, fire a
    transparent POST /bootstrap and refresh the masked fingerprint.
    Operator sees no behavior change from pre-#302.
  * Reveal (eye icon): lazy POST /reveal — secret only travels when
    the operator explicitly clicks the button.
  * Copy: lazy POST /reveal too — copying without a prior reveal
    works exactly like before, just routed through the new endpoint.
  * Regenerate: POST returns the new secret (same as before, but the
    response now has no-store headers).
  * The displayed snippet uses the masked fingerprint until the
    operator clicks Reveal or Copy.

Tests (backend/tests/test_openclaw_connect_info_reveal.py — 13 tests)
---------------------------------------------------------------------
  * GET returns masked + the full secret never appears in r.text
  * GET does NOT auto-bootstrap when missing
  * GET silently ignores any ?reveal=true query (back-compat noise)
  * POST /bootstrap mints when missing, idempotent when set
  * POST /bootstrap never returns the full secret
  * POST /reveal returns the full secret with Cache-Control: no-store,
    no-cache + Pragma: no-cache + Expires: 0
  * POST /reveal 404s with a pointer to /bootstrap when no secret
  * POST /regenerate returns the new secret with the same headers
  * Anonymous remote callers get 403 on ALL FOUR endpoints (parametric
    regression against the same allowlist used elsewhere).

Adjacent suites still green: test_openclaw_route_security,
test_no_new_duplicate_routes, test_control_surface_auth. 67/67 pass
locally.

Credit: @tg12 for the audit report.
2026-05-22 18:40:24 -06:00
Shadowbroker 49151d8b9f Merge pull request #304 from BigBodyCobain/fix/298-sentinel-creds-server-side
Fix #298: move Sentinel credentials from browser storage to backend .env
2026-05-22 18:29:11 -06:00
BigBodyCobain 767a2f6c00 Merge remote-tracking branch 'origin/main' into fix/298-sentinel-creds-server-side 2026-05-22 18:19:12 -06:00
BigBodyCobain 7bfaad17f0 Merge remote-tracking branch 'origin/main' into fix/298-sentinel-creds-server-side 2026-05-22 17:55:58 -06:00
BigBodyCobain 32b8421a1c Merge origin/main into fix/298: resolve tools.py conflict
PR #303 landed on main and added Depends(require_local_operator) to the
@router.post decorators for /api/sentinel/token and /api/sentinel/tile.
PR #298 (this branch) edited the same decorator lines AND function bodies
to add the env-credential fallback resolver.

Resolution keeps BOTH:
  * The require_local_operator dependency from #303 (the auth gate)
  * The _resolve_sentinel_credentials helper from #298
  * The env-fallback path inside the function bodies

Both layers are independent — the gate blocks anonymous callers, the env
fallback lets legitimate (gated) callers omit credentials from the body.

Verified: 46 tests pass against the merged code, including both
test_sentinel_credentials_server_side.py (#298 fallback) and
test_sentinel_routes_auth_gate.py (#303 gate).
2026-05-22 17:52:10 -06:00
BigBodyCobain b041b5e97c Fix #298: move Sentinel credentials from browser storage to backend .env
Reported by @tg12. Pre-fix, the Settings panel stored real third-party
Copernicus CDSE client_id + client_secret in browser localStorage /
sessionStorage via the privacy storage helper, and the proxy routes
required those values to come back in every tile/token request body.
Any same-origin script (XSS, malicious browser extension, dev-tools
HAR export) had read access to the credentials.

This change moves them server-side, behind the same .env-backed admin
flow every other third-party API key (OpenSky, AIS Stream, Finnhub,
Shodan, …) already uses.

Backend
-------
backend/services/api_settings.py
  * Added SENTINEL_CLIENT_ID and SENTINEL_CLIENT_SECRET entries to
    API_REGISTRY. The existing GET/PUT /api/settings/api-keys flow
    (already require_local_operator-gated, .env-backed) now manages
    them — no new route surface.

backend/routers/tools.py
  * /api/sentinel/token and /api/sentinel/tile resolve credentials via
    a new _resolve_sentinel_credentials() helper: body fields win for
    back-compat with any legacy callers, otherwise the helper reads
    SENTINEL_CLIENT_ID / SENTINEL_CLIENT_SECRET from os.environ.
  * When neither source has a value, the route returns 400 with a
    friendly pointer ("Set SENTINEL_CLIENT_ID and SENTINEL_CLIENT_SECRET
    in the API Keys panel") instead of the curt "required" message.
    The user's standing rule against hostile errors applies.
  * Function bodies only — decorator lines untouched, so this PR does
    not conflict with #303 (which adds Depends(require_local_operator)
    to the same routes).

Frontend
--------
frontend/src/lib/sentinelHub.ts — rewritten
  * Removed: getSentinelCredentials / setSentinelCredentials /
    clearSentinelCredentials / getSentinelCredentialStorageMode.
    These were the browser-storage read/write helpers; their existence
    was the bug.
  * Added: checkBackendSentinelStatus(), refreshSentinelStatus(),
    getCachedSentinelStatus(), and a kept-for-back-compat
    hasSentinelCredentials() shim. Status is sourced from
    /api/settings/api-keys (the same endpoint the API Keys panel
    already uses), so we don't add a new route just for this read.
  * Added: migrateLegacySentinelBrowserKeys() — one-shot, idempotent
    helper that clears sb_sentinel_client_id / _secret / _instance_id
    from BOTH localStorage and sessionStorage. We deliberately do NOT
    auto-POST those legacy browser values to the backend; doing so
    would silently migrate a secret across a trust boundary without
    operator consent. Operators re-enter once in the API Keys panel
    and the legacy keys get wiped here.
  * fetchSentinelTile and getSentinelToken no longer send client_id /
    client_secret in the request body. The backend uses .env.

frontend/src/components/SettingsPanel.tsx
  * Dropped sb_sentinel_client_id / _secret / _instance_id from
    PRIVACY_SENSITIVE_BROWSER_KEYS — they're no longer written.
  * SentinelTab rewritten: removed the inline Client ID / Client Secret
    inputs + Save / Clear / Test buttons. Replaced with a status panel
    that calls checkBackendSentinelStatus() on mount, a one-click
    "Open API Keys Panel" button, and a migration banner that appears
    only when migrateLegacySentinelBrowserKeys() actually cleared
    something.
  * Setup guide STEP 3 now points to the API Keys panel instead of
    the local form.

frontend/src/app/page.tsx
  * Added a one-time useEffect that fires checkBackendSentinelStatus()
    on mount so the cached value (which the synchronous
    hasSentinelCredentials() shim reads) is populated before
    MaplibreViewer's tile-URL memo runs.

Tests
-----
backend/tests/test_sentinel_credentials_server_side.py (new)
  * API_REGISTRY surface — sentinel_client_id / sentinel_client_secret
    are registered with the right env_keys, ALLOWED_ENV_KEYS lets
    /api/settings/api-keys PUT them.
  * Resolution order — body wins, env is fallback, neither → 400 with
    the friendly pointer message, and NO upstream HTTP call when
    neither source has credentials (asserted via
    MagicMock(side_effect=AssertionError)).
  * /api/sentinel/tile same shape.

frontend/src/__tests__/utils/sentinelHub.test.ts (new)
  * migrateLegacySentinelBrowserKeys clears localStorage AND
    sessionStorage, reports what it cleared, idempotent.
  * fetchSentinelTile + getSentinelToken POST WITHOUT client_id /
    client_secret in the body (plants leaked credentials in browser
    storage first to prove they are NOT picked up).
  * checkBackendSentinelStatus parses /api/settings/api-keys correctly:
    true only when both keys is_set, false on partial config or
    network errors.

All 7 backend tests + 8 frontend tests pass locally. The
test_no_new_duplicate_routes guard and the api-settings test suite
still pass.

Credit: @tg12 for the audit report.
2026-05-22 10:44:50 -06:00
39 changed files with 5446 additions and 367 deletions
+7
View File
@@ -11,6 +11,13 @@ AIS_API_KEY= # https://aisstream.io/ — free tier WebSocket key
# ── Optional ───────────────────────────────────────────────────
# AISHub REST fallback. Used when stream.aisstream.io is unreachable
# (e.g. their cert expires or server goes offline). Free tier requires
# registration at https://www.aishub.net/api. Poll cadence defaults to
# 20 min to stay courteous; tunable via AISHUB_POLL_INTERVAL_MINUTES.
# AISHUB_USERNAME=
# AISHUB_POLL_INTERVAL_MINUTES=20
# Override allowed CORS origins (comma-separated). Defaults to localhost + LAN auto-detect.
# CORS_ORIGINS=http://192.168.1.50:3000,https://my-domain.com
+59 -10
View File
@@ -1417,6 +1417,29 @@ def _peer_sync_response(peer_url: str, body: dict[str, Any]) -> dict[str, Any]:
proxy = f"socks5h://127.0.0.1:{socks_port}"
kwargs["proxies"] = {"http": proxy, "https": proxy}
response = _requests.post(f"{normalized}/api/mesh/infonet/sync", **kwargs)
# HTTP 429 must be surfaced as a typed exception carrying the
# Retry-After value, so finish_sync can honor it and stop hammering
# the upstream. Pre-fix this path just stringified the status into
# a ValueError, which finish_sync then ignored — keeping the
# upstream's rate-limit bucket full indefinitely.
if response.status_code == 429:
from services.mesh.mesh_infonet_sync_support import (
PeerSyncRateLimited,
parse_retry_after_header,
)
retry_after_s = parse_retry_after_header(
response.headers.get("Retry-After", "") or "",
)
try:
body_text = response.text[:200]
except Exception:
body_text = ""
raise PeerSyncRateLimited(
f"HTTP 429 from {normalized} (retry_after={retry_after_s}s): {body_text}",
retry_after_s=retry_after_s,
status=429,
)
try:
payload = response.json()
except Exception as exc:
@@ -1462,8 +1485,23 @@ def _hydrate_gate_store_from_chain(events: list[dict]) -> int:
return count
def _sync_from_peer(peer_url: str, *, page_limit: int = 100, max_rounds: int = 5) -> tuple[bool, str, bool]:
def _sync_from_peer(
peer_url: str,
*,
page_limit: int = 100,
max_rounds: int = 5,
) -> tuple[bool, str, bool, int]:
"""Sync the local Infonet chain against ``peer_url``.
Returns ``(ok, error, forked, retry_after_s)``. The fourth tuple
element is non-zero only when the peer responded with HTTP 429
and supplied a parseable ``Retry-After`` header see the typed
``PeerSyncRateLimited`` exception in mesh_infonet_sync_support.py.
Callers should pass that value to ``finish_sync(retry_after_s=...)``
so the next attempt actually waits.
"""
from services.mesh.mesh_hashchain import infonet
from services.mesh.mesh_infonet_sync_support import PeerSyncRateLimited
rounds = 0
while rounds < max_rounds:
@@ -1472,7 +1510,11 @@ def _sync_from_peer(peer_url: str, *, page_limit: int = 100, max_rounds: int = 5
"locator": infonet.get_locator(),
"limit": page_limit,
}
payload = _peer_sync_response(peer_url, body)
try:
payload = _peer_sync_response(peer_url, body)
except PeerSyncRateLimited as exc:
# Bubble up the retry-after so finish_sync can honor it.
return False, str(exc), False, exc.retry_after_s
if bool(payload.get("forked")):
# Auto-recover small local forks: if the local chain is tiny
# (< 20 events) and the remote has a longer chain, reset local
@@ -1488,23 +1530,23 @@ def _sync_from_peer(peer_url: str, *, page_limit: int = 100, max_rounds: int = 5
)
infonet.reset_chain()
continue # retry sync with clean genesis locator
return False, "fork detected", True
return False, "fork detected", True, 0
events = payload.get("events", [])
if not isinstance(events, list):
return False, "peer sync events must be a list", False
return False, "peer sync events must be a list", False, 0
if not events:
return True, "", False
return True, "", False, 0
result = infonet.ingest_events(events)
_hydrate_gate_store_from_chain(events)
rejected = list(result.get("rejected", []) or [])
if rejected:
return False, f"sync ingest rejected {len(rejected)} event(s)", False
return False, f"sync ingest rejected {len(rejected)} event(s)", False, 0
if int(result.get("accepted", 0) or 0) == 0 and int(result.get("duplicates", 0) or 0) >= len(events):
return True, "", False
return True, "", False, 0
if len(events) < page_limit:
return True, "", False
return True, "", False, 0
rounds += 1
return True, "", False
return True, "", False, 0
def _run_public_sync_cycle() -> SyncWorkerState:
@@ -1567,11 +1609,12 @@ def _run_public_sync_cycle() -> SyncWorkerState:
with _NODE_RUNTIME_LOCK:
set_sync_state(started)
try:
ok, error, forked = _sync_from_peer(record.peer_url)
ok, error, forked, retry_after_s = _sync_from_peer(record.peer_url)
except Exception as exc:
ok = False
error = str(exc or type(exc).__name__)
forked = False
retry_after_s = 0
if ok:
store.mark_seen(record.peer_url, "sync", now=time.time())
store.mark_sync_success(record.peer_url, now=time.time())
@@ -1618,6 +1661,12 @@ def _run_public_sync_cycle() -> SyncWorkerState:
now=time.time(),
interval_s=int(get_settings().MESH_SYNC_INTERVAL_S or 300),
failure_backoff_s=failure_backoff_s,
# 429 retry-storm fix: when the peer returned HTTP 429 with
# a Retry-After header, finish_sync uses max(exponential,
# retry_after) for next_sync_due_at — so we actually wait
# the time the upstream asked for instead of hammering
# every 60s and keeping its rate-limit bucket full forever.
retry_after_s=retry_after_s,
)
with _NODE_RUNTIME_LOCK:
set_sync_state(updated)
+193 -39
View File
@@ -2521,45 +2521,85 @@ async def api_capabilities(request: Request):
# OpenClaw Connection Management (local-operator only — NOT via HMAC)
# These endpoints manage the HMAC secret itself, so they MUST require
# local operator access to prevent privilege escalation.
#
# Issue #302 (tg12): pre-fix, GET /api/ai/connect-info had two problems:
#
# 1. ``?reveal=true`` made the full secret travel through every operator
# page-load that opened the Connect modal. Even gated to
# ``require_local_operator``, that put the secret into browser
# history, dev-tools network panels, browser disk caches, HAR
# exports, and screen captures. Every time the modal opened.
#
# 2. The same GET endpoint auto-bootstrapped (generated + persisted)
# the secret on first read. Side effects on a GET are a footgun:
# browser prefetchers, mirror tools, and casual curl-from-history
# would all silently mint+persist a fresh secret. (Gated, but
# still surprising — and noisy in the audit log.)
#
# Resolution:
#
# GET /api/ai/connect-info — always returns the MASKED
# secret. No ?reveal param.
# No auto-bootstrap; if the
# secret is missing,
# ``hmac_secret_set: false``
# tells the frontend to call
# /bootstrap.
#
# POST /api/ai/connect-info/bootstrap — NEW. Generates + persists the
# secret if missing. Idempotent.
# Returns metadata only, never
# the full secret.
#
# POST /api/ai/connect-info/reveal — NEW. Returns the full secret in
# the body with strict
# ``Cache-Control: no-store,
# no-cache, must-revalidate``
# + ``Pragma: no-cache`` so
# it does not land in browser
# caches. POST means it does
# not land in URL history.
#
# POST /api/ai/connect-info/regenerate — keeps existing one-time-reveal
# behavior (regenerate IS a
# deliberate destructive action
# the operator triggered, so
# displaying the new secret
# once is the only path that
# makes the operation useful).
# Same no-store headers added.
# ---------------------------------------------------------------------------
@router.get("/api/ai/connect-info", dependencies=[Depends(require_local_operator)])
@limiter.limit("30/minute")
async def get_connect_info(request: Request, reveal: bool = False):
"""Return connection details for the OpenClaw Connect modal.
# Cache-Control headers that should accompany every response carrying the
# full HMAC secret. Reused across the reveal + regenerate endpoints so a
# future refactor that splits or renames them can't forget the headers.
_NO_STORE_HEADERS = {
"Cache-Control": "no-store, no-cache, must-revalidate, private",
"Pragma": "no-cache",
"Expires": "0",
}
The HMAC secret is masked by default. Pass ?reveal=true to see the full key.
Private keys are NEVER returned.
def _mask_hmac_secret(secret: str) -> str:
"""Return a fingerprint-style mask (first6 + bullets + last4) suitable
for display in the UI before the operator clicks Reveal."""
if not secret:
return ""
if len(secret) > 10:
return secret[:6] + "••••••••" + secret[-4:]
return "••••••••"
def _connect_info_metadata(settings) -> dict:
"""Return everything the Connect modal needs EXCEPT the secret itself.
Shared between GET /api/ai/connect-info (where the full secret is
masked) and POST /api/ai/connect-info/bootstrap (where the operator
just generated a secret but we don't return it inline — they have to
call /reveal to see it).
"""
import os
import secrets
from services.config import get_settings
settings = get_settings()
hmac_secret = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
access_tier = str(settings.OPENCLAW_ACCESS_TIER or "restricted").strip().lower()
# Auto-generate if not set
if not hmac_secret:
hmac_secret = secrets.token_hex(24) # 48 chars
_write_env_value("OPENCLAW_HMAC_SECRET", hmac_secret)
# Clear settings cache so next read picks up the new value
get_settings.cache_clear()
masked = hmac_secret[:6] + "••••••••" + hmac_secret[-4:] if len(hmac_secret) > 10 else "••••••••"
return {
"ok": True,
"hmac_secret": hmac_secret if reveal else masked,
"hmac_secret_set": bool(hmac_secret),
"bootstrap_behavior": {
"auto_generates_when_missing": True,
"auto_generated_this_call": not bool(settings.OPENCLAW_HMAC_SECRET or ""),
"notes": [
"If no HMAC secret exists yet, this endpoint bootstraps one and persists it to .env.",
"Regenerating the HMAC secret revokes all existing direct-mode OpenClaw callers at once.",
],
},
"access_tier": access_tier,
"trust_model": {
"remote_http_principal": "holder_of_openclaw_hmac_secret",
@@ -2613,24 +2653,138 @@ async def get_connect_info(request: Request, reveal: bool = False):
}
@router.post("/api/ai/connect-info/regenerate", dependencies=[Depends(require_local_operator)])
@limiter.limit("5/minute")
async def regenerate_hmac_secret(request: Request):
"""Generate a new HMAC secret. Old secret immediately stops working."""
@router.get("/api/ai/connect-info", dependencies=[Depends(require_local_operator)])
@limiter.limit("30/minute")
async def get_connect_info(request: Request):
"""Return connection details for the OpenClaw Connect modal.
The HMAC secret is always returned as a fingerprint mask
(``first6 + bullets + last4``); the full value is only ever served by
``POST /api/ai/connect-info/reveal`` (see #302). When the secret has
not been bootstrapped yet, ``hmac_secret_set`` is false and the
frontend should call ``POST /api/ai/connect-info/bootstrap``.
Private keys are NEVER returned.
"""
from services.config import get_settings
settings = get_settings()
hmac_secret = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
return {
"ok": True,
"masked_hmac_secret": _mask_hmac_secret(hmac_secret),
"hmac_secret_set": bool(hmac_secret),
"bootstrap_behavior": {
"auto_generates_when_missing": False,
"notes": [
"Call POST /api/ai/connect-info/bootstrap to mint a secret on first use.",
"Call POST /api/ai/connect-info/reveal to see the full secret (no-store).",
"Regenerating the HMAC secret revokes all existing direct-mode OpenClaw callers at once.",
],
},
**_connect_info_metadata(settings),
}
@router.post("/api/ai/connect-info/bootstrap", dependencies=[Depends(require_local_operator)])
@limiter.limit("10/minute")
async def bootstrap_hmac_secret(request: Request):
"""Mint and persist the OpenClaw HMAC secret if it isn't already set.
Idempotent: if a secret already exists, returns ``generated: false``
and leaves the existing secret untouched. Never returns the secret
value in the response body — the operator calls
``POST /api/ai/connect-info/reveal`` to see it.
"""
import secrets
from services.config import get_settings
settings = get_settings()
existing = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
if existing:
return {
"ok": True,
"generated": False,
"hmac_secret_set": True,
"masked_hmac_secret": _mask_hmac_secret(existing),
"detail": "HMAC secret already configured. Use /reveal to see it.",
}
new_secret = secrets.token_hex(24) # 48 chars
_write_env_value("OPENCLAW_HMAC_SECRET", new_secret)
get_settings.cache_clear()
return {
"ok": True,
"hmac_secret": new_secret,
"detail": "HMAC secret regenerated. Update your OpenClaw agent configuration.",
"generated": True,
"hmac_secret_set": True,
"masked_hmac_secret": _mask_hmac_secret(new_secret),
"detail": "HMAC secret generated. Call /reveal to copy it into your OpenClaw config.",
}
@router.post("/api/ai/connect-info/reveal", dependencies=[Depends(require_local_operator)])
@limiter.limit("10/minute")
async def reveal_hmac_secret(request: Request):
"""Return the full HMAC secret in the response body.
POST (not GET) so the secret never lands in URL history, access logs,
or browser visit history. Strict ``Cache-Control: no-store`` headers
prevent intermediaries from persisting the response. Returns 404 if
no secret has been bootstrapped — the frontend should call
``POST /api/ai/connect-info/bootstrap`` first.
"""
from services.config import get_settings
settings = get_settings()
hmac_secret = str(settings.OPENCLAW_HMAC_SECRET or "").strip()
if not hmac_secret:
raise HTTPException(
404,
"No HMAC secret configured. Call POST /api/ai/connect-info/bootstrap first.",
)
return JSONResponse(
content={
"ok": True,
"hmac_secret": hmac_secret,
"masked_hmac_secret": _mask_hmac_secret(hmac_secret),
},
headers=_NO_STORE_HEADERS,
)
@router.post("/api/ai/connect-info/regenerate", dependencies=[Depends(require_local_operator)])
@limiter.limit("5/minute")
async def regenerate_hmac_secret(request: Request):
"""Generate a new HMAC secret. Old secret immediately stops working.
Returns the new secret in the response body — this is the only
operation where the full secret travels back through the response,
because regenerating IS a deliberate destructive action the operator
triggered and they need to see the new value once to update their
OpenClaw configuration. Strict ``Cache-Control: no-store`` headers
keep it from being persisted by browser caches, proxies, or HAR
capture tooling.
"""
import secrets
from services.config import get_settings
new_secret = secrets.token_hex(24) # 48 chars
_write_env_value("OPENCLAW_HMAC_SECRET", new_secret)
get_settings.cache_clear()
return JSONResponse(
content={
"ok": True,
"hmac_secret": new_secret,
"masked_hmac_secret": _mask_hmac_secret(new_secret),
"detail": "HMAC secret regenerated. Update your OpenClaw agent configuration.",
},
headers=_NO_STORE_HEADERS,
)
@router.put("/api/ai/connect-info/access-tier", dependencies=[Depends(require_local_operator)])
@limiter.limit("10/minute")
async def set_access_tier(request: Request, body: dict):
+15
View File
@@ -59,6 +59,12 @@ async def health_check(request: Request):
# when the SPKI-pinned fallback is in effect. The data plane keeps
# flowing (this is by design — see ais_proxy.js comments) but observers
# who care about MITM-protection posture deserve a visible signal.
#
# Plus connectivity health (added 2026-05-23 when stream.aisstream.io
# went fully offline): ``connected`` tells the frontend whether ship
# data is actually flowing. When false, a banner explains that ships
# are unavailable due to an upstream outage — better than the user
# silently seeing an empty ocean and assuming we broke something.
ais_status: dict = {}
try:
from services.ais_stream import ais_proxy_status
@@ -69,6 +75,15 @@ async def health_check(request: Request):
# Don't override a worse top-level status if SLOs already failed,
# but escalate ok -> degraded so the field surfaces in dashboards.
top_status = "degraded"
# AIS_API_KEY not configured is "feature off", not "system broken" —
# so we only escalate when the operator opted into AIS (key set) AND
# the stream is currently offline.
if (
os.environ.get("AIS_API_KEY")
and ais_status.get("connected") is False
and top_status == "ok"
):
top_status = "degraded"
return {
"status": top_status,
+58
View File
@@ -85,6 +85,64 @@ async def infonet_peer_push(request: Request):
return {"ok": True, **result}
@router.post("/api/mesh/dm/replicate-envelope")
@limiter.limit("60/minute")
async def dm_replicate_envelope(request: Request):
"""Accept a DM envelope replicated from a peer relay (cross-node mailbox).
Companion endpoint to ``DMRelay.replicate_to_peers`` (outbound, in
``mesh_dm_relay.py``). The sender's relay POSTs an encrypted DM
envelope here after a successful local ``deposit``; this endpoint
re-enforces the per-(sender, recipient) anti-spam cap and stores
the envelope in the local mailbox if accepted.
The cap is the network rule: a hostile sender's relay can spool
extras locally, but every honest peer enforces the cap on inbound
replication. Recipient polling from any honest peer therefore
never sees more than ``MESH_DM_PENDING_PER_SENDER_LIMIT`` pending
from any one sender, no matter how many spam attempts were tried.
Same HMAC auth pattern as ``infonet_peer_push`` and ``gate_peer_push``.
"""
content_length = request.headers.get("content-length")
if content_length:
try:
# DM envelopes are bounded by MESH_DM_MAX_MSG_BYTES + envelope
# overhead; 64 KB is a generous ceiling.
if int(content_length) > 65_536:
return Response(
content='{"ok":false,"detail":"Request body too large (max 64KB)"}',
status_code=413, media_type="application/json",
)
except (ValueError, TypeError):
pass
body_bytes = await request.body()
if not _verify_peer_push_hmac(request, body_bytes):
return Response(
content='{"ok":false,"detail":"Invalid or missing peer HMAC"}',
status_code=403, media_type="application/json",
)
try:
body = json_mod.loads(body_bytes or b"{}")
except (ValueError, TypeError):
return Response(
content='{"ok":false,"detail":"Invalid JSON body"}',
status_code=400, media_type="application/json",
)
envelope = body.get("envelope")
if not isinstance(envelope, dict):
return {"ok": False, "detail": "envelope must be an object"}
originating_peer = _peer_hmac_url_from_request(request) or ""
from services.mesh.mesh_dm_relay import dm_relay
result = dm_relay.accept_replica(
envelope=envelope,
originating_peer_url=originating_peer,
)
return result
@router.post("/api/mesh/gate/peer-push")
@limiter.limit("30/minute")
async def gate_peer_push(request: Request):
+61 -7
View File
@@ -120,18 +120,60 @@ def api_sentinel2_search(
return search_sentinel2_scene(lat, lng)
# Issue #298 (tg12): Sentinel credentials moved server-side
# ---------------------------------------------------------------------------
# Previously the frontend kept Copernicus CDSE client_id + client_secret in
# browser localStorage / sessionStorage and forwarded them on every tile
# request through this proxy. That exposed real third-party credentials to
# any same-origin script (XSS, malicious browser extension, dev-tools HAR
# export).
#
# Resolution order (first match wins):
# 1. Request body — kept for back-compat. A small number of legacy
# operator setups may still post credentials; we don't break them.
# 2. Backend .env — SENTINEL_CLIENT_ID / SENTINEL_CLIENT_SECRET, managed
# through the existing /api/settings/api-keys flow (admin-gated).
#
# The frontend in ``sentinelHub.ts`` no longer reads browser storage and no
# longer forwards credentials — every dashboard request now lands in (2).
# The require_local_operator gate (added in #303/PR #303) stays — both layers
# are independent: the gate blocks anonymous callers, the env fallback lets
# legitimate (gated) callers omit credentials from the body.
# ---------------------------------------------------------------------------
def _resolve_sentinel_credentials(body_id: str, body_secret: str) -> tuple[str, str]:
"""Return (client_id, client_secret) using body values when present,
otherwise falling back to backend .env. Empty strings if neither is set."""
import os as _os
cid = (body_id or "").strip() or (_os.environ.get("SENTINEL_CLIENT_ID", "") or "").strip()
csec = (body_secret or "").strip() or (_os.environ.get("SENTINEL_CLIENT_SECRET", "") or "").strip()
return cid, csec
@router.post("/api/sentinel/token", dependencies=[Depends(require_local_operator)])
@limiter.limit("60/minute")
async def api_sentinel_token(request: Request):
"""Proxy Copernicus CDSE OAuth2 token request (avoids browser CORS block)."""
"""Proxy Copernicus CDSE OAuth2 token request (avoids browser CORS block).
Credentials are resolved by ``_resolve_sentinel_credentials`` — body
fields are honored for back-compat, otherwise the backend .env values
populated through ``/api/settings/api-keys`` are used.
"""
import requests as req
body = await request.body()
from urllib.parse import parse_qs
params = parse_qs(body.decode("utf-8"))
client_id = params.get("client_id", [""])[0]
client_secret = params.get("client_secret", [""])[0]
body_id = params.get("client_id", [""])[0]
body_secret = params.get("client_secret", [""])[0]
client_id, client_secret = _resolve_sentinel_credentials(body_id, body_secret)
if not client_id or not client_secret:
raise HTTPException(400, "client_id and client_secret required")
# Friendly, non-hostile error — points the operator at the place
# they configure other API keys instead of just saying "required".
raise HTTPException(
400,
"Sentinel client_id/client_secret are not configured. "
"Set SENTINEL_CLIENT_ID and SENTINEL_CLIENT_SECRET in the "
"API Keys panel (Settings → API Keys) or your backend .env.",
)
token_url = "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token"
try:
resp = await asyncio.to_thread(req.post, token_url,
@@ -186,8 +228,11 @@ async def api_sentinel_tile(request: Request):
except Exception:
return JSONResponse(status_code=422, content={"ok": False, "detail": "invalid JSON body"})
client_id = body.get("client_id", "")
client_secret = body.get("client_secret", "")
# Issue #298: same resolution order as /api/sentinel/token — body
# values for back-compat, otherwise backend .env.
body_id = body.get("client_id", "")
body_secret = body.get("client_secret", "")
client_id, client_secret = _resolve_sentinel_credentials(body_id, body_secret)
preset = body.get("preset", "TRUE-COLOR")
date_str = body.get("date", "")
z = body.get("z", 0)
@@ -195,7 +240,16 @@ async def api_sentinel_tile(request: Request):
y = body.get("y", 0)
if not client_id or not client_secret or not date_str:
raise HTTPException(400, "client_id, client_secret, and date required")
# Distinguish "no creds" from "no date" so the operator knows
# what to fix. Same friendly pointer as the /token route.
if not client_id or not client_secret:
raise HTTPException(
400,
"Sentinel client_id/client_secret are not configured. "
"Set SENTINEL_CLIENT_ID and SENTINEL_CLIENT_SECRET in the "
"API Keys panel (Settings → API Keys) or your backend .env.",
)
raise HTTPException(400, "date required")
now = _time.time()
credential_fp = _credential_fingerprint(client_id, client_secret)
+54 -7
View File
@@ -350,19 +350,58 @@ _proxy_process = None
# path during an upstream cert outage. Surfaced via ais_proxy_status() for
# /api/health.
_proxy_status: dict = {}
# Upstream-connectivity telemetry (added when stream.aisstream.io went fully
# offline on 2026-05-23). ``_last_msg_at`` is the unix timestamp of the most
# recent vessel message received from the proxy. ``_proxy_spawn_count`` is
# how many times we've started the node proxy; combined with no recent
# messages it tells us the proxy is respawning in a tight loop because the
# upstream is unreachable. Surfaced via ais_proxy_status() so the operator
# can see "AIS is dead" instead of guessing whether it's their map filter,
# their api key, or upstream.
_last_msg_at: float = 0.0
_proxy_spawn_count: int = 0
_VESSEL_TRAIL_INTERVAL_S = 120
_VESSEL_TRAIL_MAX_POINTS = 240
def ais_proxy_status() -> dict:
"""Return a copy of the latest ais_proxy.js status (issue #258).
# How stale "last vessel message" can be before we consider the stream
# disconnected. AISStream typically pushes multiple messages/sec, so a 60s
# gap means something's wrong upstream or in transit.
_AIS_CONNECTED_FRESHNESS_S = 60
Currently surfaces ``degraded_tls`` (bool) which is true when the
proxy is using SPKI-pinned fallback because AISStream's cert expired.
Returns an empty dict when no status has been received yet.
def ais_proxy_status() -> dict:
"""Return a copy of the latest ais_proxy.js status + connectivity health.
Fields:
* ``degraded_tls`` (bool, issue #258) — true when the proxy is using
SPKI-pinned fallback because AISStream's cert expired.
* ``connected`` (bool) — true when we received a vessel message in
the last ``_AIS_CONNECTED_FRESHNESS_S`` seconds.
* ``last_msg_age_seconds`` (int | None) — seconds since the last
vessel message; None if we've never received one.
* ``proxy_spawn_count`` (int) — how many times we've spawned the
node proxy. Sustained increases here without ``connected`` means
we're respawning in a tight loop because upstream is dead.
Returns an empty dict when called before the AIS subsystem starts
(e.g. during tests or when no API key is set).
"""
with _vessels_lock:
return dict(_proxy_status)
status = dict(_proxy_status)
last = _last_msg_at
spawns = _proxy_spawn_count
now = time.time()
if last > 0:
last_age = int(now - last)
status["last_msg_age_seconds"] = last_age
status["connected"] = last_age <= _AIS_CONNECTED_FRESHNESS_S
else:
status["last_msg_age_seconds"] = None
status["connected"] = False
status["proxy_spawn_count"] = spawns
return status
import os
@@ -588,8 +627,10 @@ def _ais_stream_loop():
env=proxy_env,
**popen_kwargs,
)
global _proxy_spawn_count
with _vessels_lock:
_proxy_process = process
_proxy_spawn_count += 1
# Drain stderr in a background thread to prevent deadlock
import threading
@@ -645,9 +686,15 @@ def _ais_stream_loop():
if not mmsi:
continue
# Telemetry: stamp the timestamp of the most recent real
# vessel message. ais_proxy_status() reads this to decide
# whether the stream is currently "connected" — i.e. has
# any data flowed in the last 60s.
global _last_msg_at
with _vessels_lock:
_last_msg_at = time.time()
if mmsi not in _vessels:
_vessels[mmsi] = {"_updated": time.time()}
_vessels[mmsi] = {"_updated": _last_msg_at}
vessel = _vessels[mmsi]
# Update position from PositionReport or StandardClassBPositionReport
+25
View File
@@ -150,6 +150,31 @@ API_REGISTRY = [
"url": "https://finnhub.io/register",
"required": False,
},
# Issue #298 (tg12): Sentinel Hub / Copernicus Data Space Ecosystem
# credentials were previously held in browser localStorage / sessionStorage
# by the Settings panel. Moved server-side to the same .env-backed
# store every other third-party API key lives in. The Sentinel proxy
# routes (POST /api/sentinel/token, /tile) now fall back to these
# env values when the request body omits credentials — see
# backend/routers/tools.py for the resolution order.
{
"id": "sentinel_client_id",
"env_key": "SENTINEL_CLIENT_ID",
"name": "Sentinel Hub / Copernicus — Client ID",
"description": "OAuth2 client ID for Copernicus Data Space Ecosystem (CDSE). Required for the Sentinel-2 imagery overlay and the right-click Sentinel-2 Intel Card. Sign in at dataspace.copernicus.eu and create OAuth credentials.",
"category": "Imagery",
"url": "https://dataspace.copernicus.eu/",
"required": False,
},
{
"id": "sentinel_client_secret",
"env_key": "SENTINEL_CLIENT_SECRET",
"name": "Sentinel Hub / Copernicus — Client Secret",
"description": "OAuth2 client secret paired with the Client ID above. Used by the backend to mint short-lived access tokens against the CDSE identity provider. Stored in the backend .env; never sent to the browser.",
"category": "Imagery",
"url": "https://dataspace.copernicus.eu/",
"required": False,
},
]
ALLOWED_ENV_KEYS = {
+15
View File
@@ -116,6 +116,21 @@ class Settings(BaseSettings):
MESH_DM_REQUEST_MAILBOX_LIMIT: int = 12
MESH_DM_SHARED_MAILBOX_LIMIT: int = 48
MESH_DM_SELF_MAILBOX_LIMIT: int = 12
# Anti-spam: cap on distinct UNACKED messages a single sender can have
# parked in a single recipient's mailbox at any one time. Once the
# recipient pulls (acks) a message, the sender's quota for that pair
# frees up. Default 2 — a sender who wants to deliver more must wait
# for the recipient to actually read the prior messages.
#
# This cap is enforced TWICE: once on the local deposit path (the
# sender's own node refuses to spool the 3rd message) AND once on
# the replication-acceptance path (honest peer relays refuse to
# accept inbound replicas that would put them over the cap). The
# double enforcement makes the rule a NETWORK rule — patching out
# the local check on a hostile sender's relay doesn't let extras
# propagate, because every honest peer enforces the same cap on
# inbound replication.
MESH_DM_PENDING_PER_SENDER_LIMIT: int = 2
MESH_BLOCK_LEGACY_AGENT_ID_LOOKUP: bool = True
MESH_ALLOW_COMPAT_DM_INVITE_IMPORT: bool = False
MESH_ALLOW_COMPAT_DM_INVITE_IMPORT_UNTIL: str = ""
+7 -2
View File
@@ -11,8 +11,13 @@ DEFAULT_TRAIL_TTL_S = 300 # 5 min - trail TTL for non-tracked flights
HOLD_PATTERN_DEGREES = 300 # Total heading change to flag holding pattern
GPS_JAMMING_NACP_THRESHOLD = 8 # NACp below this = degraded GPS signal
GPS_JAMMING_GRID_SIZE = 1.0 # 1 degree grid for aggregation
GPS_JAMMING_MIN_RATIO = 0.30 # 30% degraded aircraft to flag zone
GPS_JAMMING_MIN_AIRCRAFT = 5 # Min aircraft in grid cell for statistical significance
# Tuned 2026-05: previously 0.30 / 5 aircraft which — combined with the
# -1 noise cushion in the detector AND the pre-fix nac_p==0 filter that
# discarded jamming victims — meant the layer almost never lit up.
# Lowering the bar so genuine jamming zones with sparser ADS-B coverage
# clear (eastern Med, Russia/Ukraine border, Iran/Iraq).
GPS_JAMMING_MIN_RATIO = 0.20 # 20% degraded aircraft to flag zone
GPS_JAMMING_MIN_AIRCRAFT = 3 # Min aircraft in grid cell for statistical significance
# ─── Network & Circuit Breaker ──────────────────────────────────────────────
CIRCUIT_BREAKER_TTL_S = 120 # Skip domain for 2 min after total failure
+38 -2
View File
@@ -777,6 +777,39 @@ def start_scheduler():
misfire_grace_time=60,
)
# Flight observation pruning — drops icao24 → first_seen_at entries we
# haven't seen in an hour. Same cadence as AIS prune for symmetry; the
# per-tick scan is O(in-flight aircraft) so it's cheap.
from services.fetchers.flight_observations import prune as _prune_flight_observations
_scheduler.add_job(
lambda: _run_task_with_health(_prune_flight_observations, "prune_flight_observations"),
"interval",
minutes=5,
id="flight_observation_prune",
max_instances=1,
misfire_grace_time=60,
)
# AISHub REST fallback — slow polling when the AISStream WebSocket
# primary is offline. Configurable interval via
# AISHUB_POLL_INTERVAL_MINUTES env (default 20 min). Operator must
# set AISHUB_USERNAME to opt in. The fetcher is gated internally on
# the primary being disconnected, so this job is cheap when the
# WebSocket is healthy (early-returns after a status check).
from services.fetchers.aishub_fallback import (
aishub_poll_interval_minutes,
fetch_aishub_vessels,
)
_aishub_interval = aishub_poll_interval_minutes()
_scheduler.add_job(
lambda: _run_task_with_health(fetch_aishub_vessels, "fetch_aishub_vessels"),
"interval",
minutes=_aishub_interval,
id="aishub_fallback",
max_instances=1,
misfire_grace_time=120,
)
# Route database — bulk refresh from vrs-standing-data.adsb.lol every 5
# days. Replaces the legacy /api/0/routeset POST (blocked under our UA,
# and broken upstream). Airline schedules change on a quarterly cycle,
@@ -960,16 +993,19 @@ def start_scheduler():
misfire_grace_time=600,
)
# UAP sightings (NUFORC) — daily at 12:00 UTC
# UAP sightings (NUFORC) — weekly on Mondays at 12:00 UTC. The layer is a
# rolling last-60-days digest; refreshing once a week is enough cadence
# for human-readable map exploration and keeps load on nuforc.org light.
_scheduler.add_job(
lambda: _run_task_with_health(
lambda: fetch_uap_sightings(force_refresh=True),
"fetch_uap_sightings",
),
"cron",
day_of_week="mon",
hour=12,
minute=0,
id="uap_sightings_daily",
id="uap_sightings_weekly",
max_instances=1,
misfire_grace_time=3600,
)
@@ -0,0 +1,290 @@
"""AISHub REST fallback for ship tracking when AISStream is unreachable.
Background
----------
On 2026-05-23 ``stream.aisstream.io`` (the primary live AIS WebSocket feed)
went fully offline. Backend's only ship signal vanished. This module polls
``data.aishub.net``'s free REST API on a slow cadence (default 20 min) when
the WebSocket primary is disconnected, so the ships layer doesn't go fully
dark during upstream outages.
Why 20 minutes
--------------
AISHub's free tier is rate-limited and explicitly asks consumers to be
courteous. 20 minutes is well inside their limits, gives ships time to
move enough to look "alive" on the map, and won't drain their service.
Configurable via the ``AISHUB_POLL_INTERVAL_MINUTES`` env var (clamped to
[1, 360]).
Why slow vs primary
-------------------
This is degraded mode, not a replacement. A ship at 20 knots moves about
6 nautical miles in 20 minutes visible on the map but coarser than the
real-time WebSocket signal. When AISStream comes back online, the
WebSocket data will overwrite these records via the same ``_vessels``
dict and ``source`` will flip from ``"aishub"`` back to upstream-live.
Opt-in
------
Operator must set ``AISHUB_USERNAME`` (free registration at
https://www.aishub.net/api). If unset, this fetcher is a no-op.
"""
from __future__ import annotations
import json
import logging
import os
import time
from typing import Any
from services.network_utils import fetch_with_curl
logger = logging.getLogger(__name__)
AISHUB_URL = "https://data.aishub.net/ws.php"
def aishub_username() -> str:
return str(os.environ.get("AISHUB_USERNAME", "")).strip()
def aishub_fallback_enabled() -> bool:
"""Returns True only when the operator has registered with AISHub and
set ``AISHUB_USERNAME``. The presence of the username is the opt-in."""
return bool(aishub_username())
def aishub_poll_interval_minutes() -> int:
"""Default 20 minutes. Clamped to [1, 360] so a hostile or
misconfigured env var can't either hammer the upstream or silence the
fallback for a day."""
raw = os.environ.get("AISHUB_POLL_INTERVAL_MINUTES", "20")
try:
value = int(str(raw).strip())
except (TypeError, ValueError):
value = 20
return max(1, min(360, value))
def _should_run_fallback() -> bool:
"""Only run when the primary WebSocket is disconnected. Avoids stomping
over fresher live data when AISStream is healthy.
Returns False if:
* AISHub isn't configured (no username)
* AISStream primary is currently connected (recent vessel messages)
Returns True only when AIS is configured-but-down. The
``proxy_spawn_count > 0`` guard means "the primary has at least tried
to run" — if the user set AISHUB_USERNAME but not AIS_API_KEY at all,
AISHub will still serve as a primary on its own slow cadence.
"""
if not aishub_fallback_enabled():
return False
try:
from services.ais_stream import ais_proxy_status
status = ais_proxy_status() or {}
except Exception:
return True # ais_stream not importable? still try AISHub.
# If the WebSocket primary is connected, skip the fallback — fresher
# data is already flowing.
if status.get("connected") is True:
return False
return True
def _parse_aishub_response(payload: str) -> list[dict]:
"""Parse the AISHub JSON response into a list of vessel records.
Successful response shape::
[
{"ERROR": false, "USERNAME": "...", "FORMAT": "1", "RECORDS": N},
[{"MMSI": ..., "LATITUDE": ..., "LONGITUDE": ..., ...}, ...]
]
Error response shape::
[{"ERROR": true, "ERROR_MESSAGE": "..."}]
Empty payload (e.g. silent rate-limit drop) returns ``[]``.
"""
if not payload or not payload.strip():
return []
try:
data = json.loads(payload)
except json.JSONDecodeError as e:
logger.warning("AISHub: response is not JSON: %s", e)
return []
if not isinstance(data, list) or not data:
return []
header = data[0] if isinstance(data[0], dict) else {}
if header.get("ERROR") is True:
logger.warning(
"AISHub: upstream error: %s",
header.get("ERROR_MESSAGE", "<unspecified>"),
)
return []
if len(data) < 2 or not isinstance(data[1], list):
return []
return [row for row in data[1] if isinstance(row, dict)]
def _normalize_record(row: dict) -> dict | None:
"""Map an AISHub vessel record to our internal vessel schema.
Returns None when the record can't be used (no MMSI, bad position,
sentinel "not available" lat/lng).
"""
try:
mmsi = int(row.get("MMSI") or 0)
except (TypeError, ValueError):
return None
if not mmsi:
return None
try:
lat = float(row.get("LATITUDE"))
lng = float(row.get("LONGITUDE"))
except (TypeError, ValueError):
return None
# AIS uses 91/181 as "no position available" sentinels.
if abs(lat) > 90 or abs(lng) > 180:
return None
if lat == 91.0 or lng == 181.0:
return None
# SOG raw 102.3 is "speed not available"; sanitize to 0.
try:
sog_raw = float(row.get("SOG") or 0)
except (TypeError, ValueError):
sog_raw = 0.0
sog = 0.0 if sog_raw >= 102.2 else sog_raw
try:
cog = float(row.get("COG") or 0)
except (TypeError, ValueError):
cog = 0.0
try:
heading_raw = int(row.get("HEADING") or 511)
except (TypeError, ValueError):
heading_raw = 511
# AIS heading sentinel 511 = "not available" — fall back to COG.
heading = heading_raw if heading_raw != 511 else cog
try:
ais_type = int(row.get("TYPE") or 0)
except (TypeError, ValueError):
ais_type = 0
return {
"mmsi": mmsi,
"lat": lat,
"lng": lng,
"sog": sog,
"cog": cog,
"heading": heading,
"name": str(row.get("NAME") or "").strip() or "UNKNOWN",
"callsign": str(row.get("CALLSIGN") or "").strip(),
"destination": str(row.get("DEST") or "").strip().replace("@", "") or "",
"imo": int(row.get("IMO") or 0),
"ais_type_code": ais_type,
}
def fetch_aishub_vessels() -> int:
"""Poll AISHub and merge vessels into the shared ``_vessels`` store.
Returns the number of vessels updated (0 on skip, error, or no data).
Designed to be called by the APScheduler tier see
``data_fetcher.py`` for the 20-minute interval job that wraps this.
"""
if not _should_run_fallback():
logger.debug("AISHub fallback skipped: primary connected or not configured")
return 0
username = aishub_username()
url = (
f"{AISHUB_URL}?username={username}&format=1&output=json"
f"&compress=0"
)
try:
response = fetch_with_curl(url, timeout=30)
except Exception as e:
logger.warning("AISHub fetch failed: %s", e)
return 0
if not response or response.status_code != 200:
logger.warning(
"AISHub HTTP %s",
getattr(response, "status_code", "None"),
)
return 0
rows = _parse_aishub_response(getattr(response, "text", "") or "")
if not rows:
return 0
# Inline imports to avoid a circular dependency at module load time
# (ais_stream imports lots of things and is loaded by main.py).
from services.ais_stream import (
_vessels,
_vessels_lock,
_record_vessel_trail_locked,
classify_vessel,
get_country_from_mmsi,
)
now = time.time()
count = 0
with _vessels_lock:
for row in rows:
normalized = _normalize_record(row)
if normalized is None:
continue
mmsi = normalized["mmsi"]
vessel = _vessels.setdefault(mmsi, {"mmsi": mmsi})
# Don't overwrite fresher live data: if the WebSocket pushed an
# update for this MMSI more recently than now-1s (race during
# the brief reconnection window) keep the live one.
last = float(vessel.get("_updated") or 0)
if last > now - 1:
continue
vessel.update(
{
"lat": normalized["lat"],
"lng": normalized["lng"],
"sog": normalized["sog"],
"cog": normalized["cog"],
"heading": normalized["heading"],
"_updated": now,
"source": "aishub",
}
)
if normalized["name"] and normalized["name"] != "UNKNOWN":
vessel["name"] = normalized["name"]
if normalized["callsign"]:
vessel["callsign"] = normalized["callsign"]
if normalized["destination"]:
vessel["destination"] = normalized["destination"]
if normalized["imo"]:
vessel["imo"] = normalized["imo"]
if normalized["ais_type_code"]:
vessel["ais_type_code"] = normalized["ais_type_code"]
vessel["type"] = classify_vessel(normalized["ais_type_code"], mmsi)
if not vessel.get("country"):
vessel["country"] = get_country_from_mmsi(mmsi)
_record_vessel_trail_locked(
mmsi,
normalized["lat"],
normalized["lng"],
normalized["sog"],
now,
)
count += 1
if count:
logger.info(
"AISHub fallback: merged %d vessels (poll interval %d min)",
count,
aishub_poll_interval_minutes(),
)
return count
@@ -1383,10 +1383,21 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
This is a resilience fallback for local/Windows runs where nuforc.org is
Cloudflare-gated and the Mapbox token is not configured. It is not as fresh
as the live NUFORC AJAX feed, but it keeps the layer visible and cached.
Date-cutoff guard: the kcimc/NUFORC HF dataset is a static snapshot whose
maintainer refreshes it sporadically. Without a cutoff, sorting by
occurred-desc and taking the top N rows returns whatever the mirror's
newest rows happen to be which can be years old if the snapshot is
stale. We apply the same ``_NUFORC_RECENT_DAYS`` window the live path
uses (60 days). If the HF mirror has nothing inside the window we return
``[]`` rather than silently serving 3-year-old "newest" rows.
"""
from services.fetchers.nuforc_enrichment import _HF_CSV_URL, _parse_date
from services.geocode_validate import coord_in_country
cutoff_dt = datetime.utcnow() - timedelta(days=_NUFORC_RECENT_DAYS)
cutoff_str = cutoff_dt.strftime("%Y-%m-%d")
try:
response = fetch_with_curl(_HF_CSV_URL, timeout=180, follow_redirects=True)
if not response or response.status_code != 200:
@@ -1400,6 +1411,7 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
return []
candidates: list[dict] = []
stale_rows_dropped = 0
try:
reader = csv.DictReader(io.StringIO(response.text))
for row in reader:
@@ -1410,6 +1422,9 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
)
if not occurred:
continue
if occurred < cutoff_str:
stale_rows_dropped += 1
continue
raw_location = _normalize_uap_location(
row.get("Location", "")
or row.get("City", "")
@@ -1444,6 +1459,19 @@ def _build_uap_sightings_from_hf_mirror() -> list[dict]:
logger.warning("UAP sightings: HF fallback parse failed: %s", e)
return []
if not candidates:
# HF mirror returned rows, but none inside the rolling window. This is
# the smoking gun for "the public HF dataset hasn't been refreshed in
# years" — log loudly so the operator sees it instead of guessing.
logger.error(
"UAP sightings: HF fallback yielded 0 rows within last %d days "
"(dropped %d stale rows). HF mirror is likely stale; the layer "
"will be empty until the live NUFORC path recovers.",
_NUFORC_RECENT_DAYS,
stale_rows_dropped,
)
return []
candidates.sort(key=lambda row: (row["occurred"], row["posted"], row["id"]), reverse=True)
candidates = candidates[:_NUFORC_HF_FALLBACK_LIMIT]
@@ -1515,13 +1543,29 @@ def fetch_uap_sightings(*, force_refresh: bool = False):
sightings = _load_nuforc_sightings_cache(force_refresh=force_refresh)
if sightings is None:
live_error: Exception | None = None
try:
sightings = _build_recent_uap_sightings()
except Exception as e:
live_error = e
logger.warning("UAP sightings: live NUFORC rebuild failed, using fallback: %s", e)
sightings = _build_uap_sightings_from_hf_mirror()
if sightings:
_save_nuforc_sightings_cache(sightings)
elif live_error is not None:
# Both paths failed: live raised AND HF fallback returned empty
# (either the HF mirror is stale beyond the cutoff or the network
# is gone entirely). The previous code silently set the layer to
# ``[]`` and kept marking it fresh; that masked the failure for
# days. Surface it via assert_canary so the health registry shows
# the layer as broken instead of "fresh and empty".
from services.slo import assert_canary
assert_canary("uap_sightings", 0)
logger.error(
"UAP sightings: both live NUFORC and HF fallback produced 0 "
"rows; layer is unavailable. Live error: %s",
live_error,
)
with _data_lock:
latest_data["uap_sightings"] = sightings or []
@@ -0,0 +1,148 @@
"""Per-aircraft observation tracking for cumulative fuel/CO2 estimates.
Background
----------
The pre-existing emissions enrichment attached a *rate* to each flight
(GPH and kg/hr) based on aircraft model. Users reasonably wanted the
running total: how much fuel HAS this plane burned since we started
seeing it? Multiplying the rate by elapsed observation time gets us
there, but it requires somewhere to remember "when did this icao24
first appear on our radar?"
Why this lives outside ``flight_trails``
----------------------------------------
``flight_trails`` is sized and pruned aggressively for map rendering
(5-minute TTL for untracked aircraft, 200 trail points max). That's
wrong for cumulative burn: if a plane has been airborne 2 hours but
its trail was pruned 30 min in, the "first trail point" timestamp is
30 min ago, not 2h ago. Worse, when the trail expires and re-creates,
the cumulative counter would reset mid-flight.
This module tracks observation lifecycle separately:
* When a hex is first observed: start a new flight session.
* While observed regularly (gap < ``REOPEN_GAP_S``): keep accumulating.
* When unseen for longer than ``REOPEN_GAP_S``: treat next sighting as
a new session (the plane landed and took off again, or it's a
different leg). Reset ``first_seen_at``.
* Stale sessions are pruned every ``PRUNE_INTERVAL_S`` so memory stays
bounded.
The user explicitly asked for this counting semantic: "as soon as a
plane appears there should be a counter that keeps a running count of
the fuel being burned... If there is no estimate take off time then it
can just be from the time the server starts to keep a log of whats in
the air."
"""
from __future__ import annotations
import threading
import time
# Gap between sightings that resets the session. ADS-B refreshes the
# whole aircraft list every minute or two, so anything over a few
# minutes means the plane left our coverage window (landed, transit
# through dead zone, etc). 15 minutes is conservative.
REOPEN_GAP_S = 15 * 60
# Don't accumulate runaway memory: drop entries unseen for an hour.
PRUNE_AFTER_S = 60 * 60
# Cap on accumulated airtime per session so a single bug elsewhere
# (e.g. ts clock skew) can't produce comically large numbers.
MAX_SESSION_SECONDS = 24 * 3600 # 24h — longest realistic civilian leg
_observations: dict[str, dict[str, float]] = {}
_lock = threading.Lock()
_last_prune_at = 0.0
def record_observation(icao_hex: str, *, now: float | None = None) -> int:
"""Record a sighting of ``icao_hex`` and return airtime so far (seconds).
Returns 0 for the first-ever sighting (no elapsed time yet) or when
``icao_hex`` is falsy. The caller can multiply the returned seconds
by ``rate_per_hour / 3600`` to get cumulative consumption.
"""
if not icao_hex:
return 0
key = str(icao_hex).strip().lower()
if not key:
return 0
current = float(now if now is not None else time.time())
with _lock:
entry = _observations.get(key)
if entry is None:
_observations[key] = {"first_seen_at": current, "last_seen_at": current}
return 0
# Use explicit ``is None`` checks instead of ``or`` short-circuit:
# ``0.0`` is a legitimate timestamp value (e.g. test fixtures
# seeding a far-past first_seen_at to exercise the clamp) but
# ``0.0 or fallback`` collapses to ``fallback`` because 0.0 is
# falsy. Bit me on my own test — leaving the safer form here.
last_raw = entry.get("last_seen_at")
last_seen = float(last_raw) if last_raw is not None else current
gap = current - last_seen
if gap > REOPEN_GAP_S:
# Treat as a new flight session — the plane landed/disappeared
# long enough that the prior cumulative count is no longer
# the same flight.
_observations[key] = {"first_seen_at": current, "last_seen_at": current}
return 0
first_raw = entry.get("first_seen_at")
first = float(first_raw) if first_raw is not None else current
# Clamp absurd values from clock skew or bad input.
elapsed = max(0, min(int(current - first), MAX_SESSION_SECONDS))
entry["last_seen_at"] = current
return elapsed
def prune(*, now: float | None = None) -> int:
"""Drop entries we haven't seen in ``PRUNE_AFTER_S`` seconds.
Returns number of entries dropped. Safe to call from a scheduler tick;
cheap (single dict scan) so cadence doesn't matter much.
"""
current = float(now if now is not None else time.time())
dropped = 0
with _lock:
stale_keys = []
for k, v in _observations.items():
last_raw = v.get("last_seen_at")
last = float(last_raw) if last_raw is not None else 0.0
if current - last > PRUNE_AFTER_S:
stale_keys.append(k)
for k in stale_keys:
del _observations[k]
dropped += 1
return dropped
def get_session_seconds(icao_hex: str, *, now: float | None = None) -> int:
"""Read-only accessor: airtime for a known icao without bumping last-seen.
Used by tests and external consumers (e.g. when rendering a snapshot
of all in-flight aircraft, you want the current value, not to update
last_seen_at as a side effect).
"""
if not icao_hex:
return 0
key = str(icao_hex).strip().lower()
with _lock:
entry = _observations.get(key)
if entry is None:
return 0
current = float(now if now is not None else time.time())
first_raw = entry.get("first_seen_at")
first = float(first_raw) if first_raw is not None else current
return max(0, min(int(current - first), MAX_SESSION_SECONDS))
def _reset_for_tests() -> None:
"""Drop all observations. Test helper only."""
with _lock:
_observations.clear()
+123 -50
View File
@@ -17,6 +17,7 @@ from services.network_utils import fetch_with_curl
from services.fetchers._store import latest_data, _data_lock, _mark_fresh
from services.fetchers.plane_alert import enrich_with_plane_alert, enrich_with_tracked_names
from services.fetchers.emissions import get_emissions_info
from services.fetchers.flight_observations import record_observation as _record_flight_observation
from services.fetchers.retry import with_retry
from services.fetchers.route_database import lookup_route
from services.fetchers.aircraft_database import lookup_aircraft_type
@@ -29,6 +30,88 @@ _RE_AIRLINE_CODE_1 = re.compile(r"^([A-Z]{3})\d")
_RE_AIRLINE_CODE_2 = re.compile(r"^([A-Z]{3})[A-Z\d]")
def detect_gps_jamming_zones(
raw_flights: list[dict],
*,
min_aircraft: int | None = None,
min_ratio: float | None = None,
nacp_threshold: int | None = None,
) -> list[dict]:
"""Detect GPS interference zones from a snapshot of raw ADS-B aircraft.
Methodology mirrors GPSJam.org / Flightradar24: bin aircraft into 1°x1°
grid cells, flag cells where the fraction of aircraft reporting degraded
NACp clears a threshold.
Inputs
------
raw_flights:
Iterable of dicts. Each item is expected to carry ``lat``, ``lng``
(or ``lon``), and ``nac_p``. Records missing position OR missing
``nac_p`` entirely (typical for OpenSky-sourced flights) are
skipped absence-of-data isn't evidence of anything.
nac_p == 0 IS counted as degraded. Pre-fix code skipped it on the theory
that "0 = old transponder, never computed accuracy." That's only half
right: modern Mode-S Enhanced Surveillance transponders also fall back
to nac_p=0 when they lose GPS lock entirely which is exactly the
jamming signature we're trying to detect. Filtering 0 out was discarding
the strongest evidence.
Denoising:
1. Require ``min_aircraft`` per grid cell for statistical validity.
2. Subtract 1 from degraded count per cell (GPSJam's technique) so
a single quirky transponder can't flag an entire zone.
3. Require ratio ``adjusted_degraded / total > min_ratio``.
All thresholds default to the module-level constants but can be
overridden for testing.
"""
min_aircraft = GPS_JAMMING_MIN_AIRCRAFT if min_aircraft is None else int(min_aircraft)
min_ratio = GPS_JAMMING_MIN_RATIO if min_ratio is None else float(min_ratio)
nacp_threshold = (
GPS_JAMMING_NACP_THRESHOLD if nacp_threshold is None else int(nacp_threshold)
)
jamming_grid: dict[str, dict[str, int]] = {}
for rf in raw_flights or []:
rlat = rf.get("lat")
rlng = rf.get("lng") if rf.get("lng") is not None else rf.get("lon")
if rlat is None or rlng is None:
continue
nacp = rf.get("nac_p")
if nacp is None:
continue
grid_key = f"{int(rlat)},{int(rlng)}"
cell = jamming_grid.setdefault(grid_key, {"degraded": 0, "total": 0})
cell["total"] += 1
if nacp < nacp_threshold:
cell["degraded"] += 1
jamming_zones: list[dict] = []
for gk, counts in jamming_grid.items():
if counts["total"] < min_aircraft:
continue
adjusted_degraded = max(counts["degraded"] - 1, 0)
if adjusted_degraded == 0:
continue
ratio = adjusted_degraded / counts["total"]
if ratio > min_ratio:
lat_i, lng_i = gk.split(",")
severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
jamming_zones.append(
{
"lat": int(lat_i) + 0.5,
"lng": int(lng_i) + 0.5,
"severity": severity,
"ratio": round(ratio, 2),
"degraded": counts["degraded"],
"total": counts["total"],
}
)
return jamming_zones
# ---------------------------------------------------------------------------
# OpenSky Network API Client (OAuth2)
# ---------------------------------------------------------------------------
@@ -459,6 +542,18 @@ def _classify_and_publish(all_adsb_flights):
ac_category = "heli" if model_upper in _HELI_TYPES_BACKEND else "plane"
# Source attribution: prefer the explicit ``source`` tag stamped
# at fetch time (adsb.lol, OpenSky). If absent, fall back to the
# legacy ``supplemental_source`` (airplanes.live, adsb.fi) so
# supplementals are still attributed without changing their
# tagger. Final fallback "adsb.lol" preserves prior behavior for
# any caller that synthesizes records without going through one
# of our fetchers (e.g. tests).
source = (
f.get("source")
or f.get("supplemental_source")
or "adsb.lol"
)
flights.append(
{
"callsign": flight_str,
@@ -480,6 +575,7 @@ def _classify_and_publish(all_adsb_flights):
"airline_code": airline_code,
"aircraft_category": ac_category,
"nac_p": f.get("nac_p"),
"source": source,
}
)
except (ValueError, TypeError, KeyError, AttributeError) as loop_e:
@@ -506,6 +602,22 @@ def _classify_and_publish(all_adsb_flights):
if model:
emi = get_emissions_info(model)
if emi:
# Cumulative fuel/CO2: multiply the per-hour rate by how
# long we've been observing this airframe. Users want to
# see the *amount* burned, not just the rate. If we've
# never seen this hex before, observed_seconds is 0 and
# the cumulative values are 0 until the next refresh —
# the rate is still useful info on its own.
observed_seconds = _record_flight_observation(
f.get("icao24") or ""
)
elapsed_h = observed_seconds / 3600.0
emi = {
**emi,
"observed_seconds": observed_seconds,
"fuel_gallons_burned": round(emi["fuel_gph"] * elapsed_h, 1),
"co2_kg_emitted": round(emi["co2_kg_per_hour"] * elapsed_h, 1),
}
f["emissions"] = emi
callsign = f.get("callsign", "").strip().upper()
@@ -724,56 +836,8 @@ def _classify_and_publish(all_adsb_flights):
latest_data["military_flights"] = military_snapshot
# --- GPS Jamming Detection ---
# Uses NACp (Navigation Accuracy Category Position) from ADS-B to infer
# GPS interference zones, similar to GPSJam.org / Flightradar24.
# NACp < 8 = position accuracy worse than the FAA-mandated 0.05 NM.
#
# Denoising (to suppress false positives from old GA transponders):
# 1. Skip nac_p == 0 ("unknown accuracy") — old transponders that never
# computed accuracy, NOT evidence of jamming. Real jamming shows 1-7.
# 2. Require minimum aircraft per grid cell for statistical validity.
# 3. Subtract 1 from degraded count per cell (GPSJam's technique) so a
# single quirky transponder can't flag an entire zone.
# 4. Require the adjusted ratio to exceed the threshold.
try:
jamming_grid = {}
raw_flights = raw_flights_snapshot
for rf in raw_flights:
rlat = rf.get("lat")
rlng = rf.get("lng") or rf.get("lon")
if rlat is None or rlng is None:
continue
nacp = rf.get("nac_p")
if nacp is None or nacp == 0:
continue
grid_key = f"{int(rlat)},{int(rlng)}"
if grid_key not in jamming_grid:
jamming_grid[grid_key] = {"degraded": 0, "total": 0}
jamming_grid[grid_key]["total"] += 1
if nacp < GPS_JAMMING_NACP_THRESHOLD:
jamming_grid[grid_key]["degraded"] += 1
jamming_zones = []
for gk, counts in jamming_grid.items():
if counts["total"] < GPS_JAMMING_MIN_AIRCRAFT:
continue
adjusted_degraded = max(counts["degraded"] - 1, 0)
if adjusted_degraded == 0:
continue
ratio = adjusted_degraded / counts["total"]
if ratio > GPS_JAMMING_MIN_RATIO:
lat_i, lng_i = gk.split(",")
severity = "low" if ratio < 0.5 else "medium" if ratio < 0.75 else "high"
jamming_zones.append(
{
"lat": int(lat_i) + 0.5,
"lng": int(lng_i) + 0.5,
"severity": severity,
"ratio": round(ratio, 2),
"degraded": counts["degraded"],
"total": counts["total"],
}
)
jamming_zones = detect_gps_jamming_zones(raw_flights_snapshot)
with _data_lock:
latest_data["gps_jamming"] = jamming_zones
if jamming_zones:
@@ -849,7 +913,15 @@ def _fetch_adsb_lol_regions():
res = fetch_with_curl(url, timeout=10)
if res.status_code == 200:
data = res.json()
return data.get("ac", [])
aircraft = data.get("ac", [])
# Stamp the source at the fetch site so attribution survives
# the OpenSky/supplemental dedupe-by-hex merge downstream.
# Previously adsb.lol records carried no marker while OpenSky
# records got ``is_opensky: True`` — which made flight tooltips
# look like everything came from OpenSky.
for a in aircraft:
a["source"] = "adsb.lol"
return aircraft
except (
requests.RequestException,
ConnectionError,
@@ -932,6 +1004,7 @@ def _enrich_with_opensky_and_supplemental(adsb_flights):
"gs": (s[9] * 1.94384) if s[9] else 0,
"t": "Unknown",
"is_opensky": True,
"source": "OpenSky",
}
)
elif os_res.status_code == 429:
+18 -1
View File
@@ -7,6 +7,7 @@ import requests
from services.network_utils import fetch_with_curl
from services.fetchers._store import latest_data, _data_lock, _mark_fresh
from services.fetchers.emissions import get_emissions_info
from services.fetchers.flight_observations import record_observation as _record_flight_observation
from services.fetchers.plane_alert import enrich_with_plane_alert
logger = logging.getLogger("services.data_fetcher")
@@ -171,6 +172,7 @@ def fetch_military_flights():
h = a.get("hex", "").lower()
if h and h not in seen_hex:
seen_hex.add(h)
a["source"] = "adsb.lol"
all_mil_ac.append(a)
except Exception as e:
logger.warning(f"adsb.lol mil fetch failed: {e}")
@@ -182,6 +184,7 @@ def fetch_military_flights():
h = a.get("hex", "").lower()
if h and h not in seen_hex:
seen_hex.add(h)
a["source"] = "airplanes.live"
all_mil_ac.append(a)
logger.info(f"airplanes.live mil: +{len(resp2.json().get('ac', []))} raw, {len(all_mil_ac)} total unique")
except Exception as e:
@@ -234,6 +237,7 @@ def fetch_military_flights():
"registration": f.get("r", "N/A"),
"icao24": icao_hex,
"squawk": f.get("squawk", ""),
"source": f.get("source") or "adsb.lol",
})
continue
@@ -258,7 +262,8 @@ def fetch_military_flights():
"model": f.get("t", "Unknown"),
"icao24": icao_hex,
"speed_knots": speed_knots,
"squawk": f.get("squawk", "")
"squawk": f.get("squawk", ""),
"source": f.get("source") or "adsb.lol",
})
except Exception as loop_e:
logger.error(f"Mil flight interpolation error: {loop_e}")
@@ -296,6 +301,18 @@ def fetch_military_flights():
if model:
emissions = get_emissions_info(model)
if emissions:
# Cumulative fuel/CO2 since first observation — mirrors
# the civilian path in flights._classify_and_publish.
observed_seconds = _record_flight_observation(
mf.get("icao24") or ""
)
elapsed_h = observed_seconds / 3600.0
emissions = {
**emissions,
"observed_seconds": observed_seconds,
"fuel_gallons_burned": round(emissions["fuel_gph"] * elapsed_h, 1),
"co2_kg_emitted": round(emissions["co2_kg_per_hour"] * elapsed_h, 1),
}
mf["emissions"] = emissions
if mf.get("alert_category"):
mf["type"] = "tracked_flight"
+293
View File
@@ -317,6 +317,39 @@ class DMRelay:
def _self_mailbox_limit(self) -> int:
return max(1, int(self._settings().MESH_DM_SELF_MAILBOX_LIMIT))
def _per_sender_pending_limit(self) -> int:
"""Anti-spam cap on UNACKED messages a single sender can have parked
in a single recipient mailbox at any one time. See ``config.py``
``MESH_DM_PENDING_PER_SENDER_LIMIT`` for the threat model this
rule is enforced both at ``deposit`` (local) and at
``accept_replica`` (peer push acceptance), making it a network
rule rather than a client-side honor system."""
try:
limit = int(getattr(self._settings(), "MESH_DM_PENDING_PER_SENDER_LIMIT", 2) or 2)
except (TypeError, ValueError):
limit = 2
return max(1, limit)
def _per_sender_pending_count(
self,
*,
mailbox_key: str,
sender_block_ref: str,
) -> int:
"""Count UNACKED messages from ``sender_block_ref`` currently parked
in ``mailbox_key``. Caller already holds ``self._lock``.
Messages that have been claimed/acked are removed from the mailbox
list (see ``claim_message_ids``), so anything still here is by
definition unacked. We count by exact ``sender_block_ref`` match
that's the per-pair sender identity used for blocking too, so
the cap is naturally per-(sender, recipient).
"""
if not mailbox_key or not sender_block_ref:
return 0
messages = self._mailboxes.get(mailbox_key, [])
return sum(1 for m in messages if m.sender_block_ref == sender_block_ref)
def _nonce_ttl_seconds(self) -> int:
return max(30, int(self._settings().MESH_DM_NONCE_TTL_S))
@@ -1515,6 +1548,29 @@ class DMRelay:
if len(self._mailboxes[mailbox_key]) >= self._mailbox_limit_for_class(delivery_class):
metrics_inc("dm_drop_full")
return {"ok": False, "detail": "Recipient mailbox full"}
# Anti-spam: per-(sender, recipient) cap on unacked messages.
# A sender who already has the configured number of messages
# parked in this mailbox can't deposit more until the recipient
# pulls (acks) at least one. The same cap is re-enforced on
# inbound replication in ``accept_replica`` so this rule isn't
# bypassable by patching out the local check on a hostile
# sender's relay — see config.py
# MESH_DM_PENDING_PER_SENDER_LIMIT for the threat model.
per_sender_limit = self._per_sender_pending_limit()
pending = self._per_sender_pending_count(
mailbox_key=mailbox_key,
sender_block_ref=sender_block_ref,
)
if pending >= per_sender_limit:
metrics_inc("dm_drop_per_sender_cap")
return {
"ok": False,
"detail": (
f"Recipient already has {pending} unread message"
f"{'s' if pending != 1 else ''} from you. Wait for "
"them to read your messages before sending more."
),
}
if not msg_id:
msg_id = f"dm_{int(time.time() * 1000)}_{secrets.token_hex(6)}"
elif any(m.msg_id == msg_id for m in self._mailboxes[mailbox_key]):
@@ -1539,8 +1595,245 @@ class DMRelay:
)
self._stats["messages_in_memory"] = sum(len(v) for v in self._mailboxes.values())
self._save()
# Cross-node mailbox replication: push the freshly-stored
# envelope to every authenticated relay peer so the recipient
# can log into ANY node and find their messages. The push is
# async (fire-and-forget thread) so deposit() returns
# immediately — slow Tor peers can't block the sender's UX.
# Each receiving peer re-enforces the per-sender cap on
# acceptance, so hostile relays can't widen the cap.
try:
envelope_for_push = self.envelope_for_replication(
mailbox_key=mailbox_key, msg_id=msg_id,
)
if envelope_for_push:
self._replicate_envelope_to_peers_async(
envelope=envelope_for_push,
)
except Exception:
metrics_inc("dm_replication_push_error")
return {"ok": True, "msg_id": msg_id}
def accept_replica(
self,
*,
envelope: dict[str, Any],
originating_peer_url: str = "",
) -> dict[str, Any]:
"""Receive a DM envelope replicated from a peer relay.
Cross-node mailbox replication entry point. When a sender's local
relay accepts a ``deposit`` and pushes the envelope to
``MESH_RELAY_PEERS`` (so the recipient can log into any peer
node and find their messages), each receiving peer calls
``accept_replica`` to ingest it.
The per-(sender, recipient) cap is re-enforced HERE. That's what
makes the rule a NETWORK rule rather than a client-side honor
system: a hostile sender who patches out the local ``deposit``
check still can't get a 3rd unacked message to spread, because
every honest peer enforces the same cap on inbound replicas.
Result: hostile relays can hold extras locally, but those extras
never reach any node a legitimate recipient is polling from.
Returns the same shape as ``deposit`` so the calling endpoint can
forward the result back to the originating peer.
"""
if not isinstance(envelope, dict):
return {"ok": False, "detail": "envelope must be an object"}
msg_id = str(envelope.get("msg_id", "") or "").strip()
mailbox_key = str(envelope.get("mailbox_key", "") or "").strip()
sender_block_ref = str(envelope.get("sender_block_ref", "") or "").strip()
ciphertext = str(envelope.get("ciphertext", "") or "")
if not msg_id or not mailbox_key or not sender_block_ref or not ciphertext:
return {"ok": False, "detail": "envelope missing required fields"}
with self._lock:
self._refresh_from_shared_relay()
self._cleanup_expired()
# Idempotent — if we already hold this exact msg_id, the
# replication round-tripped or a peer pushed the same
# envelope through multiple paths. Accept silently.
if any(m.msg_id == msg_id for m in self._mailboxes.get(mailbox_key, [])):
metrics_inc("dm_replica_duplicate")
return {"ok": True, "msg_id": msg_id, "duplicate": True}
# Same per-class cap as the deposit path — defense in depth
# against a peer that wraps a "deposit" as a "replica" to
# bypass the class limit.
delivery_class = str(envelope.get("delivery_class", "") or "")
if delivery_class in ("request", "shared", "self"):
class_limit = self._mailbox_limit_for_class(delivery_class)
else:
class_limit = self._shared_mailbox_limit()
if len(self._mailboxes.get(mailbox_key, [])) >= class_limit:
metrics_inc("dm_replica_drop_full")
return {"ok": False, "detail": "Recipient mailbox full"}
# THE network rule: per-(sender, recipient) anti-spam cap.
per_sender_limit = self._per_sender_pending_limit()
pending = self._per_sender_pending_count(
mailbox_key=mailbox_key,
sender_block_ref=sender_block_ref,
)
if pending >= per_sender_limit:
metrics_inc("dm_replica_drop_per_sender_cap")
# Returning a structured rejection — the sender's relay
# learns its envelope was rejected by an honest peer and
# can stop trying to push it.
return {
"ok": False,
"detail": (
"Per-sender cap reached on this relay; refusing replica"
),
"cap_violation": True,
"pending": pending,
"limit": per_sender_limit,
}
# Accept the replica into the local mailbox.
self._mailboxes[mailbox_key].append(
DMMessage(
sender_id=str(envelope.get("sender_id", "") or ""),
ciphertext=ciphertext,
timestamp=float(envelope.get("timestamp", time.time()) or time.time()),
msg_id=msg_id,
delivery_class=str(envelope.get("delivery_class", "shared") or "shared"),
sender_seal=str(envelope.get("sender_seal", "") or ""),
relay_salt=str(envelope.get("relay_salt", "") or ""),
sender_block_ref=sender_block_ref,
payload_format=str(envelope.get("payload_format", "dm1") or "dm1"),
session_welcome=str(envelope.get("session_welcome", "") or ""),
)
)
self._stats["messages_in_memory"] = sum(len(v) for v in self._mailboxes.values())
self._save()
metrics_inc("dm_replica_accepted")
return {"ok": True, "msg_id": msg_id}
def _replicate_envelope_to_peers_async(
self,
*,
envelope: dict[str, Any],
) -> None:
"""Push an outbound DM envelope to every authenticated relay peer.
Fire-and-forget: spawned in a background thread so ``deposit``
returns to the caller immediately. Per-peer errors are logged
and swallowed the sender's UX must not block on slow Tor
peers, and a peer that's down today gets the next message
whenever it comes back. Inbound recipient polling from a healthy
peer keeps the system functional during peer failures.
Each peer is authed with the existing per-peer HMAC pattern
(#256) — same headers and key resolver gate-message replication
uses, so a hostile node that doesn't know any peer's HMAC key
can't impersonate a legitimate relay.
"""
import threading
def _do_push():
try:
import hashlib
import hmac
import requests as _requests
from services.mesh.mesh_crypto import (
normalize_peer_url,
resolve_peer_key_for_url,
)
from services.mesh.mesh_router import (
authenticated_push_peer_urls,
)
peers = authenticated_push_peer_urls()
if not peers:
return
payload = json.dumps(
{"envelope": envelope},
separators=(",", ":"),
ensure_ascii=False,
).encode("utf-8")
timeout = max(
1,
int(getattr(self._settings(), "MESH_RELAY_PUSH_TIMEOUT_S", 10) or 10),
)
for peer_url in peers:
try:
normalized = normalize_peer_url(peer_url)
headers = {"Content-Type": "application/json"}
peer_key = resolve_peer_key_for_url(normalized)
if peer_key:
headers["X-Peer-Url"] = normalized
headers["X-Peer-HMAC"] = hmac.new(
peer_key, payload, hashlib.sha256
).hexdigest()
url = f"{peer_url}/api/mesh/dm/replicate-envelope"
resp = _requests.post(
url, data=payload, timeout=timeout, headers=headers,
)
if resp.status_code == 200:
metrics_inc("dm_replication_push_ok")
else:
# 4xx including the structured cap_violation
# rejection from accept_replica — sender's
# relay learns and stops retrying this msg_id.
metrics_inc("dm_replication_push_rejected")
except Exception:
# Per-peer failure is non-fatal — log to metrics
# but don't break the loop. Other peers and a
# future retry can still propagate the envelope.
metrics_inc("dm_replication_push_error")
continue
except Exception:
# Outer guard — never let replication errors propagate
# back to the sender's deposit() caller.
metrics_inc("dm_replication_push_error")
thread = threading.Thread(
target=_do_push,
name="dm-replicate-push",
daemon=True,
)
thread.start()
def envelope_for_replication(
self,
*,
mailbox_key: str,
msg_id: str,
) -> dict[str, Any] | None:
"""Return the wire-form envelope for a stored message, suitable
for POSTing to a peer relay's replicate-envelope endpoint.
Returns ``None`` if the message isn't in the mailbox (already
acked, expired, never existed). The caller holds the
responsibility for transport security (Tor SOCKS for .onion
peers, per-peer HMAC) and for not leaking the envelope to
clearnet peers when private transport is required.
"""
with self._lock:
for m in self._mailboxes.get(mailbox_key, []):
if m.msg_id == msg_id:
return {
"msg_id": m.msg_id,
"mailbox_key": mailbox_key,
"sender_id": m.sender_id,
"sender_block_ref": m.sender_block_ref,
"sender_seal": m.sender_seal,
"ciphertext": m.ciphertext,
"timestamp": m.timestamp,
"delivery_class": m.delivery_class,
"relay_salt": m.relay_salt,
"payload_format": m.payload_format,
"session_welcome": m.session_welcome,
}
return None
def is_blocked(self, recipient_id: str, sender_id: str) -> bool:
with self._lock:
self._refresh_from_shared_relay()
@@ -2,10 +2,64 @@ from __future__ import annotations
import time
from dataclasses import asdict, dataclass
from email.utils import parsedate_to_datetime
from datetime import timezone
from services.mesh.mesh_peer_store import PeerRecord
class PeerSyncRateLimited(Exception):
"""Upstream peer returned HTTP 429 — Too Many Requests.
Carries the ``Retry-After`` header value (parsed to seconds) so
the caller can pass it to ``finish_sync(retry_after_s=...)`` and
actually wait that long instead of hammering the upstream every
60s and keeping its rate-limit bucket full.
``retry_after_s`` is 0 when the upstream didn't provide a header.
Caller should still apply the exponential backoff in that case.
"""
def __init__(self, message: str, retry_after_s: int = 0, status: int = 429):
super().__init__(message)
self.retry_after_s = max(0, int(retry_after_s or 0))
self.status = int(status or 429)
def parse_retry_after_header(header_value: str, *, now: float | None = None) -> int:
"""Parse the ``Retry-After`` HTTP header.
Two valid forms per RFC 7231 §7.1.3:
* Delay-seconds: a non-negative integer (e.g. ``Retry-After: 120``)
* HTTP-date: an absolute time (e.g. ``Retry-After: Wed, 21 Oct 2026 07:28:00 GMT``)
Returns the wait in **seconds from now**. Unparseable / empty headers
return 0 (caller falls back to exponential backoff). Clamped at a
sane upper bound (1 hour) so a typo'd or hostile peer can't pin us
silent for days.
"""
value = str(header_value or "").strip()
if not value:
return 0
upper_bound = 3600 # never trust a peer to silence us > 1h
# Form 1: pure integer seconds.
if value.isdigit():
return min(max(0, int(value)), upper_bound)
# Form 2: HTTP-date.
try:
target = parsedate_to_datetime(value)
if target is None:
return 0
if target.tzinfo is None:
target = target.replace(tzinfo=timezone.utc)
current = float(now if now is not None else time.time())
delta = int(target.timestamp() - current)
return min(max(0, delta), upper_bound)
except (TypeError, ValueError):
return 0
@dataclass(frozen=True)
class SyncWorkerState:
last_sync_started_at: int = 0
@@ -72,6 +126,59 @@ def begin_sync(
)
def _failure_backoff_seconds(
*,
base_backoff_s: int,
consecutive_failures: int,
retry_after_s: int,
cap_s: int = 1800,
) -> int:
"""Compute the next-attempt delay after a failed sync.
Two inputs combine:
* ``retry_after_s`` when an upstream peer answered HTTP 429
with a ``Retry-After`` header, we honor it exactly. Continuing
to hammer the upstream every 60s is the bug this fix exists to
close: it keeps the upstream's rate-limit bucket full
indefinitely and no sync ever lands.
* Exponential growth on ``consecutive_failures`` even without an
explicit Retry-After, repeated failures should slow us down. The
first failure waits ``base`` (preserves pre-fix behavior for
one-off blips). Each subsequent failure doubles the wait, capped
to ``cap_s`` (default 30 minutes). With base=60 and cap=1800,
the schedule is 60s 120s 240s 480s 960s 1800s
1800s .
The actual delay is the MAX of the two whichever asks for more
patience wins. ``retry_after_s == 0`` (no header) falls back to
pure exponential. An aggressive ``Retry-After`` (say 600s while
we're only at 1 failure) wins over the exponential ladder.
"""
base = max(0, int(base_backoff_s or 0))
failures = max(0, int(consecutive_failures or 0))
cap = max(0, int(cap_s or 0))
retry_after = max(0, int(retry_after_s or 0))
# ``cap_s=0`` explicitly disables the exponential ladder entirely
# — operators who want the pre-fix "honor Retry-After only" behavior
# can set this. The default cap of 1800s is what saturates the
# ladder at the 5th-6th failure for base=60.
if cap == 0:
return retry_after
# 2^(failures-1) — so failure #1 = base (preserves the pre-fix
# default for transient blips), failure #2 = 2*base, etc. Cap on
# the exponent (16) is defense against integer overflow on a
# hostile or very large failures counter.
if base > 0 and failures > 0:
exponent = min(max(0, failures - 1), 16)
grown = base * (2 ** exponent)
else:
grown = 0
exponential = min(max(0, grown), cap)
return max(exponential, retry_after)
def finish_sync(
state: SyncWorkerState,
*,
@@ -83,7 +190,26 @@ def finish_sync(
now: float | None = None,
interval_s: int = 300,
failure_backoff_s: int = 60,
retry_after_s: int = 0,
failure_backoff_cap_s: int = 1800,
) -> SyncWorkerState:
"""Finalise a sync attempt and compute when the next one should run.
New args (added for the 429 retry storm fix):
* ``retry_after_s`` if the peer responded with HTTP 429 + a
``Retry-After`` header, pass that value here. ``finish_sync``
will use ``max(exponential, retry_after_s)`` for the delay so
we never hammer a peer that asked us to back off.
* ``failure_backoff_cap_s`` upper bound on the exponential
ladder. Default 1800 (30 min) keeps a sync queue from going
silent for hours while still cutting the request rate to
something the upstream can absorb.
The pre-fix behavior (constant 60s on every failure) is recoverable
by passing ``failure_backoff_cap_s=0`` and ``retry_after_s=0``, but
there's no reason to.
"""
timestamp = int(now if now is not None else time.time())
if ok:
return SyncWorkerState(
@@ -99,17 +225,25 @@ def finish_sync(
consecutive_failures=0,
)
next_failures = state.consecutive_failures + 1
delay_s = _failure_backoff_seconds(
base_backoff_s=failure_backoff_s,
consecutive_failures=next_failures,
retry_after_s=retry_after_s,
cap_s=failure_backoff_cap_s,
)
return SyncWorkerState(
last_sync_started_at=state.last_sync_started_at,
last_sync_finished_at=timestamp,
last_sync_ok_at=state.last_sync_ok_at,
next_sync_due_at=timestamp + max(0, int(failure_backoff_s or 0)),
next_sync_due_at=timestamp + delay_s,
last_peer_url=peer_url or state.last_peer_url,
last_error=str(error or "").strip(),
last_outcome="fork" if fork_detected else "error",
current_head=current_head or state.current_head,
fork_detected=bool(fork_detected),
consecutive_failures=state.consecutive_failures + 1,
consecutive_failures=next_failures,
)
@@ -0,0 +1,261 @@
"""Infonet sync respects upstream HTTP 429 + applies exponential backoff.
Background
----------
Before this fix, ``finish_sync`` used a constant 60s ``failure_backoff_s``
regardless of how many consecutive failures preceded. When an upstream
peer (e.g. the seed onion) returned HTTP 429 "Too Many Requests", the
sync worker would:
1. Receive 429
2. Stringify the status into a generic ``ValueError``
3. Call ``finish_sync(error=str(exc))`` -- losing the status code
4. Schedule next attempt for ``now + 60s``
5. Retry. Upstream's rate-limit bucket is still full. 429 again. Loop.
Net effect: a node with one transient 429 would hammer the upstream
every 60s forever, keeping the bucket full and never recovering. This
is what kept the user's Infonet node from reaching the seed peer.
What the fix does
-----------------
* New typed exception ``PeerSyncRateLimited`` carries the parsed
``Retry-After`` value out of the HTTP layer.
* ``_sync_from_peer`` returns ``(ok, error, forked, retry_after_s)``
instead of the old 3-tuple.
* ``finish_sync`` honors ``retry_after_s`` AND applies exponential
backoff: ``delay = max(retry_after_s, base * 2^failures, cap=1800)``.
* ``parse_retry_after_header`` handles both RFC 7231 forms (delay
seconds, and HTTP-date).
These tests pin every part of the new contract.
"""
from __future__ import annotations
import time
import pytest
# ---------------------------------------------------------------------------
# parse_retry_after_header — both RFC 7231 forms + edge cases
# ---------------------------------------------------------------------------
class TestParseRetryAfter:
def test_integer_seconds(self):
from services.mesh.mesh_infonet_sync_support import parse_retry_after_header
assert parse_retry_after_header("120") == 120
assert parse_retry_after_header(" 30 ") == 30
assert parse_retry_after_header("0") == 0
def test_http_date(self):
"""RFC 7231 §7.1.3 explicitly allows ``Retry-After: <HTTP-date>``.
We compute seconds-from-now so callers can use the same field
regardless of which form the upstream chose."""
from services.mesh.mesh_infonet_sync_support import parse_retry_after_header
# Pin "now" so the test is deterministic.
now = 1_700_000_000.0 # 2023-11-14T22:13:20Z
# 300 seconds in the future, formatted per RFC 7231.
future = "Tue, 14 Nov 2023 22:18:20 GMT"
result = parse_retry_after_header(future, now=now)
assert 295 <= result <= 305, f"expected ~300s, got {result}"
def test_http_date_in_past_returns_zero(self):
from services.mesh.mesh_infonet_sync_support import parse_retry_after_header
now = 1_700_000_000.0
past = "Mon, 13 Nov 2023 00:00:00 GMT"
assert parse_retry_after_header(past, now=now) == 0
def test_empty_and_whitespace_return_zero(self):
from services.mesh.mesh_infonet_sync_support import parse_retry_after_header
assert parse_retry_after_header("") == 0
assert parse_retry_after_header(" ") == 0
def test_malformed_returns_zero(self):
from services.mesh.mesh_infonet_sync_support import parse_retry_after_header
assert parse_retry_after_header("not a header") == 0
assert parse_retry_after_header("xyz") == 0
def test_clamps_to_one_hour(self):
"""A hostile peer can't silence us for a week by claiming a
24h Retry-After. We cap at 1 hour."""
from services.mesh.mesh_infonet_sync_support import parse_retry_after_header
assert parse_retry_after_header("86400") == 3600 # 24h -> 1h
assert parse_retry_after_header("99999999") == 3600
def test_negative_returns_zero(self):
"""RFC 7231 says ``Retry-After`` is a non-negative integer;
leading-minus parses as a non-digit and yields 0 here."""
from services.mesh.mesh_infonet_sync_support import parse_retry_after_header
assert parse_retry_after_header("-10") == 0
# ---------------------------------------------------------------------------
# _failure_backoff_seconds — exponential growth, retry-after override, cap
# ---------------------------------------------------------------------------
class TestFailureBackoffSeconds:
def test_exponential_growth(self):
"""First failure uses the base (preserves pre-fix behavior
for one-off blips). Each subsequent failure doubles the wait,
capped at 1800s. With base=60: 60, 120, 240, 480, 960, 1800,
1800, 1800."""
from services.mesh.mesh_infonet_sync_support import _failure_backoff_seconds
delays = [
_failure_backoff_seconds(
base_backoff_s=60,
consecutive_failures=n,
retry_after_s=0,
cap_s=1800,
)
for n in range(1, 9)
]
assert delays == [60, 120, 240, 480, 960, 1800, 1800, 1800], delays
def test_retry_after_wins_when_larger(self):
"""If the upstream says ``Retry-After: 600`` but exponential
would only ask for 60s (one failure), we honor the upstream."""
from services.mesh.mesh_infonet_sync_support import _failure_backoff_seconds
assert _failure_backoff_seconds(
base_backoff_s=60,
consecutive_failures=1,
retry_after_s=600,
cap_s=1800,
) == 600
def test_exponential_wins_when_larger(self):
"""If exponential is asking for 1800s (6+ failures) but
upstream only sent ``Retry-After: 30``, we honor exponential.
The 30s was the upstream's view at one moment; our exponential
reflects sustained failure."""
from services.mesh.mesh_infonet_sync_support import _failure_backoff_seconds
result = _failure_backoff_seconds(
base_backoff_s=60,
consecutive_failures=7,
retry_after_s=30,
cap_s=1800,
)
assert result == 1800
def test_cap_zero_disables_exponential(self):
"""Operators who want pre-fix behavior can set cap=0; only the
upstream's Retry-After is respected. (Pre-fix had no
exponential growth at all.)"""
from services.mesh.mesh_infonet_sync_support import _failure_backoff_seconds
assert _failure_backoff_seconds(
base_backoff_s=60,
consecutive_failures=10,
retry_after_s=120,
cap_s=0,
) == 120
def test_zero_inputs_return_zero(self):
from services.mesh.mesh_infonet_sync_support import _failure_backoff_seconds
assert _failure_backoff_seconds(
base_backoff_s=0,
consecutive_failures=0,
retry_after_s=0,
) == 0
# ---------------------------------------------------------------------------
# finish_sync end-to-end — failure path with retry-after + growing counter
# ---------------------------------------------------------------------------
class TestFinishSyncBackoff:
def _state(self, **overrides):
from services.mesh.mesh_infonet_sync_support import SyncWorkerState
base = {
"last_sync_started_at": 0,
"last_sync_finished_at": 0,
"last_sync_ok_at": 0,
"next_sync_due_at": 0,
"last_peer_url": "",
"last_error": "",
"last_outcome": "idle",
"current_head": "",
"fork_detected": False,
"consecutive_failures": 0,
}
base.update(overrides)
return SyncWorkerState(**base)
def test_first_failure_uses_base_unchanged(self):
"""One failure means consecutive_failures becomes 1, which uses
``base * 2^0 = base``. Preserves the pre-fix behavior so a
single transient upstream blip doesn't suddenly take 2 minutes
to retry that change has to be earned by sustained failure."""
from services.mesh.mesh_infonet_sync_support import finish_sync
result = finish_sync(
self._state(),
ok=False,
error="some upstream blip",
now=1000.0,
failure_backoff_s=60,
)
assert result.consecutive_failures == 1
assert result.next_sync_due_at == 1000 + 60
assert result.last_error == "some upstream blip"
assert result.last_outcome == "error"
def test_consecutive_failures_grow_the_delay(self):
"""After 5 prior failures already in state, the next failure
sets consecutive=6 and uses the cap (1800s = 60 * 2^5)."""
from services.mesh.mesh_infonet_sync_support import finish_sync
result = finish_sync(
self._state(consecutive_failures=5),
ok=False,
error="HTTP 429",
now=2000.0,
failure_backoff_s=60,
)
assert result.consecutive_failures == 6
assert result.next_sync_due_at == 2000 + 1800
def test_retry_after_honored_at_low_failure_count(self):
"""When the upstream says ``Retry-After: 900`` but we'd
otherwise only wait 240s (4 failures = 60*2^3), wait 900s."""
from services.mesh.mesh_infonet_sync_support import finish_sync
result = finish_sync(
self._state(consecutive_failures=3),
ok=False,
error="HTTP 429",
now=5000.0,
failure_backoff_s=60,
retry_after_s=900,
)
assert result.consecutive_failures == 4
assert result.next_sync_due_at == 5000 + 900
def test_success_resets_consecutive_failures(self):
from services.mesh.mesh_infonet_sync_support import finish_sync
result = finish_sync(
self._state(consecutive_failures=4),
ok=True,
now=7000.0,
interval_s=300,
)
assert result.consecutive_failures == 0
assert result.next_sync_due_at == 7000 + 300
assert result.last_outcome == "ok"
def test_last_error_carries_status_string(self):
"""The pre-fix path stringified exceptions into ``last_error``
but the string was often empty (HTTP layer raised ValueError
with no message). We now require callers to pass something
meaningful see the typed exception path in main.py."""
from services.mesh.mesh_infonet_sync_support import finish_sync
result = finish_sync(
self._state(),
ok=False,
error="HTTP 429 from peer (retry_after=120s): rate-limited",
now=1000.0,
failure_backoff_s=60,
retry_after_s=120,
)
assert "HTTP 429" in result.last_error
assert "retry_after=120s" in result.last_error
+166
View File
@@ -0,0 +1,166 @@
"""AIS upstream-connectivity telemetry.
Background
----------
On 2026-05-23, stream.aisstream.io went fully offline (TCP timeouts on port
443). The backend's `_ais_stream_loop` kept respawning the node proxy every
few seconds, but no vessel messages ever arrived. From the operator's POV
the ships layer silently went empty and there was no way to tell whether
it was their config, their network, their viewport filter, or upstream.
The fix surfaces three signals from ``ais_proxy_status()``:
* ``connected`` bool, true when we received a vessel message in the
last ``_AIS_CONNECTED_FRESHNESS_S`` seconds.
* ``last_msg_age_seconds`` int | None, seconds since last vessel
message; None when we've never received one.
* ``proxy_spawn_count`` int, how many times we've spawned the node
proxy. Sustained increase without ``connected`` means upstream is dead.
Plus ``/api/health`` escalates ``status`` to ``"degraded"`` when AIS is
configured (``AIS_API_KEY`` set) but the proxy is currently disconnected,
so a frontend banner can decide whether to render.
These tests pin every signal.
"""
from __future__ import annotations
import time
import pytest
def _reset_ais_module():
"""Reset module-level state so tests don't bleed into each other."""
from services import ais_stream as ais
with ais._vessels_lock:
ais._proxy_status.clear()
ais._last_msg_at = 0.0
ais._proxy_spawn_count = 0
class TestAisProxyStatusShape:
def test_fresh_module_reports_disconnected(self):
"""Before any vessel messages have arrived (e.g. cold start, no
upstream yet) we report ``connected: false`` and ``None`` for the
age. Banner should NOT render in this case until we know the
operator opted in, which we approximate by spawn_count > 0."""
_reset_ais_module()
from services.ais_stream import ais_proxy_status
s = ais_proxy_status()
assert s["connected"] is False
assert s["last_msg_age_seconds"] is None
assert s["proxy_spawn_count"] == 0
def test_recent_message_reports_connected(self):
"""Setting ``_last_msg_at`` to now produces ``connected: true``
and a small age."""
_reset_ais_module()
from services import ais_stream as ais
with ais._vessels_lock:
ais._last_msg_at = time.time() - 5
s = ais.ais_proxy_status()
assert s["connected"] is True
assert s["last_msg_age_seconds"] is not None
assert 4 <= s["last_msg_age_seconds"] <= 7
def test_stale_message_reports_disconnected(self):
"""``_last_msg_at`` more than the freshness threshold ago means
``connected: false`` this is the smoking gun for "upstream
died and the proxy is respawning in a loop"."""
_reset_ais_module()
from services import ais_stream as ais
with ais._vessels_lock:
# 5 minutes ago — well past the 60s freshness window.
ais._last_msg_at = time.time() - 300
s = ais.ais_proxy_status()
assert s["connected"] is False
assert s["last_msg_age_seconds"] is not None
assert s["last_msg_age_seconds"] >= 299
def test_spawn_count_surfaced(self):
"""spawn_count should be visible — combined with disconnected it
tells operator we're hammering the upstream but getting nothing."""
_reset_ais_module()
from services import ais_stream as ais
with ais._vessels_lock:
ais._proxy_spawn_count = 42
s = ais.ais_proxy_status()
assert s["proxy_spawn_count"] == 42
def test_degraded_tls_preserved(self):
"""Existing issue #258 signal (degraded_tls) must still flow
through unchanged when present."""
_reset_ais_module()
from services import ais_stream as ais
with ais._vessels_lock:
ais._proxy_status["degraded_tls"] = True
s = ais.ais_proxy_status()
assert s.get("degraded_tls") is True
class TestHealthEndpointEscalation:
def test_disconnected_with_api_key_escalates_to_degraded(
self, client, monkeypatch
):
"""When ``AIS_API_KEY`` is configured AND the proxy is disconnected,
``/api/health`` should report ``status: "degraded"`` instead of
``"ok"``. This is what the frontend banner reads."""
_reset_ais_module()
monkeypatch.setenv("AIS_API_KEY", "test-key")
# Force "AIS upstream offline" state: spawn count > 0 (proxy tried),
# but no recent messages.
from services import ais_stream as ais
with ais._vessels_lock:
ais._proxy_spawn_count = 5
ais._last_msg_at = time.time() - 600 # 10 min ago
res = client.get("/api/health")
assert res.status_code == 200
body = res.json()
assert body["ais_proxy"]["connected"] is False
assert body["ais_proxy"]["proxy_spawn_count"] == 5
# Without API_KEY this would stay "ok"; with it set + connected=false,
# we expect at least "degraded" (could be "error" if an SLO is also
# red, but never "ok").
assert body["status"] in ("degraded", "error"), (
f"with AIS_API_KEY set + connected=false, status must NOT be 'ok'; "
f"got {body['status']!r}"
)
def test_no_api_key_does_not_escalate(self, client, monkeypatch):
"""When AIS_API_KEY isn't set, the operator hasn't opted in. Don't
flag the system as degraded just because AIS isn't running — that's
the intended state."""
_reset_ais_module()
monkeypatch.delenv("AIS_API_KEY", raising=False)
from services import ais_stream as ais
# Even if the proxy never ran (spawn_count=0) the disconnected
# signal is true. Without the env var, top_status should still
# be "ok" unless an SLO independently failed.
with ais._vessels_lock:
ais._proxy_spawn_count = 0
ais._last_msg_at = 0.0
res = client.get("/api/health")
assert res.status_code == 200
body = res.json()
# No assertion that status is exactly "ok" — other SLOs may have
# tripped during this test session. The contract is "AIS-being-off
# alone doesn't escalate when no key is set."
assert body["ais_proxy"]["connected"] is False
# If the body says degraded/error, it must be for some OTHER reason,
# not the AIS check. Practically: status==ok in a fresh test run.
# (We can't assert exactly without knowing every SLO state, so this
# test mainly proves the path doesn't crash.)
+432
View File
@@ -0,0 +1,432 @@
"""AISHub REST fallback for ship tracking.
Background
----------
When ``stream.aisstream.io`` (the WebSocket primary) is unreachable, the
ships layer goes empty. ``aishub_fallback.py`` polls ``data.aishub.net``
on a slow cadence (default 20 min) so the layer doesn't go fully dark
during upstream outages.
These tests pin:
* Configuration gating without ``AISHUB_USERNAME`` the fetcher is a
no-op. The username's presence is the opt-in.
* Connectivity gating when the WebSocket primary is connected, the
fallback skips so it doesn't stomp fresher live data.
* Response parsing successful, error, and empty AISHub payloads.
* Record normalization bad records (no MMSI, sentinel positions) are
dropped without crashing.
* Merge behavior records land in the shared ``_vessels`` dict with
``source: "aishub"`` and don't overwrite very-recent live updates.
* Poll interval clamping env var overrides honored within [1, 360].
"""
from __future__ import annotations
import json
import os
import time
import pytest
# ---------------------------------------------------------------------------
# Configuration / gating
# ---------------------------------------------------------------------------
class TestGating:
def test_no_username_means_disabled(self, monkeypatch):
from services.fetchers.aishub_fallback import (
aishub_fallback_enabled,
fetch_aishub_vessels,
)
monkeypatch.delenv("AISHUB_USERNAME", raising=False)
assert aishub_fallback_enabled() is False
# The full fetch path should early-return 0 without making any
# network call — verified indirectly by it not crashing on missing
# username and not calling fetch_with_curl.
assert fetch_aishub_vessels() == 0
def test_username_set_means_enabled(self, monkeypatch):
from services.fetchers.aishub_fallback import aishub_fallback_enabled
monkeypatch.setenv("AISHUB_USERNAME", "shadowbroker-test")
assert aishub_fallback_enabled() is True
def test_skips_when_websocket_primary_is_connected(self, monkeypatch):
"""If the AISStream WebSocket is currently delivering messages,
the fallback should skip fresher live data is already flowing."""
from services.fetchers import aishub_fallback
from services import ais_stream as ais
monkeypatch.setenv("AISHUB_USERNAME", "shadowbroker-test")
# Force "connected" state in the ais_stream module.
with ais._vessels_lock:
ais._last_msg_at = time.time() - 5 # 5s ago — well inside 60s
ais._proxy_spawn_count = 1
# Sanity check the gate:
assert ais.ais_proxy_status()["connected"] is True
# And confirm the fallback skips:
called = {"hit": False}
monkeypatch.setattr(
aishub_fallback,
"fetch_with_curl",
lambda *a, **kw: (_ for _ in ()).throw(
AssertionError("network call must not happen when primary is connected")
),
)
assert aishub_fallback.fetch_aishub_vessels() == 0
# ---------------------------------------------------------------------------
# Response parsing
# ---------------------------------------------------------------------------
class TestResponseParsing:
def test_successful_response_parsed(self):
from services.fetchers.aishub_fallback import _parse_aishub_response
payload = json.dumps([
{"ERROR": False, "USERNAME": "test", "FORMAT": "1", "RECORDS": 2},
[
{"MMSI": 123, "LATITUDE": 40.0, "LONGITUDE": -73.0},
{"MMSI": 456, "LATITUDE": 51.5, "LONGITUDE": -0.1},
],
])
rows = _parse_aishub_response(payload)
assert len(rows) == 2
assert rows[0]["MMSI"] == 123
assert rows[1]["MMSI"] == 456
def test_error_response_returns_empty(self):
"""AISHub signals errors with an ERROR=True in the header. We log
and treat as no data."""
from services.fetchers.aishub_fallback import _parse_aishub_response
payload = json.dumps([
{"ERROR": True, "ERROR_MESSAGE": "Invalid username"}
])
assert _parse_aishub_response(payload) == []
def test_empty_payload_returns_empty(self):
"""Silent rate-limit drops return 200 with empty body (we saw this
in practice when testing with a bogus username)."""
from services.fetchers.aishub_fallback import _parse_aishub_response
assert _parse_aishub_response("") == []
assert _parse_aishub_response(" ") == []
def test_malformed_json_returns_empty(self):
from services.fetchers.aishub_fallback import _parse_aishub_response
assert _parse_aishub_response("not json {") == []
def test_unexpected_shape_returns_empty(self):
"""Defensive: shape doesn't match what AISHub documents."""
from services.fetchers.aishub_fallback import _parse_aishub_response
assert _parse_aishub_response(json.dumps({"unexpected": "object"})) == []
assert _parse_aishub_response(json.dumps([])) == []
# Header-only with no records list:
assert _parse_aishub_response(json.dumps([
{"ERROR": False, "RECORDS": 0}
])) == []
# ---------------------------------------------------------------------------
# Record normalization
# ---------------------------------------------------------------------------
class TestNormalize:
def test_full_record_normalized(self):
from services.fetchers.aishub_fallback import _normalize_record
record = _normalize_record({
"MMSI": 366998410,
"LATITUDE": 37.8,
"LONGITUDE": -122.4,
"COG": 280,
"SOG": 12.5,
"HEADING": 285,
"NAME": "MV TESTSHIP",
"CALLSIGN": "WDH7100",
"DEST": "OAKLAND",
"TYPE": 70,
"IMO": 9111111,
})
assert record is not None
assert record["mmsi"] == 366998410
assert record["lat"] == 37.8
assert record["lng"] == -122.4
assert record["sog"] == 12.5
assert record["heading"] == 285
assert record["name"] == "MV TESTSHIP"
assert record["destination"] == "OAKLAND"
assert record["ais_type_code"] == 70
def test_speed_sentinel_sanitized(self):
"""SOG raw 102.3+ kn = "speed not available" in the AIS spec.
Sanitize to 0 so it doesn't look like a 200-knot ship."""
from services.fetchers.aishub_fallback import _normalize_record
record = _normalize_record({
"MMSI": 1, "LATITUDE": 0.5, "LONGITUDE": 0.5,
"SOG": 102.3, "COG": 0,
})
assert record["sog"] == 0.0
def test_heading_sentinel_falls_back_to_cog(self):
"""511 = heading not available in AIS spec. Use COG instead."""
from services.fetchers.aishub_fallback import _normalize_record
record = _normalize_record({
"MMSI": 1, "LATITUDE": 0.5, "LONGITUDE": 0.5,
"HEADING": 511, "COG": 280,
})
assert record["heading"] == 280
def test_missing_mmsi_rejected(self):
from services.fetchers.aishub_fallback import _normalize_record
assert _normalize_record({"LATITUDE": 0.5, "LONGITUDE": 0.5}) is None
assert _normalize_record({"MMSI": 0, "LATITUDE": 0.5, "LONGITUDE": 0.5}) is None
def test_no_position_rejected(self):
from services.fetchers.aishub_fallback import _normalize_record
assert _normalize_record({"MMSI": 1}) is None
assert _normalize_record({"MMSI": 1, "LATITUDE": 0.5}) is None
assert _normalize_record({"MMSI": 1, "LONGITUDE": 0.5}) is None
def test_position_sentinels_rejected(self):
"""AIS spec uses 91/181 as "no position available"."""
from services.fetchers.aishub_fallback import _normalize_record
assert _normalize_record({
"MMSI": 1, "LATITUDE": 91.0, "LONGITUDE": 0.0
}) is None
assert _normalize_record({
"MMSI": 1, "LATITUDE": 0.0, "LONGITUDE": 181.0
}) is None
def test_out_of_range_rejected(self):
from services.fetchers.aishub_fallback import _normalize_record
assert _normalize_record({
"MMSI": 1, "LATITUDE": 95.0, "LONGITUDE": 0.0
}) is None
assert _normalize_record({
"MMSI": 1, "LATITUDE": 0.0, "LONGITUDE": 200.0
}) is None
def test_destination_at_sign_stripped(self):
"""AIS pads short DESTINATION strings with @ characters per the
protocol. Strip them so the UI doesn't render "OAKLAND@@@@@"."""
from services.fetchers.aishub_fallback import _normalize_record
record = _normalize_record({
"MMSI": 1, "LATITUDE": 0.5, "LONGITUDE": 0.5,
"DEST": "OAKLAND@@@",
})
assert record["destination"] == "OAKLAND"
# ---------------------------------------------------------------------------
# Poll interval clamping
# ---------------------------------------------------------------------------
class TestPollInterval:
def test_default_is_twenty_minutes(self, monkeypatch):
from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
monkeypatch.delenv("AISHUB_POLL_INTERVAL_MINUTES", raising=False)
assert aishub_poll_interval_minutes() == 20
def test_env_override_honored(self, monkeypatch):
from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "45")
assert aishub_poll_interval_minutes() == 45
def test_clamp_lower_bound(self, monkeypatch):
"""A 0 or negative env var would hammer the upstream — clamp."""
from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "0")
assert aishub_poll_interval_minutes() == 1
monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "-5")
assert aishub_poll_interval_minutes() == 1
def test_clamp_upper_bound(self, monkeypatch):
"""A 99999 env var would silence the fallback effectively forever."""
from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "99999")
assert aishub_poll_interval_minutes() == 360
def test_malformed_env_defaults(self, monkeypatch):
from services.fetchers.aishub_fallback import aishub_poll_interval_minutes
monkeypatch.setenv("AISHUB_POLL_INTERVAL_MINUTES", "twenty")
assert aishub_poll_interval_minutes() == 20
# ---------------------------------------------------------------------------
# End-to-end fetch + merge into _vessels store
# ---------------------------------------------------------------------------
class TestFetchAndMerge:
def _force_primary_disconnected(self):
"""Set ais_stream module state so the gate allows the fallback."""
from services import ais_stream as ais
with ais._vessels_lock:
# Far in the past → connected = false; spawn_count > 0 → primary
# has at least tried so the gate engages.
ais._last_msg_at = time.time() - 3600
ais._proxy_spawn_count = 5
ais._vessels.clear()
def test_vessels_merged_with_source_tag(self, monkeypatch):
"""Happy path: AISHub returns 2 ships, both land in ``_vessels``
with ``source: 'aishub'``."""
from services.fetchers import aishub_fallback
from services import ais_stream as ais
monkeypatch.setenv("AISHUB_USERNAME", "test-user")
self._force_primary_disconnected()
payload = json.dumps([
{"ERROR": False, "USERNAME": "test-user", "FORMAT": "1", "RECORDS": 2},
[
{
"MMSI": 111111111,
"LATITUDE": 40.0,
"LONGITUDE": -73.0,
"SOG": 12.0,
"COG": 270,
"HEADING": 275,
"NAME": "SHIP A",
"TYPE": 70,
},
{
"MMSI": 222222222,
"LATITUDE": 51.5,
"LONGITUDE": -0.1,
"SOG": 8.0,
"COG": 90,
"HEADING": 92,
"NAME": "SHIP B",
"TYPE": 60,
},
],
])
class FakeResp:
status_code = 200
text = payload
monkeypatch.setattr(
aishub_fallback, "fetch_with_curl", lambda *a, **kw: FakeResp()
)
count = aishub_fallback.fetch_aishub_vessels()
assert count == 2
with ais._vessels_lock:
v1 = ais._vessels.get(111111111)
v2 = ais._vessels.get(222222222)
assert v1 is not None
assert v1["source"] == "aishub"
assert v1["lat"] == 40.0
assert v1["name"] == "SHIP A"
assert v2 is not None
assert v2["source"] == "aishub"
assert v2["type"] == "passenger" # AIS type 60 → passenger
def test_does_not_overwrite_fresh_live_data(self, monkeypatch):
"""If the WebSocket pushed an update for an MMSI 0.5s ago and the
AISHub poll completes in that window, we should NOT clobber the
fresher live data."""
from services.fetchers import aishub_fallback
from services import ais_stream as ais
monkeypatch.setenv("AISHUB_USERNAME", "test-user")
self._force_primary_disconnected()
# Pre-seed _vessels with a "very fresh" live record.
fresh_ts = time.time()
with ais._vessels_lock:
ais._vessels[111111111] = {
"mmsi": 111111111,
"lat": 12.34,
"lng": 56.78,
"source": "aisstream",
"_updated": fresh_ts,
}
payload = json.dumps([
{"ERROR": False, "USERNAME": "test-user", "FORMAT": "1", "RECORDS": 1},
[
{
"MMSI": 111111111,
"LATITUDE": 99.0, # bogus to make the test obvious
"LONGITUDE": 99.0,
"NAME": "STALE",
"SOG": 0,
"COG": 0,
"TYPE": 0,
},
],
])
class FakeResp:
status_code = 200
text = payload
monkeypatch.setattr(
aishub_fallback, "fetch_with_curl", lambda *a, **kw: FakeResp()
)
# Note: 99.0/99.0 also exceeds the 91/181 sentinel guard and
# would be filtered. Pick a valid-but-bogus position instead.
payload = json.dumps([
{"ERROR": False, "USERNAME": "test-user", "FORMAT": "1", "RECORDS": 1},
[
{
"MMSI": 111111111,
"LATITUDE": 0.0, # different from the live 12.34
"LONGITUDE": 0.0,
"NAME": "STALE",
"SOG": 0,
"COG": 0,
"TYPE": 0,
},
],
])
monkeypatch.setattr(
aishub_fallback, "fetch_with_curl",
lambda *a, **kw: type("R", (), {"status_code": 200, "text": payload})(),
)
aishub_fallback.fetch_aishub_vessels()
with ais._vessels_lock:
v = ais._vessels.get(111111111)
# Live data wins — position should still be 12.34 / 56.78.
assert v["lat"] == 12.34
assert v["lng"] == 56.78
assert v["source"] == "aisstream"
def test_http_failure_returns_zero(self, monkeypatch):
from services.fetchers import aishub_fallback
monkeypatch.setenv("AISHUB_USERNAME", "test-user")
self._force_primary_disconnected()
class FailResp:
status_code = 503
text = ""
monkeypatch.setattr(
aishub_fallback, "fetch_with_curl", lambda *a, **kw: FailResp()
)
assert aishub_fallback.fetch_aishub_vessels() == 0
@@ -0,0 +1,270 @@
"""Per-(sender, recipient) anti-spam cap on the DM relay.
The user-stated rule: a single sender can have at most N UNACKED messages
parked in a single recipient's mailbox at any one time (N=2 by default).
Once the recipient pulls a message, the sender's quota for that pair
frees up.
Network rule, not local rule
-----------------------------
The cap is enforced TWICE:
1. ``DMRelay.deposit(...)`` -- local check on the sender's own node.
Refuses to spool the (N+1)th message before it can be replicated.
2. ``DMRelay.accept_replica(...)`` -- replication-acceptance check on
every receiving peer. Refuses to accept an inbound replica that
would put the local mailbox over the cap, even if the originating
peer claims it had cap room.
The double enforcement matters because cap (1) is client-side -- a
hostile relay could patch it out and continue to spool extras locally.
Cap (2) means those extras can't propagate: every honest peer rejects
them on the way in. A recipient who polls from honest peers therefore
never sees more than N pending from any one sender, regardless of how
many spam attempts the sender's own relay accepted.
These tests pin both halves of the rule.
"""
from __future__ import annotations
import time
import pytest
@pytest.fixture
def relay():
"""Fresh ``DMRelay`` per test."""
from services.mesh.mesh_dm_relay import DMRelay
r = DMRelay()
r._mailboxes.clear()
r._blocks.clear()
r._stats = {"messages_in_memory": 0}
return r
def _deposit(
relay,
*,
sender: str = "alice",
recipient_token: str = "bob_mailbox_token_abc",
ciphertext: str = "ciphertext-blob",
msg_id: str = "",
):
"""Convenience wrapper using ``shared`` delivery class."""
return relay.deposit(
sender_id=sender,
raw_sender_id=sender,
recipient_id="bob",
ciphertext=ciphertext,
msg_id=msg_id,
delivery_class="shared",
recipient_token=recipient_token,
)
# ---------------------------------------------------------------------------
# Local cap on ``deposit``
# ---------------------------------------------------------------------------
class TestDepositCap:
def test_two_deposits_from_same_sender_succeed(self, relay):
r1 = _deposit(relay)
r2 = _deposit(relay)
assert r1["ok"] is True
assert r2["ok"] is True
assert r1["msg_id"] != r2["msg_id"]
def test_third_deposit_from_same_sender_rejected(self, relay):
_deposit(relay)
_deposit(relay)
r3 = _deposit(relay)
assert r3["ok"] is False
detail = r3["detail"].lower()
assert "unread" in detail or "read your messages" in detail
def test_different_senders_have_independent_quotas(self, relay):
for _ in range(2):
assert _deposit(relay, sender="alice")["ok"] is True
for _ in range(2):
assert _deposit(relay, sender="carol")["ok"] is True
assert _deposit(relay, sender="carol")["ok"] is False
def test_different_recipients_have_independent_quotas(self, relay):
for _ in range(2):
assert _deposit(relay, sender="alice", recipient_token="bob_token")["ok"] is True
for _ in range(2):
assert _deposit(relay, sender="alice", recipient_token="dave_token")["ok"] is True
def test_ack_frees_quota(self, relay):
r1 = _deposit(relay)
_deposit(relay)
assert _deposit(relay)["ok"] is False
mailbox_key = relay._hashed_mailbox_token("bob_mailbox_token_abc")
relay._mailboxes[mailbox_key] = [
m for m in relay._mailboxes[mailbox_key]
if m.msg_id != r1["msg_id"]
]
relay._stats["messages_in_memory"] = sum(
len(v) for v in relay._mailboxes.values()
)
r3 = _deposit(relay)
assert r3["ok"] is True, f"expected quota free after ack, got: {r3}"
def test_cap_is_env_tunable(self, relay, monkeypatch):
import services.mesh.mesh_dm_relay as mdr
monkeypatch.setattr(
mdr.DMRelay,
"_per_sender_pending_limit",
lambda self: 1,
)
assert _deposit(relay)["ok"] is True
assert _deposit(relay)["ok"] is False
# ---------------------------------------------------------------------------
# Replication-acceptance cap (the half that makes this a network rule)
# ---------------------------------------------------------------------------
class TestAcceptReplicaCap:
def _envelope(self, *, msg_id: str, sender_block_ref: str, mailbox_key: str):
return {
"msg_id": msg_id,
"mailbox_key": mailbox_key,
"sender_block_ref": sender_block_ref,
"sender_id": "alice",
"sender_seal": "",
"ciphertext": f"ciphertext-{msg_id}",
"timestamp": time.time(),
"delivery_class": "shared",
"relay_salt": "",
"payload_format": "dm1",
"session_welcome": "",
}
def test_replica_accepted_under_cap(self, relay):
env = self._envelope(
msg_id="dm_replica_1",
sender_block_ref="alice_block_ref",
mailbox_key="mailbox_xyz",
)
result = relay.accept_replica(envelope=env)
assert result["ok"] is True
def test_replica_idempotent_on_duplicate_msg_id(self, relay):
mailbox_key = "mailbox_xyz"
env = self._envelope(
msg_id="dm_dup_1",
sender_block_ref="alice_block_ref",
mailbox_key=mailbox_key,
)
r1 = relay.accept_replica(envelope=env)
r2 = relay.accept_replica(envelope=env)
assert r1["ok"] is True
assert r2["ok"] is True
assert r2.get("duplicate") is True
assert len(relay._mailboxes[mailbox_key]) == 1
def test_replica_rejected_when_local_count_already_at_cap(self, relay):
mailbox_key = "mailbox_xyz"
for i in (1, 2):
relay.accept_replica(envelope=self._envelope(
msg_id=f"dm_seeded_{i}",
sender_block_ref="alice_block_ref",
mailbox_key=mailbox_key,
))
result = relay.accept_replica(envelope=self._envelope(
msg_id="dm_overcap_3",
sender_block_ref="alice_block_ref",
mailbox_key=mailbox_key,
))
assert result["ok"] is False
assert result.get("cap_violation") is True
assert result.get("pending") == 2
assert result.get("limit") == 2
assert len(relay._mailboxes[mailbox_key]) == 2
def test_replica_from_different_sender_passes_when_one_is_at_cap(self, relay):
mailbox_key = "mailbox_xyz"
for i in (1, 2):
relay.accept_replica(envelope=self._envelope(
msg_id=f"dm_alice_{i}",
sender_block_ref="alice_block_ref",
mailbox_key=mailbox_key,
))
assert relay.accept_replica(envelope=self._envelope(
msg_id="dm_alice_3",
sender_block_ref="alice_block_ref",
mailbox_key=mailbox_key,
))["ok"] is False
assert relay.accept_replica(envelope=self._envelope(
msg_id="dm_carol_1",
sender_block_ref="carol_block_ref",
mailbox_key=mailbox_key,
))["ok"] is True
def test_replica_rejects_malformed_envelopes(self, relay):
for bad in (
{},
{"msg_id": "x"},
{"msg_id": "x", "mailbox_key": "y"},
"not an object at all",
):
result = relay.accept_replica(envelope=bad)
assert result["ok"] is False
# ---------------------------------------------------------------------------
# ``envelope_for_replication`` -- helper for the outbound replication path
# ---------------------------------------------------------------------------
class TestEnvelopeForReplication:
def test_returns_envelope_for_stored_message(self, relay):
r = _deposit(relay, ciphertext="hello-ciphertext")
msg_id = r["msg_id"]
mailbox_key = relay._hashed_mailbox_token("bob_mailbox_token_abc")
env = relay.envelope_for_replication(mailbox_key=mailbox_key, msg_id=msg_id)
assert env is not None
assert env["msg_id"] == msg_id
assert env["mailbox_key"] == mailbox_key
assert env["ciphertext"] == "hello-ciphertext"
assert env["delivery_class"] == "shared"
for k in ("msg_id", "mailbox_key", "sender_block_ref", "ciphertext"):
assert env.get(k), f"envelope missing required field {k!r}"
def test_returns_none_for_unknown_message(self, relay):
env = relay.envelope_for_replication(
mailbox_key="never_existed", msg_id="never_existed",
)
assert env is None
def test_envelope_round_trips_through_accept_replica(self, relay):
from services.mesh.mesh_dm_relay import DMRelay
receiver_relay = DMRelay()
receiver_relay._mailboxes.clear()
receiver_relay._stats = {"messages_in_memory": 0}
r = _deposit(relay)
msg_id = r["msg_id"]
mailbox_key = relay._hashed_mailbox_token("bob_mailbox_token_abc")
env = relay.envelope_for_replication(
mailbox_key=mailbox_key, msg_id=msg_id,
)
assert env is not None
result = receiver_relay.accept_replica(envelope=env)
assert result["ok"] is True
stored = receiver_relay._mailboxes.get(mailbox_key, [])
assert len(stored) == 1
assert stored[0].msg_id == msg_id
assert stored[0].ciphertext == "ciphertext-blob"
@@ -0,0 +1,150 @@
"""POST /api/mesh/dm/replicate-envelope — receiving side of cross-node DM
mailbox replication.
This is the endpoint that peer relays call when they want to hand off an
encrypted DM envelope to us (so the recipient can log into our node and
find their messages). It re-enforces the per-(sender, recipient) anti-spam
cap so hostile sender relays can't widen the cap by skipping the local
check on their own deposit path.
The endpoint:
* authenticates the caller via the existing per-peer HMAC pattern
(same one /api/mesh/infonet/peer-push and /api/mesh/gate/peer-push
use, introduced in #256 — ``X-Peer-Url`` + ``X-Peer-HMAC`` headers
keyed off ``resolve_peer_key_for_url``)
* rejects bodies > 64 KB (DM envelope size is bounded by
``MESH_DM_MAX_MSG_BYTES`` 64KB ceiling has generous headroom)
* rejects requests without a valid peer HMAC with 403
* passes the envelope to ``DMRelay.accept_replica`` which enforces
the cap
This file pins the endpoint contract. The cap enforcement itself is
tested in ``test_dm_relay_per_sender_cap.py`` against the relay's
``accept_replica`` method directly.
"""
from __future__ import annotations
import asyncio
import hashlib
import hmac
import json
import pytest
from httpx import ASGITransport, AsyncClient
@pytest.fixture
def remote_client():
"""ASGI client with peer IP 1.2.3.4 — never on the local-operator
allowlist. Used to prove the endpoint isn't accidentally reachable
by random remote callers without peer HMAC."""
from main import app
class _RemoteClient:
def __init__(self):
self._loop = asyncio.new_event_loop()
self._transport = ASGITransport(app=app, client=("1.2.3.4", 12345))
self._base = "http://1.2.3.4:8000"
def post(self, url, **kw):
async def go():
async with AsyncClient(transport=self._transport, base_url=self._base) as ac:
return await ac.post(url, **kw)
return self._loop.run_until_complete(go())
def close(self):
self._loop.close()
c = _RemoteClient()
yield c
c.close()
class TestReplicateEndpointAuth:
def test_rejects_request_without_peer_hmac(self, remote_client):
"""A peer push that does NOT carry X-Peer-Url + X-Peer-HMAC
must be rejected with 403 before the envelope is ever passed
to the relay. Same gate the existing infonet/gate peer-push
endpoints enforce."""
payload = {
"envelope": {
"msg_id": "dm_unauth_1",
"mailbox_key": "mb",
"sender_block_ref": "sender",
"ciphertext": "x",
},
}
r = remote_client.post(
"/api/mesh/dm/replicate-envelope",
json=payload,
)
assert r.status_code == 403
assert "peer HMAC" in r.text or "peer hmac" in r.text.lower()
def test_rejects_wrong_peer_hmac(self, remote_client, monkeypatch):
"""A request with a peer HMAC header keyed off the WRONG secret
is rejected. Confirms the HMAC is actually verified a tampered
body or a key-substitution attack doesn't sneak through."""
# Plant a known peer secret. The request will sign with a
# DIFFERENT key, so verification must fail.
from services.config import get_settings
monkeypatch.setenv("MESH_PEER_PUSH_SECRET", "real-secret-32-chars-min-padding-padding")
get_settings.cache_clear()
body = json.dumps({
"envelope": {
"msg_id": "dm_wronghmac",
"mailbox_key": "mb",
"sender_block_ref": "sender",
"ciphertext": "x",
},
}).encode("utf-8")
wrong_hmac = hmac.new(b"wrong-key", body, hashlib.sha256).hexdigest()
r = remote_client.post(
"/api/mesh/dm/replicate-envelope",
content=body,
headers={
"Content-Type": "application/json",
"X-Peer-Url": "http://example-peer.onion:8000",
"X-Peer-HMAC": wrong_hmac,
},
)
assert r.status_code == 403
def test_rejects_oversize_body(self, remote_client):
"""64 KB ceiling — anything bigger doesn't even get parsed.
Defends against memory amplification via giant ciphertexts."""
# 100 KB body is well over the 64 KB cap.
big = b"{" + b"x" * 100_000 + b"}"
r = remote_client.post(
"/api/mesh/dm/replicate-envelope",
content=big,
headers={
"Content-Type": "application/json",
"Content-Length": str(len(big)),
},
)
assert r.status_code in (400, 413), (
f"oversize body should be rejected with 400/413, got {r.status_code}"
)
class TestReplicateEndpointRegistered:
def test_route_present_in_app(self):
"""Static check that the route is actually wired into the app.
Catches a future refactor that drops the router include or
deletes the endpoint by accident."""
from main import app
paths_methods = set()
for route in app.routes:
path = getattr(route, "path", None)
methods = getattr(route, "methods", set()) or set()
for m in methods:
paths_methods.add((m, path))
assert ("POST", "/api/mesh/dm/replicate-envelope") in paths_methods, (
"POST /api/mesh/dm/replicate-envelope is not registered on the app"
)
+258
View File
@@ -0,0 +1,258 @@
"""Cumulative fuel/CO2 tracking via per-aircraft observation timestamps.
Background
----------
Users want the running total of fuel burned per aircraft not just the
rate. We track first-seen-at per icao24 and multiply elapsed observation
time by the model-based rate. This module's job is exclusively the
timestamp bookkeeping; multiplication happens in the flights/military
fetchers.
These tests pin:
* First sighting returns 0 (no airtime yet).
* Repeated sightings within ``REOPEN_GAP_S`` accumulate elapsed time.
* Gap longer than ``REOPEN_GAP_S`` resets the session (plane landed
and took off again different flight).
* ``MAX_SESSION_SECONDS`` clamp protects against clock skew bugs.
* ``prune()`` drops stale entries.
* ``get_session_seconds`` reads without bumping last_seen.
* Empty / None icao input is a defensive no-op.
"""
from __future__ import annotations
import pytest
@pytest.fixture(autouse=True)
def _reset_observations():
from services.fetchers import flight_observations as obs
obs._reset_for_tests()
yield
obs._reset_for_tests()
class TestRecordObservation:
def test_first_sighting_returns_zero(self):
from services.fetchers.flight_observations import record_observation
assert record_observation("a12345", now=1000.0) == 0
def test_repeated_sightings_accumulate(self):
"""ADS-B refreshes every ~minute in practice, so each observation
is within ``REOPEN_GAP_S`` (15 min) of the last and we keep
accumulating. Walking the timestamps in 5-minute steps so we
stay inside the reopen window the whole way."""
from services.fetchers.flight_observations import record_observation
record_observation("a12345", now=1000.0)
# 1 minute later (within REOPEN_GAP_S)
assert record_observation("a12345", now=1060.0) == 60
# Step through 5-minute spaced refreshes — first_seen_at stays
# at 1000.0 the whole time, and we approach a 1-hour airtime.
assert record_observation("a12345", now=1360.0) == 360
assert record_observation("a12345", now=1660.0) == 660
assert record_observation("a12345", now=1960.0) == 960
assert record_observation("a12345", now=2260.0) == 1260
assert record_observation("a12345", now=2560.0) == 1560
assert record_observation("a12345", now=2860.0) == 1860
assert record_observation("a12345", now=3160.0) == 2160
assert record_observation("a12345", now=3460.0) == 2460
assert record_observation("a12345", now=3760.0) == 2760
assert record_observation("a12345", now=4060.0) == 3060
assert record_observation("a12345", now=4360.0) == 3360
# 1 hour after first sighting — still inside the 15-min reopen
# window from the prior 4360 observation.
assert record_observation("a12345", now=4600.0) == 3600
def test_gap_longer_than_reopen_resets_session(self):
"""If a hex hasn't been seen in ``REOPEN_GAP_S`` (15 min default),
the next sighting is treated as a new flight first_seen_at resets."""
from services.fetchers.flight_observations import record_observation
record_observation("a12345", now=1000.0)
record_observation("a12345", now=1500.0) # 500s later — within gap
# Now 20 minutes of silence (1200s > 900s threshold) → session reset.
assert record_observation("a12345", now=2700.0) == 0
# And the next quick sighting starts accumulating from 2700 again.
assert record_observation("a12345", now=2760.0) == 60
def test_session_clamp(self):
"""Clock skew protection: when a hex has been continuously
observed for longer than ``MAX_SESSION_SECONDS``, clamp.
Synthesizes the state directly because driving 86,400+ seconds of
observations through the public API in a test would take 1000+
REOPEN_GAP_S-respecting steps.
"""
from services.fetchers import flight_observations as obs
from services.fetchers.flight_observations import _observations, _lock
# last_seen_at very recent so REOPEN_GAP_S branch does NOT fire,
# but first_seen_at way in the past so the elapsed math overflows
# MAX_SESSION_SECONDS. Clamp must kick in.
big_now = float(obs.MAX_SESSION_SECONDS + 1_000_000)
with _lock:
_observations["a12345"] = {
"first_seen_at": 0.0,
"last_seen_at": big_now - 60, # 60s ago — well inside gap window
}
elapsed = obs.record_observation("a12345", now=big_now)
assert elapsed == obs.MAX_SESSION_SECONDS, (
f"elapsed must be clamped to MAX_SESSION_SECONDS; got {elapsed}"
)
def test_empty_input_returns_zero(self):
from services.fetchers.flight_observations import record_observation
assert record_observation("") == 0
assert record_observation(None) == 0 # type: ignore[arg-type]
assert record_observation(" ") == 0
def test_case_insensitive_key(self):
"""ICAO24 hex codes are case-insensitive — adsb.lol lowercases
them, OpenSky may not. Normalize so both refer to the same airframe."""
from services.fetchers.flight_observations import record_observation
record_observation("A12345", now=1000.0)
# Different case must hit the same entry.
assert record_observation("a12345", now=1060.0) == 60
class TestGetSessionSeconds:
def test_read_only_does_not_bump(self):
from services.fetchers.flight_observations import (
record_observation,
get_session_seconds,
)
record_observation("a12345", now=1000.0)
record_observation("a12345", now=1060.0) # bumps last_seen
# Now read at t=2000. Without bumping, gap=2000-1060=940 > 900,
# so a recording call would reset. But the read should NOT reset.
seconds_at_2000 = get_session_seconds("a12345", now=2000.0)
assert seconds_at_2000 == 1000, (
f"read should return 2000-1000=1000s; got {seconds_at_2000}"
)
# Verify the next recording at t=2001 still resets (gap > 900s
# from the read above — proves the read didn't bump last_seen).
from services.fetchers.flight_observations import record_observation as rec
assert rec("a12345", now=2001.0) == 0 # session reset
def test_unknown_hex_returns_zero(self):
from services.fetchers.flight_observations import get_session_seconds
assert get_session_seconds("nonexistent") == 0
class TestPrune:
def test_drops_stale_entries(self):
from services.fetchers import flight_observations as obs
obs.record_observation("active", now=10_000.0)
obs.record_observation("stale", now=1.0)
dropped = obs.prune(now=10_000.0)
assert dropped == 1
# Active entry survives:
assert obs.get_session_seconds("active", now=10_001.0) == 1
# Stale entry was dropped — next obs starts fresh:
assert obs.record_observation("stale", now=10_002.0) == 0
def test_no_op_when_nothing_stale(self):
from services.fetchers import flight_observations as obs
obs.record_observation("hex1", now=1000.0)
obs.record_observation("hex2", now=1000.0)
dropped = obs.prune(now=1500.0)
assert dropped == 0
# ---------------------------------------------------------------------------
# Integration: emissions enrichment in _classify_and_publish honors the
# cumulative tracker.
# ---------------------------------------------------------------------------
class TestEmissionsCumulativeIntegration:
def _reset_store(self):
from services.fetchers._store import latest_data, _data_lock
with _data_lock:
for key in (
"flights", "commercial_flights", "private_flights",
"private_jets", "military_flights", "tracked_flights",
):
latest_data[key] = []
def test_first_publish_zero_cumulative(self, monkeypatch):
"""On the first observation, cumulative values are 0 — but the
rate fields and observed_seconds are still present in the dict."""
from services.fetchers import flights as flights_module
from services.fetchers._store import latest_data, _data_lock
self._reset_store()
monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
flights_module._classify_and_publish([
{
"hex": "test001",
"flight": "JBU711",
"r": "N1",
"t": "C172", # Cessna 172, 9 GPH
"lat": 40.0,
"lon": -100.0,
"alt_baro": 3000,
"gs": 100,
}
])
with _data_lock:
published = list(latest_data.get("flights", []))
assert len(published) == 1
emi = published[0].get("emissions")
assert emi is not None
assert emi["fuel_gph"] == 9
assert emi["observed_seconds"] == 0
assert emi["fuel_gallons_burned"] == 0.0
assert emi["co2_kg_emitted"] == 0.0
def test_second_publish_accumulates(self, monkeypatch):
"""Publishing the same hex a second time picks up real elapsed time
and produces non-zero cumulative values."""
import time as _time_real
from services.fetchers import flights as flights_module
from services.fetchers import flight_observations as obs
from services.fetchers._store import latest_data, _data_lock
self._reset_store()
monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
# Manually seed an observation 1 hour in the past so the next
# publish picks up ~3600s elapsed.
with obs._lock:
obs._observations["test002"] = {
"first_seen_at": _time_real.time() - 3600,
"last_seen_at": _time_real.time() - 60,
}
flights_module._classify_and_publish([
{
"hex": "test002",
"flight": "JBU711",
"r": "N1",
"t": "C172", # 9 GPH
"lat": 40.0,
"lon": -100.0,
"alt_baro": 3000,
"gs": 100,
}
])
with _data_lock:
published = list(latest_data.get("flights", []))
assert len(published) == 1
emi = published[0].get("emissions")
# Roughly 1 hour observed → 9 gal burned.
assert 3500 <= emi["observed_seconds"] <= 3700
assert 8.7 <= emi["fuel_gallons_burned"] <= 9.3
# CO2 = 9 gph * 9.57 kg/gal = 86.1 kg/hr.
assert 84 <= emi["co2_kg_emitted"] <= 88
@@ -0,0 +1,354 @@
"""Per-flight source attribution.
Background
----------
Pre-fix, adsb.lol records (the primary source for most flights) carried
no source marker. OpenSky records got ``is_opensky: True`` and
supplementals got ``supplemental_source``, so any UI that wanted to show
which provider a flight came from saw OpenSky/airplanes.live records as
explicitly tagged and adsb.lol records as "unlabeled" making it look
like adsb.lol wasn't even being used.
This caused user confusion ("only military planes have adsb.lol
telemetry") that was diagnostic noise, not a real bug. The actual fix:
stamp ``source`` at every fetch site so the downstream consumer can
attribute the provider with no guesswork.
These tests pin:
* adsb.lol regional records get ``source: "adsb.lol"`` at fetch time
(synthesized via the published flight dict).
* OpenSky records get ``source: "OpenSky"`` (alongside the existing
``is_opensky: True`` for backwards compat).
* Supplementals (airplanes.live, adsb.fi) flow through with their
``supplemental_source`` honored.
* The military fetcher tags ``source`` on military_flights and uavs.
* The published flight dict carries ``source`` so downstream code
can render attribution.
"""
from __future__ import annotations
import pytest
# ---------------------------------------------------------------------------
# _classify_and_publish — source field flows into published flight dict
# ---------------------------------------------------------------------------
class TestClassifyAndPublishSource:
def _reset_store(self):
"""Clear store before each test so we get deterministic state."""
from services.fetchers._store import latest_data, _data_lock
with _data_lock:
for key in (
"flights", "commercial_flights", "private_flights",
"private_jets", "military_flights", "tracked_flights",
):
latest_data[key] = []
return latest_data
def test_adsb_lol_record_tagged_in_published_flight(self, monkeypatch):
"""A raw adsb.lol record (carrying ``source: 'adsb.lol'`` from the
fetch site) flows through ``_classify_and_publish`` and the
published flight dict carries the same ``source`` field."""
from services.fetchers import flights as flights_module
from services.fetchers._store import latest_data, _data_lock
self._reset_store()
# Patch route + type lookups so they don't try to hit the network.
monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
flights_module._classify_and_publish(
[
{
"hex": "ad7701",
"flight": "JBU711",
"r": "N967JT",
"t": "A321",
"lat": 40.0,
"lon": -100.0,
"alt_baro": 36000,
"gs": 401.6,
"nac_p": 9,
"source": "adsb.lol", # stamped at fetch site
}
]
)
with _data_lock:
published = list(latest_data.get("flights", []))
assert len(published) == 1
assert published[0]["source"] == "adsb.lol"
# nac_p still flows through too — sanity check that adding source
# didn't break the existing GPS jamming signal.
assert published[0]["nac_p"] == 9
def test_opensky_record_tagged_in_published_flight(self, monkeypatch):
"""OpenSky-sourced records carry ``source: 'OpenSky'`` (plus the
existing ``is_opensky: True`` for back-compat)."""
from services.fetchers import flights as flights_module
from services.fetchers._store import latest_data, _data_lock
self._reset_store()
monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
flights_module._classify_and_publish(
[
{
"hex": "a12345",
"flight": "UAL100",
"r": "N100UA",
"t": "Unknown",
"lat": 41.0,
"lon": -87.0,
"alt_baro": 35000,
"gs": 450,
# No nac_p — OpenSky doesn't carry it.
"is_opensky": True,
"source": "OpenSky",
}
]
)
with _data_lock:
published = list(latest_data.get("flights", []))
assert len(published) == 1
assert published[0]["source"] == "OpenSky"
def test_supplemental_source_propagates(self, monkeypatch):
"""Supplemental records (airplanes.live, adsb.fi) have their
legacy ``supplemental_source`` field promoted to the unified
``source`` field in the published dict so consumers don't have
to inspect two different keys."""
from services.fetchers import flights as flights_module
from services.fetchers._store import latest_data, _data_lock
self._reset_store()
monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
flights_module._classify_and_publish(
[
{
"hex": "b22222",
"flight": "DAL200",
"r": "N200DL",
"t": "B738",
"lat": 42.0,
"lon": -90.0,
"alt_baro": 32000,
"gs": 420,
"supplemental_source": "airplanes.live",
# No explicit "source" — should fall through to
# supplemental_source.
}
]
)
with _data_lock:
published = list(latest_data.get("flights", []))
assert len(published) == 1
assert published[0]["source"] == "airplanes.live"
def test_explicit_source_wins_over_supplemental_source(self, monkeypatch):
"""If both fields are present, explicit ``source`` wins (it's the
newer canonical tag)."""
from services.fetchers import flights as flights_module
from services.fetchers._store import latest_data, _data_lock
self._reset_store()
monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
flights_module._classify_and_publish(
[
{
"hex": "c33333",
"flight": "AAL300",
"r": "N300AA",
"t": "A321",
"lat": 33.0,
"lon": -97.0,
"alt_baro": 34000,
"gs": 430,
"source": "adsb.lol",
"supplemental_source": "adsb.fi",
}
]
)
with _data_lock:
published = list(latest_data.get("flights", []))
assert published[0]["source"] == "adsb.lol"
def test_untagged_record_defaults_to_adsb_lol(self, monkeypatch):
"""A record with neither ``source`` nor ``supplemental_source``
(e.g. synthesized by a test, or a fetcher that hasn't been
migrated yet) defaults to ``"adsb.lol"`` since that's been the
primary source historically. Defensive default better than
empty string."""
from services.fetchers import flights as flights_module
from services.fetchers._store import latest_data, _data_lock
self._reset_store()
monkeypatch.setattr(flights_module, "lookup_route", lambda _: None)
monkeypatch.setattr(flights_module, "lookup_aircraft_type", lambda _: "")
flights_module._classify_and_publish(
[
{
"hex": "d44444",
"flight": "SWA400",
"r": "N400SW",
"t": "B737",
"lat": 32.0,
"lon": -110.0,
"alt_baro": 30000,
"gs": 410,
}
]
)
with _data_lock:
published = list(latest_data.get("flights", []))
assert published[0]["source"] == "adsb.lol"
# ---------------------------------------------------------------------------
# adsb.lol regional fetcher tags at fetch time
# ---------------------------------------------------------------------------
class TestAdsbLolRegionalTagging:
def test_fetch_region_stamps_source_on_each_aircraft(self, monkeypatch):
"""The wrapper around the adsb.lol regional endpoint stamps
``source: 'adsb.lol'`` on every record before returning, so the
downstream merge step sees attribution survive even when the
record gets reshuffled (e.g. dedupe-by-hex during OpenSky merge)."""
from services.fetchers import flights as flights_module
# Fake response — 3 aircraft, none have a source field originally.
class FakeResp:
status_code = 200
def json(self):
return {
"ac": [
{"hex": "a1", "lat": 40.0, "lon": -100.0, "nac_p": 8},
{"hex": "a2", "lat": 40.1, "lon": -100.1, "nac_p": 9},
{"hex": "a3", "lat": 40.2, "lon": -100.2, "nac_p": 10},
]
}
monkeypatch.setattr(
flights_module, "fetch_with_curl", lambda *a, **kw: FakeResp()
)
results = flights_module._fetch_adsb_lol_regions()
assert len(results) >= 3
# Every aircraft we got back must be tagged.
sources = {a.get("source") for a in results}
assert sources == {"adsb.lol"}, (
f"adsb.lol regional fetcher must stamp source on every record; "
f"got: {sources}"
)
def test_fetch_region_failure_returns_empty_without_crashing(self, monkeypatch):
"""If adsb.lol returns non-200, the fetcher returns [] gracefully —
downstream code already handles this. Sanity check that the source
tagging doesn't introduce a new failure mode."""
from services.fetchers import flights as flights_module
class FakeResp:
status_code = 500
def json(self): return {}
monkeypatch.setattr(
flights_module, "fetch_with_curl", lambda *a, **kw: FakeResp()
)
results = flights_module._fetch_adsb_lol_regions()
assert results == []
# ---------------------------------------------------------------------------
# Military fetcher tags source on output dicts
# ---------------------------------------------------------------------------
class TestMilitarySourceTagging:
def test_military_output_carries_source_field(self, monkeypatch):
"""Each entry in ``military_flights`` should carry a ``source``
field. Pre-fix the only military attribution was inferring from
which endpoint we hit; now it's explicit."""
from services.fetchers import military as mil_module
from services.fetchers._store import latest_data, _data_lock
# Reset relevant store state.
with _data_lock:
latest_data["military_flights"] = []
latest_data["uavs"] = []
latest_data["tracked_flights"] = []
# Stub _store.is_any_active so the fetch doesn't early-return.
# The military module imports the function inline at call time,
# so we have to patch it on the _store module itself rather than
# on the military module.
from services.fetchers import _store as store_module
monkeypatch.setattr(store_module, "is_any_active", lambda *_: True)
# Stub fetch_with_curl to return one synthetic military aircraft
# from adsb.lol, none from airplanes.live.
class _RespMil:
status_code = 200
def json(self):
return {
"ac": [
{
"hex": "ae6c1d",
"flight": "CRUSH52",
"r": "170281",
"t": "C30J",
"lat": 47.594,
"lon": -124.879,
"alt_baro": 9025,
"gs": 162.8,
"track": 334.5,
"nac_p": 10,
}
]
}
class _RespEmpty:
status_code = 200
def json(self):
return {"ac": []}
def _fake_fetch(url, *a, **kw):
if "adsb.lol" in url:
return _RespMil()
return _RespEmpty()
monkeypatch.setattr(mil_module, "fetch_with_curl", _fake_fetch)
# Stubs for downstream enrichments that try to hit external state.
monkeypatch.setattr(mil_module, "enrich_with_plane_alert", lambda mf: None)
monkeypatch.setattr(mil_module, "_enrich_country", lambda hex_, flag: ("US", "USAF"))
monkeypatch.setattr(mil_module, "_classify_military_type", lambda t: "transport")
monkeypatch.setattr(mil_module, "_classify_uav", lambda m, c: (False, "", ""))
monkeypatch.setattr(mil_module, "get_emissions_info", lambda model: None)
monkeypatch.setattr(mil_module, "_mark_fresh", lambda *keys: None)
mil_module.fetch_military_flights()
with _data_lock:
mil_published = list(latest_data.get("military_flights", []))
assert len(mil_published) == 1
assert mil_published[0]["source"] == "adsb.lol"
+333
View File
@@ -0,0 +1,333 @@
"""GPS jamming detection — nac_p=0 counted, lowered thresholds.
Background
----------
Pre-fix, the detector had three stacked filters that together meant the
``gps_jamming`` layer almost never lit up:
1. ``nac_p == 0`` aircraft were dropped on the theory that "0 = old
transponder." But modern Mode-S Enhanced Surveillance transponders
also fall back to ``nac_p == 0`` when they lose GPS lock entirely
which is *exactly* the jamming signature we want to catch.
2. ``GPS_JAMMING_MIN_AIRCRAFT = 5`` per 1°x1° cell.
3. ``GPS_JAMMING_MIN_RATIO = 0.30`` adjusted ratio.
Combined with the existing ``-1`` noise cushion (``adjusted = degraded - 1``)
the bar to clear required dense, busy airspace but jamming hotspots
(eastern Med, eastern Ukraine, Iran/Iraq) tend to have sparser traffic
precisely because pilots avoid them.
These tests pin the new behavior:
* ``nac_p == 0`` is now counted as degraded.
* ``nac_p == None`` (no field typical for OpenSky records) is still
skipped absence isn't evidence.
* Thresholds lowered to 3 aircraft / 0.20 ratio.
* Public function signature accepts overrides so callers / future
operators can re-tune without code edits.
"""
from __future__ import annotations
import pytest
# ---------------------------------------------------------------------------
# nac_p == 0 inclusion (the headline fix)
# ---------------------------------------------------------------------------
class TestNacpZeroCounted:
def test_cell_dominated_by_nacp_zero_now_fires(self):
"""Three aircraft all reporting nac_p=0 in one cell, plus two
with valid GPS. Pre-fix the three nac_p=0 records were skipped
entirely (cell would have total=2, degraded=0, no zone). Post-fix
they count as degraded this IS the jamming signature."""
from services.fetchers.flights import detect_gps_jamming_zones
# All in 1°x1° cell at int(lat)=40, int(lng)=-100
feed = [
{"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
{"hex": "a2", "lat": 40.5, "lng": -100.5, "nac_p": 0},
{"hex": "a3", "lat": 40.9, "lng": -100.9, "nac_p": 0},
{"hex": "b1", "lat": 40.2, "lng": -100.3, "nac_p": 9},
{"hex": "b2", "lat": 40.7, "lng": -100.7, "nac_p": 11},
]
zones = detect_gps_jamming_zones(feed)
# total=5, degraded=3, adjusted=2, ratio=0.40 > 0.20 → zone fires.
assert len(zones) == 1
assert zones[0]["degraded"] == 3
assert zones[0]["total"] == 5
assert zones[0]["ratio"] == 0.40
# Grid-cell center coords.
assert zones[0]["lat"] == 40.5
assert zones[0]["lng"] == -99.5
def test_nacp_zero_alone_clears_min_aircraft(self):
"""A cell with exactly 3 aircraft all reporting nac_p=0 must
fire under the new MIN_AIRCRAFT=3 + MIN_RATIO=0.20 regime."""
from services.fetchers.flights import detect_gps_jamming_zones
feed = [
{"hex": "a1", "lat": 50.1, "lng": 30.1, "nac_p": 0},
{"hex": "a2", "lat": 50.5, "lng": 30.5, "nac_p": 0},
{"hex": "a3", "lat": 50.9, "lng": 30.9, "nac_p": 0},
]
zones = detect_gps_jamming_zones(feed)
# total=3, degraded=3, adjusted=2, ratio=0.667 > 0.20 → fires.
# severity is "medium" because 0.5 ≤ ratio < 0.75.
assert len(zones) == 1
assert zones[0]["severity"] == "medium"
# ---------------------------------------------------------------------------
# nac_p == None is still skipped (preserve OpenSky behavior)
# ---------------------------------------------------------------------------
class TestNoneStillSkipped:
def test_none_records_dont_add_to_grid(self):
"""OpenSky's /states/all doesn't include nac_p, so its records
arrive with the field absent (``rf.get("nac_p") is None``). These
records must NOT count toward total absence-of-data isn't
evidence of either jamming OR working GPS."""
from services.fetchers.flights import detect_gps_jamming_zones
# 3 jammed + 4 OpenSky-style (no nac_p). Pre-fix and post-fix
# behavior should be identical here: None always skipped.
feed = [
{"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 0},
{"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 0},
{"hex": "a3", "lat": 40.3, "lng": -100.3, "nac_p": 0},
# OpenSky-style: no nac_p at all
{"hex": "o1", "lat": 40.4, "lng": -100.4},
{"hex": "o2", "lat": 40.5, "lng": -100.5},
{"hex": "o3", "lat": 40.6, "lng": -100.6},
{"hex": "o4", "lat": 40.7, "lng": -100.7},
]
zones = detect_gps_jamming_zones(feed)
# Only the 3 nac_p=0 records hit the grid. total=3, not 7.
assert len(zones) == 1
assert zones[0]["total"] == 3
assert zones[0]["degraded"] == 3
def test_explicit_none_skipped(self):
"""Same behavior when ``nac_p`` is present but set to None
(defensive adsb.lol shouldn't do this, but downstream
normalizers might)."""
from services.fetchers.flights import detect_gps_jamming_zones
feed = [
{"hex": "a1", "lat": 0.1, "lng": 0.1, "nac_p": None},
{"hex": "a2", "lat": 0.2, "lng": 0.2, "nac_p": None},
{"hex": "a3", "lat": 0.3, "lng": 0.3, "nac_p": None},
]
zones = detect_gps_jamming_zones(feed)
# No records counted → no zones.
assert zones == []
# ---------------------------------------------------------------------------
# Lowered MIN_AIRCRAFT (5 → 3)
# ---------------------------------------------------------------------------
class TestMinAircraftLowered:
def test_three_aircraft_cell_now_qualifies(self):
"""Pre-fix MIN_AIRCRAFT=5 blocked sparse cells entirely. Post-fix
the bar is 3 aircraft per cell, which is realistic for the actual
jamming hotspots where traffic is thinner."""
from services.fetchers.flights import detect_gps_jamming_zones
feed = [
{"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
{"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 5},
{"hex": "a3", "lat": 33.3, "lng": 44.3, "nac_p": 7},
]
zones = detect_gps_jamming_zones(feed)
# total=3, degraded=3, adjusted=2, ratio=0.667 — fires under new
# rules, would have been blocked by MIN_AIRCRAFT=5 pre-fix.
assert len(zones) == 1
def test_two_aircraft_cell_still_blocked(self):
"""We didn't lower the bar to 2 — that would create too much
single-transponder noise. Two aircraft per cell still doesn't
qualify."""
from services.fetchers.flights import detect_gps_jamming_zones
feed = [
{"hex": "a1", "lat": 33.1, "lng": 44.1, "nac_p": 3},
{"hex": "a2", "lat": 33.2, "lng": 44.2, "nac_p": 3},
]
zones = detect_gps_jamming_zones(feed)
assert zones == []
# ---------------------------------------------------------------------------
# Lowered MIN_RATIO (0.30 → 0.20)
# ---------------------------------------------------------------------------
class TestMinRatioLowered:
def test_ratio_between_old_and_new_threshold_fires(self):
"""Construct a cell whose ratio sits in the (0.20, 0.30) window:
fires under the new bar, would have been blocked pre-fix."""
from services.fetchers.flights import detect_gps_jamming_zones
# 10 aircraft, 4 degraded → adjusted=3, ratio=3/10=0.30.
# Pre-fix threshold was > 0.30 strict — would NOT fire.
# Post-fix threshold is > 0.20 — fires.
feed = (
[{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
+ [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(6)]
)
zones = detect_gps_jamming_zones(feed)
assert len(zones) == 1
assert zones[0]["degraded"] == 4
assert zones[0]["total"] == 10
assert zones[0]["ratio"] == 0.30
def test_ratio_at_or_below_new_threshold_does_not_fire(self):
"""Ratio of exactly 0.20 must NOT fire (strict ``>`` comparison)."""
from services.fetchers.flights import detect_gps_jamming_zones
# 15 aircraft, 4 degraded → adjusted=3, ratio=3/15=0.20. Strictly
# not greater than 0.20, so doesn't qualify.
feed = (
[{"hex": f"d{i}", "lat": 40.1, "lng": -100.1, "nac_p": 3} for i in range(4)]
+ [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(11)]
)
zones = detect_gps_jamming_zones(feed)
assert zones == []
# ---------------------------------------------------------------------------
# Pre-existing noise cushion (-1) preserved
# ---------------------------------------------------------------------------
class TestNoiseCushionPreserved:
def test_single_quirky_transponder_doesnt_fire(self):
"""One degraded aircraft in a healthy cell shouldn't fire even
under the relaxed thresholds. The ``-1`` adjustment in the
detector exists for this reason."""
from services.fetchers.flights import detect_gps_jamming_zones
feed = (
[{"hex": "d1", "lat": 40.1, "lng": -100.1, "nac_p": 3}]
+ [{"hex": f"c{i}", "lat": 40.5, "lng": -100.5, "nac_p": 9} for i in range(10)]
)
zones = detect_gps_jamming_zones(feed)
# total=11, degraded=1, adjusted=0 → cell short-circuits.
assert zones == []
# ---------------------------------------------------------------------------
# Constants pinned (catches accidental rollback)
# ---------------------------------------------------------------------------
class TestConstantsPinned:
def test_min_aircraft_is_three(self):
from services.constants import GPS_JAMMING_MIN_AIRCRAFT
assert GPS_JAMMING_MIN_AIRCRAFT == 3, (
"MIN_AIRCRAFT must be 3; raising it back to 5 brings back the "
"'jamming never shows' bug."
)
def test_min_ratio_is_0_20(self):
from services.constants import GPS_JAMMING_MIN_RATIO
assert GPS_JAMMING_MIN_RATIO == 0.20, (
"MIN_RATIO must be 0.20; raising it back to 0.30 brings back "
"the 'jamming never shows' bug."
)
# ---------------------------------------------------------------------------
# Overrides honored
# ---------------------------------------------------------------------------
class TestOverridesHonored:
def test_overrides_supersede_constants(self):
"""The public signature accepts overrides so an operator can
re-tune at the call site (e.g. for a more aggressive setup in
an active conflict zone) without editing the module constants."""
from services.fetchers.flights import detect_gps_jamming_zones
feed = [
{"hex": "a1", "lat": 40.1, "lng": -100.1, "nac_p": 3},
{"hex": "a2", "lat": 40.2, "lng": -100.2, "nac_p": 3},
]
# With defaults (min_aircraft=3) this is blocked. With override=2 it fires.
assert detect_gps_jamming_zones(feed) == []
zones = detect_gps_jamming_zones(feed, min_aircraft=2)
assert len(zones) == 1
# ---------------------------------------------------------------------------
# lon vs lng compatibility
# ---------------------------------------------------------------------------
class TestLonLngCompat:
def test_lon_key_accepted(self):
"""adsb.lol records arrive with ``lon`` (no g). The OpenSky merge
normalizes to ``lng`` but raw records flowing into the detector
may use either. Make sure both work."""
from services.fetchers.flights import detect_gps_jamming_zones
feed = [
{"hex": "a1", "lat": 40.1, "lon": -100.1, "nac_p": 0},
{"hex": "a2", "lat": 40.2, "lon": -100.2, "nac_p": 0},
{"hex": "a3", "lat": 40.3, "lon": -100.3, "nac_p": 0},
]
zones = detect_gps_jamming_zones(feed)
assert len(zones) == 1
# ---------------------------------------------------------------------------
# Empty / malformed inputs don't crash
# ---------------------------------------------------------------------------
class TestRobustness:
def test_empty_feed(self):
from services.fetchers.flights import detect_gps_jamming_zones
assert detect_gps_jamming_zones([]) == []
def test_none_feed(self):
"""The wrapper at the call site passes ``raw_flights_snapshot``
which could in principle be None on a startup race. Handle it."""
from services.fetchers.flights import detect_gps_jamming_zones
assert detect_gps_jamming_zones(None) == []
def test_records_missing_position_skipped(self):
from services.fetchers.flights import detect_gps_jamming_zones
feed = [
{"hex": "noloc", "nac_p": 0},
{"hex": "nolat", "lng": -100.0, "nac_p": 0},
{"hex": "nolng", "lat": 40.0, "nac_p": 0},
]
assert detect_gps_jamming_zones(feed) == []
@@ -0,0 +1,334 @@
"""Issue #302 (tg12): OpenClaw connect-info HMAC secret disclosure.
Before this change, ``GET /api/ai/connect-info?reveal=true`` returned the
full HMAC secret in the response body on every modal open AND the same
GET endpoint auto-bootstrapped (generated + persisted) the secret on a
mere read. Even gated to ``require_local_operator``, that put the full
secret into:
* browser visit history
* dev-tools network panel
* browser disk cache
* HAR exports
* screen captures / shoulder-surfing
Every single time the OpenClaw Connect modal opened.
After this change:
GET /api/ai/connect-info always returns the MASKED
fingerprint. No ?reveal param.
No side effects (auto-bootstrap
gone).
POST /api/ai/connect-info/bootstrap mints+persists the secret if
missing. Idempotent. Never
returns the full secret.
POST /api/ai/connect-info/reveal returns the full secret with
strict Cache-Control: no-store
headers. POST so the body
doesn't land in URL history.
POST /api/ai/connect-info/regenerate keeps the one-time-disclosure
for the new secret (regen IS a
deliberate destructive action).
Same no-store headers added.
These tests pin every property.
"""
from __future__ import annotations
import asyncio
from unittest.mock import patch
import pytest
from httpx import ASGITransport, AsyncClient
# ---------------------------------------------------------------------------
# Loopback test client. ``require_local_operator`` resolves true for
# request.client.host == "127.0.0.1"; FastAPI's TestClient sets it to
# "testclient" which isn't on the allowlist. Use raw ASGITransport.
# ---------------------------------------------------------------------------
@pytest.fixture
def loopback():
from main import app
class _Client:
def __init__(self, peer_ip: str = "127.0.0.1"):
self._loop = asyncio.new_event_loop()
self._transport = ASGITransport(app=app, client=(peer_ip, 12345))
self._base = f"http://{peer_ip}:8000"
def _do(self, method: str, url: str, **kw):
async def go():
async with AsyncClient(transport=self._transport, base_url=self._base) as ac:
return await ac.request(method, url, **kw)
return self._loop.run_until_complete(go())
def get(self, url, **kw): return self._do("GET", url, **kw)
def post(self, url, **kw): return self._do("POST", url, **kw)
def close(self): self._loop.close()
c = _Client()
yield c
c.close()
@pytest.fixture
def remote():
from main import app
class _Client:
def __init__(self):
self._loop = asyncio.new_event_loop()
self._transport = ASGITransport(app=app, client=("1.2.3.4", 12345))
self._base = "http://1.2.3.4:8000"
def _do(self, method: str, url: str, **kw):
async def go():
async with AsyncClient(transport=self._transport, base_url=self._base) as ac:
return await ac.request(method, url, **kw)
return self._loop.run_until_complete(go())
def get(self, url, **kw): return self._do("GET", url, **kw)
def post(self, url, **kw): return self._do("POST", url, **kw)
def close(self): self._loop.close()
c = _Client()
yield c
c.close()
@pytest.fixture
def stub_env(monkeypatch):
"""Isolate connect-info tests from the dev's real backend .env.
Pydantic ``Settings()`` reads from ``.env`` file directly on
instantiation, so monkey-patching ``os.environ`` isn't sufficient
the real ``OPENCLAW_HMAC_SECRET`` would leak through. Instead we
override ``get_settings()`` in the route module to return a fresh
``Settings`` instance whose env values are driven entirely by an
in-test dict, AND we replace ``_write_env_value`` so writes update
that same dict instead of touching the developer's filesystem.
Yields the dict so individual tests can pre-seed values or assert
that writes happened.
"""
import routers.ai_intel as ai_intel
import services.config as config
state: dict[str, str] = {}
class _FakeSettings:
@property
def OPENCLAW_HMAC_SECRET(self) -> str:
return state.get("OPENCLAW_HMAC_SECRET", "")
@property
def OPENCLAW_ACCESS_TIER(self) -> str:
return state.get("OPENCLAW_ACCESS_TIER", "restricted")
fake = _FakeSettings()
def _fake_get_settings():
return fake
# Route code calls ``get_settings.cache_clear()`` after writing the
# env. The production version is wrapped with ``@lru_cache``, so
# cache_clear exists. Attach a no-op shim here.
_fake_get_settings.cache_clear = lambda: None # type: ignore[attr-defined]
monkeypatch.setattr(config, "get_settings", _fake_get_settings)
def _fake_write_env_value(key: str, value: str) -> None:
state[key] = value
monkeypatch.setattr(ai_intel, "_write_env_value", _fake_write_env_value)
yield state
# ---------------------------------------------------------------------------
# GET /api/ai/connect-info — always masked, no auto-bootstrap
# ---------------------------------------------------------------------------
class TestGetConnectInfoMasking:
def test_returns_masked_when_secret_set(self, loopback, stub_env):
secret = "abcdef" + "0" * 38 + "wxyz"
stub_env["OPENCLAW_HMAC_SECRET"] = secret
r = loopback.get("/api/ai/connect-info")
assert r.status_code == 200
body = r.json()
# Body must NOT carry the full secret value anywhere.
assert secret not in r.text, (
"GET /api/ai/connect-info MUST NOT include the full HMAC "
"secret. Response body contained the secret value."
)
assert body["hmac_secret_set"] is True
assert body["masked_hmac_secret"].startswith("abcdef")
assert body["masked_hmac_secret"].endswith("wxyz")
assert "" in body["masked_hmac_secret"]
# Pre-fix field is gone.
assert "hmac_secret" not in body
def test_no_auto_bootstrap_when_secret_missing(self, loopback, stub_env):
"""Side-effect-on-GET was the second half of issue #302. A GET
with no secret configured must NOT mint one that should
require an explicit POST /bootstrap."""
r = loopback.get("/api/ai/connect-info")
assert r.status_code == 200
body = r.json()
assert body["hmac_secret_set"] is False
assert body["masked_hmac_secret"] == ""
# The bootstrap_behavior block should advertise the new flow.
assert body["bootstrap_behavior"]["auto_generates_when_missing"] is False
# And no _write_env_value call happened.
assert "OPENCLAW_HMAC_SECRET" not in stub_env
def test_no_reveal_query_param(self, loopback, stub_env):
"""Pre-fix, ?reveal=true would return the full secret. Post-fix
the param is silently ignored the response is the same as
without it (still masked, no leak)."""
secret = "abcdef" + "0" * 38 + "wxyz"
stub_env["OPENCLAW_HMAC_SECRET"] = secret
r = loopback.get("/api/ai/connect-info?reveal=true")
assert r.status_code == 200
assert secret not in r.text, (
"?reveal=true must be a no-op on GET — the full secret "
"MUST NOT come back in the response body."
)
# ---------------------------------------------------------------------------
# POST /api/ai/connect-info/bootstrap
# ---------------------------------------------------------------------------
class TestBootstrap:
def test_mints_when_missing(self, loopback, stub_env):
r = loopback.post("/api/ai/connect-info/bootstrap")
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["generated"] is True
assert body["hmac_secret_set"] is True
# Bootstrap must NOT return the full secret in-line.
assert "hmac_secret" not in body or not body.get("hmac_secret")
assert "" in body["masked_hmac_secret"]
# _write_env_value was actually called.
assert stub_env.get("OPENCLAW_HMAC_SECRET")
# The full value isn't echoed back in the response text either.
assert stub_env["OPENCLAW_HMAC_SECRET"] not in r.text
def test_idempotent_when_already_set(self, loopback, stub_env):
existing = "abcdef" + "0" * 38 + "wxyz"
stub_env["OPENCLAW_HMAC_SECRET"] = existing
r = loopback.post("/api/ai/connect-info/bootstrap")
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["generated"] is False
assert body["hmac_secret_set"] is True
# Existing secret untouched — value is still the seeded one.
assert stub_env["OPENCLAW_HMAC_SECRET"] == existing
# No full secret in the response.
assert existing not in r.text
# ---------------------------------------------------------------------------
# POST /api/ai/connect-info/reveal
# ---------------------------------------------------------------------------
class TestReveal:
def test_returns_full_secret_when_set(self, loopback, stub_env):
secret = "abcdef" + "0" * 38 + "wxyz"
stub_env["OPENCLAW_HMAC_SECRET"] = secret
r = loopback.post("/api/ai/connect-info/reveal")
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["hmac_secret"] == secret
def test_strict_cache_control_headers(self, loopback, stub_env):
"""The whole point of POST /reveal vs GET ?reveal=true is that
the response carries headers that prevent every cache layer
from persisting the secret."""
secret = "abcdef" + "0" * 38 + "wxyz"
stub_env["OPENCLAW_HMAC_SECRET"] = secret
r = loopback.post("/api/ai/connect-info/reveal")
cc = r.headers.get("cache-control", "")
assert "no-store" in cc, (
f"reveal MUST set Cache-Control: no-store — got {cc!r}"
)
assert "no-cache" in cc
# Pragma + Expires as well for HTTP/1.0 caches.
assert r.headers.get("pragma", "").lower() == "no-cache"
assert r.headers.get("expires") == "0"
def test_404_when_no_secret_configured(self, loopback, stub_env):
r = loopback.post("/api/ai/connect-info/reveal")
assert r.status_code == 404
# Hint should point at the bootstrap endpoint, not just say "404".
detail = r.json().get("detail", "")
assert "/bootstrap" in detail or "bootstrap" in detail.lower()
# ---------------------------------------------------------------------------
# POST /api/ai/connect-info/regenerate — still returns the new secret
# inline (deliberate destructive action), but with no-store headers.
# ---------------------------------------------------------------------------
class TestRegenerate:
def test_returns_new_secret_with_no_store_headers(self, loopback, stub_env):
# Seed an existing secret so we can prove it changes.
old = "oldold" + "0" * 38 + "1234"
stub_env["OPENCLAW_HMAC_SECRET"] = old
r = loopback.post("/api/ai/connect-info/regenerate")
assert r.status_code == 200
body = r.json()
assert body["ok"] is True
assert body["hmac_secret"]
assert body["hmac_secret"] != old
# no-store headers MUST be present so the new secret doesn't
# land in browser disk cache after the regenerate click.
cc = r.headers.get("cache-control", "")
assert "no-store" in cc and "no-cache" in cc
assert r.headers.get("pragma", "").lower() == "no-cache"
# ---------------------------------------------------------------------------
# Auth-gate regression — every endpoint still rejects anonymous remote
# callers. This is the property we already enforce for the rest of the
# operator-only surface; adding the three new endpoints to the audit
# coverage prevents a future refactor from dropping the dependency.
# ---------------------------------------------------------------------------
class TestAnonymousRejection:
@pytest.mark.parametrize(
"method,path,body",
[
("get", "/api/ai/connect-info", None),
("post", "/api/ai/connect-info/bootstrap", None),
("post", "/api/ai/connect-info/reveal", None),
("post", "/api/ai/connect-info/regenerate", None),
],
)
def test_remote_rejected(self, remote, method, path, body):
fn = getattr(remote, method)
r = fn(path, json=body) if body is not None else fn(path)
assert r.status_code == 403, (
f"{method.upper()} {path} must reject anonymous remote callers; "
f"got {r.status_code}"
)
@@ -0,0 +1,277 @@
"""Issue #298 (tg12): Sentinel credentials must live server-side.
Before the fix, ``frontend/src/components/SettingsPanel.tsx`` stored
``client_id`` and ``client_secret`` in ``localStorage`` /
``sessionStorage`` via the privacy storage helper, and the proxy routes
in ``backend/routers/tools.py`` REQUIRED those values to come in the
request body. Any same-origin script (XSS, malicious extension,
dev-tools HAR export) had read access to real third-party Sentinel
credentials.
After the fix:
* ``SENTINEL_CLIENT_ID`` and ``SENTINEL_CLIENT_SECRET`` are entries
in the ``api_settings.API_REGISTRY`` and are persisted via the
existing ``/api/settings/api-keys`` flow (admin-gated, .env-backed,
never returned to the browser).
* The proxy routes prefer request-body values for back-compat but
fall back to ``os.environ.get("SENTINEL_CLIENT_ID")`` /
``os.environ.get("SENTINEL_CLIENT_SECRET")`` when the body omits
them. The dashboard's ``sentinelHub.ts`` no longer sends credentials
in the body every request now hits the env path.
* When neither source has a value, the route returns a 400 with a
pointer to the API Keys panel rather than a curt "client_id and
client_secret required" message.
These tests cover the resolution order and the registry surface.
"""
from __future__ import annotations
from unittest.mock import patch, MagicMock
import pytest
# ---------------------------------------------------------------------------
# Helper: import the routes module fresh per test so monkey-patched
# environment variables are picked up by the route's os.environ.get call.
# (The lookup is per-request, not at import time, so this isn't strictly
# required — but it makes the test layout obvious.)
# ---------------------------------------------------------------------------
@pytest.fixture
def loopback_client():
"""ASGI client with peer IP 127.0.0.1 so the Sentinel routes' (post-#303)
``require_local_operator`` gate passes.
Built without a context manager so the privacy-core lifespan check
doesn't run in the test env.
"""
import asyncio
from httpx import ASGITransport, AsyncClient
from main import app
class _Loop:
def __init__(self):
self._loop = asyncio.new_event_loop()
self._transport = ASGITransport(app=app, client=("127.0.0.1", 12345))
self._base = "http://127.0.0.1:8000"
def _do(self, method: str, url: str, **kw):
async def go():
async with AsyncClient(transport=self._transport, base_url=self._base) as ac:
return await ac.request(method, url, **kw)
return self._loop.run_until_complete(go())
def get(self, url, **kw): return self._do("GET", url, **kw)
def post(self, url, **kw): return self._do("POST", url, **kw)
def put(self, url, **kw): return self._do("PUT", url, **kw)
def close(self): self._loop.close()
c = _Loop()
yield c
c.close()
# ---------------------------------------------------------------------------
# API_REGISTRY surface
# ---------------------------------------------------------------------------
class TestApiRegistry:
def test_sentinel_keys_registered(self):
"""Both Sentinel keys must be entries in API_REGISTRY so the
existing /api/settings/api-keys PUT flow can write them to .env."""
from services.api_settings import API_REGISTRY, ALLOWED_ENV_KEYS
ids = {row["id"] for row in API_REGISTRY}
assert "sentinel_client_id" in ids
assert "sentinel_client_secret" in ids
# Critical: ALLOWED_ENV_KEYS is the gate on which .env keys the
# API can mutate. If we forgot to add the env_key field on the
# registry rows, callers couldn't actually save the values.
assert "SENTINEL_CLIENT_ID" in ALLOWED_ENV_KEYS
assert "SENTINEL_CLIENT_SECRET" in ALLOWED_ENV_KEYS
def test_api_keys_put_accepts_sentinel_keys(self, loopback_client, monkeypatch, tmp_path):
"""End-to-end: PUT /api/settings/api-keys with SENTINEL_CLIENT_ID
+ SENTINEL_CLIENT_SECRET must persist to .env."""
import services.api_settings as api_settings
# Redirect both .env paths to tmp so the test doesn't mutate
# the developer's real backend .env.
tmp_env = tmp_path / ".env"
monkeypatch.setattr(api_settings, "ENV_PATH", tmp_env)
monkeypatch.setattr(api_settings, "OPERATOR_KEYS_ENV_PATH", tmp_path / "operator_api_keys.env")
r = loopback_client.put(
"/api/settings/api-keys",
json={
"SENTINEL_CLIENT_ID": "test-sentinel-id",
"SENTINEL_CLIENT_SECRET": "test-sentinel-secret",
},
)
assert r.status_code == 200, f"PUT failed: {r.text}"
body = r.json()
assert body.get("ok") is True
# File on disk should now carry both keys.
parsed = api_settings._parse_env_file(tmp_env)
assert parsed.get("SENTINEL_CLIENT_ID") == "test-sentinel-id"
assert parsed.get("SENTINEL_CLIENT_SECRET") == "test-sentinel-secret"
# ---------------------------------------------------------------------------
# Credential resolution — body wins, env is fallback, neither is 400
# ---------------------------------------------------------------------------
class TestSentinelTokenCredResolution:
def test_env_fallback_when_body_empty(self, loopback_client, monkeypatch):
"""No body credentials → backend reads .env values."""
monkeypatch.setenv("SENTINEL_CLIENT_ID", "env-id")
monkeypatch.setenv("SENTINEL_CLIENT_SECRET", "env-secret")
# Mock the upstream Copernicus call so we don't hit the network.
# Capture what was sent so we can prove env values were used.
captured: dict = {}
fake_resp = MagicMock()
fake_resp.status_code = 200
fake_resp.content = b'{"access_token": "stub", "expires_in": 300}'
def fake_post(url, *args, **kwargs):
captured["url"] = url
captured["data"] = kwargs.get("data", {})
return fake_resp
with patch("requests.post", side_effect=fake_post):
r = loopback_client.post(
"/api/sentinel/token",
data={}, # ← deliberately empty body
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
assert r.status_code == 200
# The forwarded creds must come from env, not from a stale cache
# or fallback string.
assert captured.get("data", {}).get("client_id") == "env-id"
assert captured.get("data", {}).get("client_secret") == "env-secret"
def test_body_credentials_win_over_env(self, loopback_client, monkeypatch):
"""Body values (back-compat path) must win when both sources
are present. This preserves the pre-#298 behavior for any
legacy callers that still post credentials."""
monkeypatch.setenv("SENTINEL_CLIENT_ID", "env-id")
monkeypatch.setenv("SENTINEL_CLIENT_SECRET", "env-secret")
captured: dict = {}
fake_resp = MagicMock()
fake_resp.status_code = 200
fake_resp.content = b'{"access_token": "stub"}'
def fake_post(url, *args, **kwargs):
captured["data"] = kwargs.get("data", {})
return fake_resp
with patch("requests.post", side_effect=fake_post):
r = loopback_client.post(
"/api/sentinel/token",
data={"client_id": "body-id", "client_secret": "body-secret"},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
assert r.status_code == 200
assert captured["data"]["client_id"] == "body-id"
assert captured["data"]["client_secret"] == "body-secret"
def test_400_when_neither_source_has_credentials(self, loopback_client, monkeypatch):
"""If body is empty AND env is empty, return 400 with a
friendly pointer to the API Keys panel not a curt
"required" message and not a 500."""
monkeypatch.delenv("SENTINEL_CLIENT_ID", raising=False)
monkeypatch.delenv("SENTINEL_CLIENT_SECRET", raising=False)
# If the route ever calls requests.post here, the gate is broken
# — empty creds should never produce an outbound HTTP call.
fake = MagicMock(side_effect=AssertionError(
"requests.post should not be called when no credentials are configured"
))
with patch("requests.post", fake):
r = loopback_client.post(
"/api/sentinel/token",
data={},
headers={"Content-Type": "application/x-www-form-urlencoded"},
)
assert r.status_code == 400
detail = r.json().get("detail", "")
# The pointer to the API Keys panel is what makes this non-hostile.
assert "API Keys panel" in detail or "SENTINEL_CLIENT_ID" in detail
assert fake.call_count == 0
class TestSentinelTileCredResolution:
def test_env_fallback_when_body_omits_credentials(self, loopback_client, monkeypatch):
"""Tile route: no body credentials → uses env values."""
monkeypatch.setenv("SENTINEL_CLIENT_ID", "env-id")
monkeypatch.setenv("SENTINEL_CLIENT_SECRET", "env-secret")
token_resp = MagicMock()
token_resp.status_code = 200
token_resp.json = MagicMock(return_value={"access_token": "stub", "expires_in": 300})
process_resp = MagicMock()
process_resp.status_code = 200
process_resp.content = b"<png bytes>"
process_resp.headers = {"content-type": "image/png"}
captured: list = []
def fake_post(url, *args, **kwargs):
captured.append({"url": url, "data": kwargs.get("data"), "json": kwargs.get("json")})
if "openid-connect/token" in url:
return token_resp
return process_resp
with patch("requests.post", side_effect=fake_post):
r = loopback_client.post(
"/api/sentinel/tile",
json={
# Note: no client_id / client_secret in body
"preset": "TRUE-COLOR",
"date": "2026-01-01",
"z": 6, "x": 30, "y": 20,
},
)
assert r.status_code == 200
# First call was the token mint; verify it used env creds.
token_call = next(c for c in captured if "openid-connect/token" in c["url"])
assert token_call["data"]["client_id"] == "env-id"
assert token_call["data"]["client_secret"] == "env-secret"
def test_400_when_neither_source_has_credentials(self, loopback_client, monkeypatch):
monkeypatch.delenv("SENTINEL_CLIENT_ID", raising=False)
monkeypatch.delenv("SENTINEL_CLIENT_SECRET", raising=False)
fake = MagicMock(side_effect=AssertionError(
"requests.post should not be called when no credentials are configured"
))
with patch("requests.post", fake):
r = loopback_client.post(
"/api/sentinel/tile",
json={
"preset": "TRUE-COLOR",
"date": "2026-01-01",
"z": 6, "x": 30, "y": 20,
},
)
assert r.status_code == 400
detail = r.json().get("detail", "")
assert "API Keys panel" in detail or "SENTINEL_CLIENT_ID" in detail
assert fake.call_count == 0
@@ -0,0 +1,252 @@
"""HF NUFORC fallback honors the rolling cutoff window.
Background
----------
The UAP sightings layer is sourced primarily from a live scrape of
nuforc.org. When that fails (Cloudflare 403, curl disabled on Windows,
wdtNonce regex stale, etc.) the code falls back to a static CSV mirror
hosted on Hugging Face at ``kcimc/NUFORC/nuforc_str.csv``.
The HF mirror is maintained by a third party and refreshed sporadically.
Pre-fix, the fallback parsed every row, sorted by ``occurred`` descending,
and took the top 250 **with no date cutoff**. When the HF mirror is
stale (its "newest" rows are ~2-3 years old), users saw a map full of
2022-2023 sightings labeled as the "last 60 days" layer.
These tests pin the new behavior:
* Rows older than ``_NUFORC_RECENT_DAYS`` are dropped before the take-top-N.
* If the HF mirror has nothing in the window, the fallback returns ``[]``
and logs ERROR (don't silently serve stale data).
* ``fetch_uap_sightings`` records the failure when BOTH paths fail, so
the layer shows as broken in the health registry instead of "fresh".
"""
from __future__ import annotations
import logging
from datetime import datetime as real_datetime
class _FixedDateTime(real_datetime):
"""A datetime whose utcnow() returns a pinned value, for deterministic
cutoff math. Subclasses real datetime so existing operations still work."""
@classmethod
def utcnow(cls):
return cls(2026, 5, 1, 12, 0, 0)
class _StubResponse:
status_code = 200
def __init__(self, text: str):
self.text = text
def _stub_geocode_cache(*_args, **_kwargs):
"""Pre-populated location cache so the fallback doesn't try to hit
Photon during the test."""
return {
"Denver, CO, USA": [39.7392, -104.9903],
"Seattle, WA, USA": [47.6062, -122.3321],
"Phoenix, AZ, USA": [33.4484, -112.0740],
}
def test_hf_fallback_drops_rows_older_than_60_days(monkeypatch):
"""Pre-fix: a row from 2023 would make it into the layer if it was
among the newest 250 in the HF mirror. Post-fix: it's filtered out
before we even count to 250."""
from services.fetchers import earth_observation as eo
# 2026-05-01 - 60 days = 2026-03-02. So 2026-03-01 is one day too old.
csv_text = (
"Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
'1,2026-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2026-04-16,"In-window sighting"\n'
'2,2023-06-01 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2023-06-02,"Three years old"\n'
'3,2022-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2022-01-16,"Even older"\n'
)
monkeypatch.setattr(eo, "datetime", _FixedDateTime)
monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
# If the cutoff is missing, the geocoder may still get called for the
# 2022/2023 rows. We assert geocoder is NEVER invoked for stale rows.
geocode_calls: list[str] = []
def _geocode_spy(location, city, state, country=""):
geocode_calls.append(location)
return None # already in cache, shouldn't be hit anyway
monkeypatch.setattr(eo, "_geocode_uap_location", _geocode_spy)
sightings = eo._build_uap_sightings_from_hf_mirror()
ids = [s["id"] for s in sightings]
assert ids == ["NUFORC-1"], f"only the 2026 row should survive: got {ids}"
# Stale rows must not have been geocoded — they should be dropped
# before the geocoding loop is reached.
assert geocode_calls == []
def test_hf_fallback_returns_empty_when_mirror_is_fully_stale(monkeypatch, caplog):
"""The smoking-gun case: the HF mirror is so stale that NO rows are
within the rolling window. Pre-fix returned 250 ancient rows. Post-fix
returns ``[]`` and logs ERROR so the operator knows the layer is dead."""
from services.fetchers import earth_observation as eo
csv_text = (
"Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
'1,2023-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2023-04-16,"Old"\n'
'2,2022-06-01 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2022-06-02,"Older"\n'
'3,2021-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2021-01-16,"Ancient"\n'
)
monkeypatch.setattr(eo, "datetime", _FixedDateTime)
monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
monkeypatch.setattr(eo, "_geocode_uap_location", lambda *a, **kw: None)
with caplog.at_level(logging.ERROR, logger="services.fetchers.earth_observation"):
sightings = eo._build_uap_sightings_from_hf_mirror()
assert sightings == []
# The error log should mention how many stale rows were dropped so the
# operator can tell the mirror is the problem (not "we got 0 rows" which
# could also mean the download failed).
relevant = [r for r in caplog.records if "HF fallback yielded 0 rows" in r.getMessage()]
assert relevant, "expected loud ERROR when HF mirror is fully stale"
# The message should report the count of dropped stale rows.
assert any("dropped 3" in r.getMessage() for r in relevant)
def test_hf_fallback_still_returns_data_when_some_rows_are_in_window(monkeypatch):
"""Mixed-age mirror: some rows in the window, some not. The fallback
should return only the in-window rows and not log the doomsday ERROR."""
from services.fetchers import earth_observation as eo
csv_text = (
"Sighting,Occurred,Location,Shape,Duration,Posted,Summary\n"
'1,2026-04-15 21:00:00 Local,"Denver, CO, USA",Triangle,5 minutes,2026-04-16,"Fresh"\n'
'2,2026-04-10 21:00:00 Local,"Seattle, WA, USA",Light,30 seconds,2026-04-10,"Also fresh"\n'
'3,2020-01-15 20:00:00 Local,"Phoenix, AZ, USA",Disk,2 minutes,2020-01-16,"Ancient"\n'
)
monkeypatch.setattr(eo, "datetime", _FixedDateTime)
monkeypatch.setattr(eo, "fetch_with_curl", lambda *a, **kw: _StubResponse(csv_text))
monkeypatch.setattr(eo, "_load_nuforc_location_cache", _stub_geocode_cache)
monkeypatch.setattr(eo, "_save_nuforc_location_cache", lambda cache: None)
monkeypatch.setattr(eo, "_geocode_uap_location", lambda *a, **kw: None)
sightings = eo._build_uap_sightings_from_hf_mirror()
ids = sorted(s["id"] for s in sightings)
assert ids == ["NUFORC-1", "NUFORC-2"], f"only in-window rows should appear: got {ids}"
def test_fetch_uap_sightings_marks_failure_when_both_paths_empty(monkeypatch, caplog):
"""When the live path raises AND the HF fallback returns empty,
``fetch_uap_sightings`` must:
* NOT mark the layer fresh (pre-fix bug: it did, so the layer
showed as healthy-but-empty for days)
* call ``assert_canary("uap_sightings", 0)`` so the health
registry surfaces the broken layer
* log an ERROR with the live-path exception for debugging
"""
from services.fetchers import earth_observation as eo
from services.fetchers import _store
monkeypatch.setattr(_store, "is_any_active", lambda layer: True)
monkeypatch.setattr(eo, "_load_nuforc_sightings_cache", lambda force_refresh=False: None)
def _boom():
raise RuntimeError("NUFORC live: zero rows pulled across 3 months")
monkeypatch.setattr(eo, "_build_recent_uap_sightings", _boom)
monkeypatch.setattr(eo, "_build_uap_sightings_from_hf_mirror", lambda: [])
marked: list[str] = []
monkeypatch.setattr(eo, "_mark_fresh", lambda *keys: marked.extend(keys))
canary_calls: list[tuple[str, int]] = []
import services.slo as slo
monkeypatch.setattr(
slo, "assert_canary", lambda key, value: canary_calls.append((key, int(value)))
)
with caplog.at_level(logging.ERROR, logger="services.fetchers.earth_observation"):
eo.fetch_uap_sightings()
assert marked == [], "broken layer must NOT be marked fresh"
assert canary_calls == [("uap_sightings", 0)], (
f"expected canary trip when both paths fail; got {canary_calls}"
)
# The live error message should propagate into the error log so the
# operator can tell live failed AND fallback was empty (not the other
# way around).
assert any(
"both live NUFORC and HF fallback" in r.getMessage()
for r in caplog.records
)
def test_fetch_uap_sightings_succeeds_when_fallback_returns_data(monkeypatch):
"""Positive path: live fails, fallback returns rows. The layer is
populated and marked fresh; assert_canary is NOT tripped (we only
trip the canary when the layer has zero data)."""
from services.fetchers import earth_observation as eo
from services.fetchers import _store
monkeypatch.setattr(_store, "is_any_active", lambda layer: True)
monkeypatch.setattr(eo, "_load_nuforc_sightings_cache", lambda force_refresh=False: None)
monkeypatch.setattr(
eo, "_build_recent_uap_sightings", lambda: (_ for _ in ()).throw(RuntimeError("live down"))
)
fallback_rows = [{"id": "NUFORC-fb-1", "date_time": "2026-04-20", "lat": 0.0, "lng": 0.0}]
monkeypatch.setattr(eo, "_build_uap_sightings_from_hf_mirror", lambda: fallback_rows)
monkeypatch.setattr(eo, "_save_nuforc_sightings_cache", lambda s: None)
marked: list[str] = []
monkeypatch.setattr(eo, "_mark_fresh", lambda *keys: marked.extend(keys))
canary_calls: list[tuple[str, int]] = []
import services.slo as slo
monkeypatch.setattr(
slo, "assert_canary", lambda key, value: canary_calls.append((key, int(value)))
)
eo.fetch_uap_sightings()
assert marked == ["uap_sightings"]
assert canary_calls == [], "canary should not trip when fallback supplies data"
def test_uap_scheduler_runs_weekly_not_daily():
"""The cron job for the UAP layer must be configured for Mondays at
12:00 UTC, not daily. Daily was the pre-fix default; weekly matches
the layer's stated cadence (a rolling 60-day digest) and keeps load
on nuforc.org light."""
from services import data_fetcher
src = data_fetcher.__file__
with open(src, "r", encoding="utf-8") as f:
text = f.read()
# Anchor on the scheduler block by id, then assert the cron triggers.
assert "uap_sightings_weekly" in text, (
"scheduler id should be uap_sightings_weekly (was uap_sightings_daily pre-fix)"
)
# The day_of_week directive is the difference between daily and weekly.
# If somebody flips it back to daily, this fires.
weekly_block = text.split("uap_sightings_weekly", 1)[0]
# Walk backwards for the matching add_job call.
add_job_idx = weekly_block.rfind("add_job(")
assert add_job_idx >= 0, "could not locate add_job block for UAP scheduler"
job_block = text[add_job_idx : text.find(")", text.index("uap_sightings_weekly")) + 1]
assert 'day_of_week="mon"' in job_block, (
f"expected day_of_week='mon' in UAP scheduler block:\n{job_block}"
)
@@ -46,12 +46,18 @@ function prepareBuildTree() {
const stagedLayoutPath = path.join(buildFrontendDir, 'src', 'app', 'layout.tsx');
if (fs.existsSync(stagedLayoutPath)) {
const layoutSource = fs.readFileSync(stagedLayoutPath, 'utf8');
// CRLF compatibility: on Windows checkouts without ``core.autocrlf=input``
// (the default) layout.tsx has CRLF line endings, but the original regexes
// only matched LF. The strip silently no-op'd, ``force-dynamic`` stayed,
// and Next's static-export refused to render ``/_not-found`` ("Page with
// `dynamic = \"force-dynamic\"` couldn't be exported"). Use ``\r?\n`` so
// the strip works regardless of line-ending normalization.
fs.writeFileSync(
stagedLayoutPath,
layoutSource
.replace(/\n\/\/ The dashboard is a live local runtime[\s\S]*?client polling ever hydrates\.\n/g, '\n')
.replace(/\nexport const dynamic = ['"]force-dynamic['"];\n/g, '\n')
.replace(/\nexport const revalidate = 0;\n/g, '\n'),
.replace(/\r?\n\/\/ The dashboard is a live local runtime[\s\S]*?client polling ever hydrates\.\r?\n/g, '\n')
.replace(/\r?\nexport const dynamic = ['"]force-dynamic['"];\r?\n/g, '\n')
.replace(/\r?\nexport const revalidate = 0;\r?\n/g, '\n'),
);
}
@@ -0,0 +1,169 @@
/**
* Issue #298 (tg12): Sentinel credentials must no longer live in browser
* storage, and the proxy calls must not forward them in request bodies.
* These tests pin both invariants on ``lib/sentinelHub``:
*
* 1. ``migrateLegacySentinelBrowserKeys()`` clears the legacy keys
* idempotently and reports what it cleared.
* 2. ``fetchSentinelTile()`` and ``getSentinelToken()`` POST WITHOUT
* ``client_id`` or ``client_secret`` in the body the backend
* resolves credentials from its ``.env``. A future refactor that
* accidentally re-introduces browser-storage reads (e.g. by
* restoring ``getSentinelCredentials()`` and forwarding it) gets a
* loud test failure here rather than a silent privacy regression.
* 3. ``checkBackendSentinelStatus()`` queries ``/api/settings/api-keys``
* and returns true only when both Sentinel keys report ``is_set``.
*/
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import {
migrateLegacySentinelBrowserKeys,
fetchSentinelTile,
getSentinelToken,
checkBackendSentinelStatus,
refreshSentinelStatus,
} from '@/lib/sentinelHub';
const originalFetch = globalThis.fetch;
describe('lib/sentinelHub — issue #298 server-side credentials', () => {
beforeEach(() => {
window.localStorage.clear();
window.sessionStorage.clear();
refreshSentinelStatus();
});
afterEach(() => {
globalThis.fetch = originalFetch;
window.localStorage.clear();
window.sessionStorage.clear();
refreshSentinelStatus();
});
describe('migrateLegacySentinelBrowserKeys', () => {
it('clears legacy localStorage keys and reports what it cleared', () => {
window.localStorage.setItem('sb_sentinel_client_id', 'sh-leaked-id');
window.localStorage.setItem('sb_sentinel_client_secret', 'leaked-secret');
window.localStorage.setItem('sb_sentinel_instance_id', 'leaked-instance');
const result = migrateLegacySentinelBrowserKeys();
expect(window.localStorage.getItem('sb_sentinel_client_id')).toBeNull();
expect(window.localStorage.getItem('sb_sentinel_client_secret')).toBeNull();
expect(window.localStorage.getItem('sb_sentinel_instance_id')).toBeNull();
expect(result.cleared.sort()).toEqual([
'sb_sentinel_client_id',
'sb_sentinel_client_secret',
'sb_sentinel_instance_id',
].sort());
});
it('clears sessionStorage too (privacy-strict mode used to put them there)', () => {
window.sessionStorage.setItem('sb_sentinel_client_id', 'sh-session-id');
window.sessionStorage.setItem('sb_sentinel_client_secret', 'session-secret');
const result = migrateLegacySentinelBrowserKeys();
expect(window.sessionStorage.getItem('sb_sentinel_client_id')).toBeNull();
expect(window.sessionStorage.getItem('sb_sentinel_client_secret')).toBeNull();
expect(result.cleared).toContain('sb_sentinel_client_id');
expect(result.cleared).toContain('sb_sentinel_client_secret');
});
it('is idempotent — calling it on a clean store reports nothing cleared', () => {
const result = migrateLegacySentinelBrowserKeys();
expect(result.cleared).toEqual([]);
});
});
describe('proxy requests no longer forward credentials', () => {
it('fetchSentinelTile POSTs without client_id/client_secret in the body', async () => {
// Plant credentials in browser storage to prove they would NOT be
// picked up even if present. Pre-#298, this would have been read
// from localStorage and posted in the body.
window.localStorage.setItem('sb_sentinel_client_id', 'sh-leaked-id');
window.localStorage.setItem('sb_sentinel_client_secret', 'leaked-secret');
const fetchMock = vi.fn(async () => new Response(new ArrayBuffer(0), { status: 200 }));
globalThis.fetch = fetchMock as unknown as typeof globalThis.fetch;
await fetchSentinelTile(6, 30, 20, 'TRUE-COLOR', '2026-01-01');
expect(fetchMock).toHaveBeenCalledTimes(1);
const [, init] = fetchMock.mock.calls[0] as [unknown, RequestInit];
const body = JSON.parse(String(init.body));
expect(body).not.toHaveProperty('client_id');
expect(body).not.toHaveProperty('client_secret');
// Sanity: the legitimate fields are still there.
expect(body).toMatchObject({ preset: 'TRUE-COLOR', date: '2026-01-01', z: 6, x: 30, y: 20 });
});
it('getSentinelToken POSTs with an empty form body (backend uses env)', async () => {
window.localStorage.setItem('sb_sentinel_client_id', 'sh-leaked-id');
window.localStorage.setItem('sb_sentinel_client_secret', 'leaked-secret');
const fetchMock = vi.fn(async () =>
new Response(JSON.stringify({ access_token: 'stub', expires_in: 300 }), { status: 200 }),
);
globalThis.fetch = fetchMock as unknown as typeof globalThis.fetch;
const token = await getSentinelToken();
expect(token).toBe('stub');
expect(fetchMock).toHaveBeenCalledTimes(1);
const [, init] = fetchMock.mock.calls[0] as [unknown, RequestInit];
const body = String(init.body);
// Body is a URLSearchParams stringification. We assert that the
// leaked credential never appears in it.
expect(body).not.toContain('sh-leaked-id');
expect(body).not.toContain('leaked-secret');
});
});
describe('checkBackendSentinelStatus', () => {
it('returns true when both Sentinel keys report is_set on /api/settings/api-keys', async () => {
const fetchMock = vi.fn(async (input: unknown) => {
const url = String(input);
if (url.endsWith('/api/settings/api-keys')) {
return new Response(
JSON.stringify([
{ id: 'sentinel_client_id', env_key: 'SENTINEL_CLIENT_ID', is_set: true },
{ id: 'sentinel_client_secret', env_key: 'SENTINEL_CLIENT_SECRET', is_set: true },
{ id: 'opensky_client_id', env_key: 'OPENSKY_CLIENT_ID', is_set: false },
]),
{ status: 200 },
);
}
return new Response('not found', { status: 404 });
});
globalThis.fetch = fetchMock as unknown as typeof globalThis.fetch;
const configured = await checkBackendSentinelStatus();
expect(configured).toBe(true);
});
it('returns false when only one of the two keys is set', async () => {
const fetchMock = vi.fn(async () =>
new Response(
JSON.stringify([
{ id: 'sentinel_client_id', env_key: 'SENTINEL_CLIENT_ID', is_set: true },
{ id: 'sentinel_client_secret', env_key: 'SENTINEL_CLIENT_SECRET', is_set: false },
]),
{ status: 200 },
),
);
globalThis.fetch = fetchMock as unknown as typeof globalThis.fetch;
const configured = await checkBackendSentinelStatus();
expect(configured).toBe(false);
});
it('fails safely (false) when the backend errors', async () => {
const fetchMock = vi.fn(async () => { throw new Error('network down'); });
globalThis.fetch = fetchMock as unknown as typeof globalThis.fetch;
const configured = await checkBackendSentinelStatus();
expect(configured).toBe(false);
});
});
});
+16
View File
@@ -39,6 +39,7 @@ import { useFeedHealth } from '@/hooks/useFeedHealth';
import { useKeyboardShortcuts } from '@/hooks/useKeyboardShortcuts';
import KeyboardShortcutsOverlay from '@/components/KeyboardShortcutsOverlay';
import AlertToast from '@/components/AlertToast';
import AisUpstreamBanner from '@/components/AisUpstreamBanner';
import { useAlertToasts } from '@/hooks/useAlertToasts';
import { useWatchlist } from '@/hooks/useWatchlist';
import WatchlistWidget from '@/components/WatchlistWidget';
@@ -50,6 +51,7 @@ import {
hasSentinelInfoBeenSeen,
markSentinelInfoSeen,
hasSentinelCredentials,
checkBackendSentinelStatus,
} from '@/lib/sentinelHub';
import { useTranslation } from '@/i18n';
import { LocateBar } from './LocateBar';
@@ -107,6 +109,15 @@ export default function Dashboard() {
useEffect(() => {
localStorage.setItem('sb_ticker_open', tickerOpen.toString());
}, [tickerOpen]);
// Issue #298: kick the one-time backend Sentinel-status check on mount.
// This populates the cached value that ``hasSentinelCredentials()`` reads
// synchronously elsewhere (MaplibreViewer's tile-URL memo, the
// Sentinel-info modal flow). Fire-and-forget — the cache stays false
// until resolved so the UI fails safely.
useEffect(() => {
void checkBackendSentinelStatus();
}, []);
const [settingsOpen, setSettingsOpen] = useState(false);
const [legendOpen, setLegendOpen] = useState(false);
const [shortcutsOpen, setShortcutsOpen] = useState(false);
@@ -923,6 +934,11 @@ export default function Dashboard() {
onFlyTo={handleFlyTo}
/>
{/* AIS UPSTREAM OUTAGE BANNER renders only when AIS is configured
but the WebSocket upstream is unreachable. Tells users the empty
ocean isn't their fault. */}
<AisUpstreamBanner />
{/* ONBOARDING MODAL */}
{showOnboarding && (
<OnboardingModal
+120 -19
View File
@@ -357,8 +357,15 @@ function ConnectModalBody({ apiEndpoint, handleCopy, copied }: ConnectModalBodyP
const [riskAccepted, setRiskAccepted] = React.useState(false);
const [accessTier, setAccessTier] = React.useState<'restricted' | 'full'>('restricted');
const [connectionMode, setConnectionMode] = React.useState<'local' | 'remote'>('local');
// hmacSecret holds the FULL secret once the operator has clicked
// Reveal (or after a regenerate). maskedHmacSecret is the safe-to-show
// fingerprint returned by GET /api/ai/connect-info and is loaded on
// mount. The two are independent state slots so a stale full secret
// can never leak back into the UI after a regenerate.
const [hmacSecret, setHmacSecret] = React.useState('');
const [maskedHmacSecret, setMaskedHmacSecret] = React.useState('');
const [hmacLoading, setHmacLoading] = React.useState(false);
const [revealing, setRevealing] = React.useState(false);
const [tierSaving, setTierSaving] = React.useState(false);
const [showAdvanced, setShowAdvanced] = React.useState(false);
const [showResetConfirm, setShowResetConfirm] = React.useState(false);
@@ -381,16 +388,40 @@ function ConnectModalBody({ apiEndpoint, handleCopy, copied }: ConnectModalBodyP
const [torError, setTorError] = React.useState('');
const [torOnion, setTorOnion] = React.useState('');
// Fetch connect-info + node status on mount
// Issue #302 (tg12): the full HMAC secret no longer travels through
// GET /api/ai/connect-info on every modal open. The flow is now:
//
// 1. GET /api/ai/connect-info — always returns the masked fingerprint
// (first6 + bullets + last4). `hmacSecret` stays empty until the
// operator clicks the Reveal (eye) button below.
// 2. POST /api/ai/connect-info/bootstrap — fires once on mount if the
// backend reports `hmac_secret_set: false`. Idempotent and never
// returns the secret in the response.
// 3. POST /api/ai/connect-info/reveal — fires when the operator clicks
// Reveal or Copy without the secret yet loaded. Returns the full
// secret with strict `Cache-Control: no-store` so it doesn't land
// in browser caches or HAR exports.
React.useEffect(() => {
(async () => {
try {
setHmacLoading(true);
const res = await fetch(`${API_BASE}/api/ai/connect-info?reveal=true`);
if (res.ok) {
const data = await res.json();
setHmacSecret(data.hmac_secret || '');
setAccessTier(data.access_tier === 'full' ? 'full' : 'restricted');
const res = await fetch(`${API_BASE}/api/ai/connect-info`);
if (!res.ok) return;
const data = await res.json();
setMaskedHmacSecret(data.masked_hmac_secret || '');
setAccessTier(data.access_tier === 'full' ? 'full' : 'restricted');
// Transparent first-use bootstrap. Mirrors the pre-#302 UX of
// "open modal → secret exists" without the GET side-effect.
if (!data.hmac_secret_set) {
const bootRes = await fetch(
`${API_BASE}/api/ai/connect-info/bootstrap`,
{ method: 'POST' },
);
if (bootRes.ok) {
const bootData = await bootRes.json();
setMaskedHmacSecret(bootData.masked_hmac_secret || '');
}
}
} catch { /* ignore */ }
finally { setHmacLoading(false); }
@@ -477,8 +508,17 @@ function ConnectModalBody({ apiEndpoint, handleCopy, copied }: ConnectModalBodyP
const res = await fetch(`${API_BASE}/api/settings/agent/reset-all`, { method: 'POST' });
const data = await res.json();
if (data.ok) {
// Update local state with new credentials
if (data.new_hmac_secret) setHmacSecret(data.new_hmac_secret);
// Update local state with new credentials. reset-all returns
// the new HMAC secret in-band (same one-time-disclosure rule
// as /regenerate — a deliberate destructive action). Refresh
// both slots so the masked display stays in sync.
if (data.new_hmac_secret) {
setHmacSecret(data.new_hmac_secret);
const s = String(data.new_hmac_secret);
setMaskedHmacSecret(
s.length > 10 ? s.slice(0, 6) + '•'.repeat(8) + s.slice(-4) : '•'.repeat(16),
);
}
if (data.new_onion) {
setTorOnion(data.new_onion);
setRemoteUrl(data.new_onion);
@@ -502,13 +542,41 @@ function ConnectModalBody({ apiEndpoint, handleCopy, copied }: ConnectModalBodyP
finally { setTierSaving(false); }
};
// Issue #302: POST /reveal returns the full secret with strict
// no-store headers. Lazily fetched — never on mount. Returns the
// secret string so callers can copy it immediately without waiting
// for React state propagation.
const revealHmacSecret = async (): Promise<string> => {
if (hmacSecret) return hmacSecret;
setRevealing(true);
try {
const res = await fetch(`${API_BASE}/api/ai/connect-info/reveal`, {
method: 'POST',
});
if (!res.ok) return '';
const data = await res.json();
const secret = String(data.hmac_secret || '');
setHmacSecret(secret);
return secret;
} catch {
return '';
} finally {
setRevealing(false);
}
};
const handleRegenerate = async () => {
setRegenerating(true);
try {
const res = await fetch(`${API_BASE}/api/ai/connect-info/regenerate`, { method: 'POST' });
if (res.ok) {
const data = await res.json();
// Regenerate is a deliberate destructive action — operator needs
// to see the new secret once to update their OpenClaw config.
// Both the full and masked forms refresh in one shot.
setHmacSecret(data.hmac_secret || '');
setMaskedHmacSecret(data.masked_hmac_secret || '');
setShowSecret(true);
}
} catch { /* ignore */ }
finally { setRegenerating(false); }
@@ -543,9 +611,17 @@ function ConnectModalBody({ apiEndpoint, handleCopy, copied }: ConnectModalBodyP
finally { setNodeToggling(false); }
};
const maskedSecret = hmacSecret
? hmacSecret.slice(0, 6) + '\u2022'.repeat(8) + hmacSecret.slice(-4)
: '\u2022'.repeat(16);
// Issue #302: prefer the server-supplied fingerprint
// (maskedHmacSecret) \u2014 it's filled on mount via the (no-secret) GET.
// If the operator has clicked Reveal, fall through to deriving the
// mask from the in-memory full secret so we keep the same shape
// (first6 + bullets + last4) regardless of source. Final fallback
// (no secret loaded yet) is a generic bullet string.
const maskedSecret =
maskedHmacSecret ||
(hmacSecret
? hmacSecret.slice(0, 6) + '\u2022'.repeat(8) + hmacSecret.slice(-4)
: '\u2022'.repeat(16));
// Resolve the endpoint URL
const resolvedUrl = connectionMode === 'local'
@@ -672,10 +748,15 @@ function ConnectModalBody({ apiEndpoint, handleCopy, copied }: ConnectModalBodyP
return lines.join('\n');
};
const displaySnippet = buildSnippet(maskedSecret);
const copySnippet = buildSnippet(hmacSecret);
const handleCopySnippet = () => {
navigator.clipboard.writeText(copySnippet);
// Issue #302: the copy snippet needs the FULL secret. Pre-#302 we kept
// it in memory from the GET-with-reveal load; now we lazy-fetch via
// POST /reveal only when the operator actually clicks Copy. If they
// already revealed, the in-memory value is reused (no extra request).
const handleCopySnippet = async () => {
const secret = hmacSecret || (await revealHmacSecret());
if (!secret) return;
navigator.clipboard.writeText(buildSnippet(secret));
setSnippetCopied(true);
setTimeout(() => setSnippetCopied(false), 2000);
};
@@ -913,18 +994,38 @@ function ConnectModalBody({ apiEndpoint, handleCopy, copied }: ConnectModalBodyP
</div>
<div className="flex items-center gap-2">
<code className="flex-1 bg-black/60 border border-violet-800/40 px-3 py-2 text-xs font-mono text-violet-300 overflow-hidden text-ellipsis">
{showSecret ? hmacSecret : maskedSecret}
{/* Issue #302: when the operator hasn't clicked
Reveal yet, hmacSecret is empty and we fall
back to maskedHmacSecret (the safe fingerprint
returned by GET /api/ai/connect-info). */}
{showSecret && hmacSecret ? hmacSecret : (maskedHmacSecret || maskedSecret)}
</code>
<button
onClick={() => setShowSecret(!showSecret)}
className="p-2 bg-violet-600/20 border border-violet-500/40 text-violet-400 hover:bg-violet-600/40 transition-colors shrink-0"
onClick={async () => {
if (showSecret) {
setShowSecret(false);
return;
}
// Need the full secret in state before showing it.
const secret = await revealHmacSecret();
if (secret) setShowSecret(true);
}}
disabled={revealing}
className="p-2 bg-violet-600/20 border border-violet-500/40 text-violet-400 hover:bg-violet-600/40 transition-colors shrink-0 disabled:opacity-50"
title={showSecret ? 'Hide' : 'Reveal'}
>
{showSecret ? <EyeOff size={14} /> : <Eye size={14} />}
</button>
<button
onClick={() => handleCopy(hmacSecret)}
className="p-2 bg-violet-600/20 border border-violet-500/40 text-violet-400 hover:bg-violet-600/40 transition-colors shrink-0"
onClick={async () => {
// Copy needs the full secret. Fetch it lazily if
// the operator hasn't clicked Reveal yet — no
// point making them reveal first just to copy.
const secret = hmacSecret || (await revealHmacSecret());
if (secret) handleCopy(secret);
}}
disabled={revealing}
className="p-2 bg-violet-600/20 border border-violet-500/40 text-violet-400 hover:bg-violet-600/40 transition-colors shrink-0 disabled:opacity-50"
title="Copy key"
>
{copied ? <Check size={14} /> : <Copy size={14} />}
@@ -0,0 +1,61 @@
/**
* AisUpstreamBanner visible notice that AIS ship data is unavailable
* because the upstream provider (AISStream) is offline.
*
* Renders nothing when AIS is healthy or when AIS isn't configured at all.
* Mounted at the app shell level so users see it before they wonder why
* the ocean looks empty.
*/
import { useState } from 'react';
import { useAisUpstreamHealth } from '@/hooks/useAisUpstreamHealth';
export function AisUpstreamBanner() {
const health = useAisUpstreamHealth();
const [dismissed, setDismissed] = useState(false);
if (!health || !health.aisEnabled || health.connected || dismissed) {
return null;
}
// Format the staleness for the operator. ``null`` means we never received
// anything since startup; otherwise show minutes if > 60s.
let stalenessLabel = 'never received';
if (health.lastMsgAgeSeconds != null) {
const minutes = Math.floor(health.lastMsgAgeSeconds / 60);
if (minutes >= 1) {
stalenessLabel = `last update ${minutes} min ago`;
} else {
stalenessLabel = `last update ${health.lastMsgAgeSeconds}s ago`;
}
}
return (
<div
role="status"
aria-live="polite"
className="pointer-events-auto fixed top-3 left-1/2 z-[100] -translate-x-1/2 max-w-[640px] rounded-md border border-amber-500/60 bg-amber-900/85 px-4 py-2 text-sm text-amber-50 shadow-lg backdrop-blur"
>
<div className="flex items-start gap-3">
<span aria-hidden className="mt-0.5 text-amber-300"></span>
<div className="flex-1">
<div className="font-semibold">Ship data temporarily unavailable</div>
<div className="text-xs opacity-90">
AISStream upstream is offline ({stalenessLabel}). The map will
refill once their service comes back online nothing is wrong
with your install.
</div>
</div>
<button
type="button"
onClick={() => setDismissed(true)}
aria-label="Dismiss"
className="text-amber-200 hover:text-white"
>
</button>
</div>
</div>
);
}
export default AisUpstreamBanner;
+47 -11
View File
@@ -249,34 +249,70 @@ const VESSEL_TYPE_WIKI: Record<string, string> = {
type FlightTrailPoint = { lat?: number; lng?: number; alt?: number; ts?: number } | number[];
function formatObservedDuration(seconds: number): string {
// Compact "1h 14m" / "23m" / "45s" — matches the density of the rest
// of the flight tooltip. < 60s is shown as "<1m" so the user knows
// we've JUST started observing this hex (cumulative will still be 0).
if (!Number.isFinite(seconds) || seconds <= 0) return '<1m';
if (seconds < 60) return '<1m';
const totalMinutes = Math.floor(seconds / 60);
const hours = Math.floor(totalMinutes / 60);
const minutes = totalMinutes % 60;
if (hours > 0) return `${hours}h ${minutes}m`;
return `${minutes}m`;
}
function EmissionsEstimateBlock({ flight }: { flight: any }) {
const emissions = flight?.emissions;
const context = emissions ? 'Model-based cruise estimate' : null;
// Cumulative fuel/CO2 since the backend first saw this hex this
// flight session. Prefer these big numbers — the user explicitly
// wanted "the actual fuel that has been burned", not the rate.
// Rates are still shown below as smaller context.
const observedSec = Number(emissions?.observed_seconds ?? 0);
const fuelBurned = Number(emissions?.fuel_gallons_burned ?? 0);
const co2Emitted = Number(emissions?.co2_kg_emitted ?? 0);
const haveCumulative = emissions && observedSec > 0;
return (
<div className="border-b border-[var(--border-primary)] pb-2">
<span className="text-[var(--text-muted)] text-[10px] block mb-1.5">EMISSIONS ESTIMATE</span>
<div className="flex gap-3">
<div className="flex-1 bg-[var(--bg-primary)]/50 border border-[var(--border-primary)] px-2 py-1.5">
<div className="text-[11px] text-[var(--text-muted)] tracking-widest">FUEL RATE</div>
<div className="text-xs font-bold text-orange-400">
{emissions ? (
<>{emissions.fuel_gph} <span className="text-[11px] text-[var(--text-muted)] font-normal">GPH</span></>
<div className="text-[11px] text-[var(--text-muted)] tracking-widest">FUEL BURNED</div>
<div className="text-sm font-bold text-orange-400">
{haveCumulative ? (
<>{fuelBurned.toLocaleString(undefined, { maximumFractionDigits: 1 })} <span className="text-[11px] text-[var(--text-muted)] font-normal">gal</span></>
) : emissions ? (
<span className="text-[var(--text-muted)] font-normal text-xs"></span>
) : 'UNKNOWN'}
</div>
{emissions && (
<div className="text-[10px] text-[var(--text-muted)] mt-0.5">
@ {emissions.fuel_gph} gph
</div>
)}
</div>
<div className="flex-1 bg-[var(--bg-primary)]/50 border border-[var(--border-primary)] px-2 py-1.5">
<div className="text-[11px] text-[var(--text-muted)] tracking-widest">CO2 RATE</div>
<div className="text-xs font-bold text-red-400">
{emissions ? (
<>{emissions.co2_kg_per_hour.toLocaleString()} <span className="text-[11px] text-[var(--text-muted)] font-normal">KG/HR</span></>
<div className="text-[11px] text-[var(--text-muted)] tracking-widest">CO2 EMITTED</div>
<div className="text-sm font-bold text-red-400">
{haveCumulative ? (
<>{co2Emitted.toLocaleString(undefined, { maximumFractionDigits: 1 })} <span className="text-[11px] text-[var(--text-muted)] font-normal">kg</span></>
) : emissions ? (
<span className="text-[var(--text-muted)] font-normal text-xs"></span>
) : 'UNKNOWN'}
</div>
{emissions && (
<div className="text-[10px] text-[var(--text-muted)] mt-0.5">
@ {emissions.co2_kg_per_hour.toLocaleString()} kg/hr
</div>
)}
</div>
</div>
{context && (
{emissions && (
<div className="mt-1.5 text-[10px] text-[var(--text-muted)] leading-relaxed">
{context}
{haveCumulative
? `Observed in flight for ${formatObservedDuration(observedSec)} · model-based cruise estimate`
: 'Just observed · totals will appear on next refresh'}
</div>
)}
</div>
+40 -6
View File
@@ -140,17 +140,51 @@ const OnboardingModal = React.memo(function OnboardingModal({
].join('\n');
const remoteAgentNeedsTor = agentMode === 'remote' && !torAddress;
// Issue #302 (tg12): the full HMAC secret no longer comes back from
// GET /api/ai/connect-info. We fetch metadata + the masked fingerprint
// first; if the operator has explicitly asked to see the key (the
// ``reveal`` flag), we follow up with POST /api/ai/connect-info/reveal
// (after a transparent POST /bootstrap if the secret hasn't been
// minted yet) which carries the secret with strict no-store headers.
const fetchAgentConnectInfo = async (reveal = true) => {
setAgentLoading(true);
setAgentMsg(null);
try {
const res = await fetch(`/api/ai/connect-info?reveal=${reveal ? 'true' : 'false'}`);
const data = await res.json().catch(() => ({}));
if (!res.ok || data?.ok === false) {
throw new Error(data?.detail || 'Could not prepare agent credentials.');
// 1) GET metadata + masked fingerprint.
const metaRes = await fetch('/api/ai/connect-info');
const metaData = await metaRes.json().catch(() => ({}));
if (!metaRes.ok || metaData?.ok === false) {
throw new Error(metaData?.detail || 'Could not prepare agent credentials.');
}
setAgentTier(metaData.access_tier === 'full' ? 'full' : 'restricted');
// 2) Mint the secret if it isn't set yet — transparent, idempotent.
let secretSet = !!metaData.hmac_secret_set;
if (!secretSet) {
const bootRes = await fetch('/api/ai/connect-info/bootstrap', {
method: 'POST',
});
const bootData = await bootRes.json().catch(() => ({}));
if (!bootRes.ok || bootData?.ok === false) {
throw new Error(bootData?.detail || 'Could not generate agent credentials.');
}
secretSet = !!bootData.hmac_secret_set;
}
// 3) If the caller asked to see the secret, fetch it explicitly.
// Otherwise the masked fingerprint is enough for the UI.
if (reveal && secretSet) {
const revealRes = await fetch('/api/ai/connect-info/reveal', {
method: 'POST',
});
const revealData = await revealRes.json().catch(() => ({}));
if (!revealRes.ok || revealData?.ok === false) {
throw new Error(revealData?.detail || 'Could not reveal agent credentials.');
}
setAgentSecret(revealData.hmac_secret || '');
} else {
setAgentSecret(metaData.masked_hmac_secret || '');
}
setAgentSecret(data.hmac_secret || '');
setAgentTier(data.access_tier === 'full' ? 'full' : 'restricted');
setAgentMsg({ type: 'ok', text: 'Agent key is ready. Copy it into your local or remote agent runtime.' });
} catch (error) {
setAgentMsg({
+109 -140
View File
@@ -74,17 +74,18 @@ import {
Trash2,
RotateCcw,
Satellite,
Eye,
EyeOff,
Copy,
Check,
Radar,
} from 'lucide-react';
import {
clearSentinelCredentials,
getSentinelCredentialStorageMode,
getSentinelCredentials,
setSentinelCredentials,
// Issue #298: Sentinel credentials now live server-side. The legacy
// browser-storage helpers (getSentinelCredentials / setSentinelCredentials
// / clearSentinelCredentials / getSentinelCredentialStorageMode) have
// been removed from sentinelHub.ts. We use the new status check + the
// one-time migration helper instead.
checkBackendSentinelStatus,
migrateLegacySentinelBrowserKeys,
} from '@/lib/sentinelHub';
import {
getPrivacyProfilePreference,
@@ -143,10 +144,14 @@ const WEIGHT_COLORS: Record<number, string> = {
const SETTINGS_FOCUS_KEY = 'sb_settings_focus';
const WORMHOLE_RETURN_KEY = 'sb_wormhole_return_target';
const WORMHOLE_READY_EVENT = 'sb:wormhole-ready';
// Issue #298 (tg12): Sentinel credentials moved from browser storage to
// the backend ``.env`` (managed through the API Keys panel). The legacy
// keys (``sb_sentinel_client_id`` / ``sb_sentinel_client_secret`` /
// ``sb_sentinel_instance_id``) are no longer treated as sensitive
// browser state because they are no longer written. ``SentinelTab``
// runs ``migrateLegacySentinelBrowserKeys()`` once on mount to clear
// any leftover values from pre-#298 installs.
const PRIVACY_SENSITIVE_BROWSER_KEYS = [
'sb_sentinel_client_id',
'sb_sentinel_client_secret',
'sb_sentinel_instance_id',
'sb_infonet_head',
'sb_infonet_head_history',
'sb_infonet_peers',
@@ -2615,7 +2620,9 @@ const SettingsPanel = React.memo(function SettingsPanel({
)}
{/* ==================== SENTINEL HUB TAB ==================== */}
{activeTab === 'sentinel' && <SentinelTab />}
{activeTab === 'sentinel' && (
<SentinelTab onGoToApiKeys={() => setActiveTab('api-keys')} />
)}
{activeTab === 'sar' && <SarSettingsTab />}
</motion.div>
</>
@@ -2625,63 +2632,58 @@ const SettingsPanel = React.memo(function SettingsPanel({
});
// ─── Sentinel Hub Settings Tab ─────────────────────────────────────────────
function SentinelTab() {
const [clientId, setClientId] = useState(() => getSentinelCredentials().clientId);
const [clientSecret, setClientSecret] = useState(() => getSentinelCredentials().clientSecret);
const [testing, setTesting] = useState(false);
const [status, setStatus] = useState<{ ok: boolean; msg: string } | null>(null);
const [dirty, setDirty] = useState(false);
const [showSecret, setShowSecret] = useState(false);
const storageMode = getSentinelCredentialStorageMode();
// Issue #298 (tg12): Sentinel credentials now live in the backend ``.env``
// and are managed through the existing API Keys panel — same flow as every
// other third-party API key (OpenSky, AIS Stream, Finnhub, …). This tab no
// longer collects credentials. It does three things:
// 1. Runs migrateLegacySentinelBrowserKeys() once to wipe pre-#298
// values out of localStorage / sessionStorage.
// 2. Shows the operator whether the backend has the credentials.
// 3. Offers a one-click jump to the API Keys panel where they enter them.
function SentinelTab({ onGoToApiKeys }: { onGoToApiKeys: () => void }) {
const [backendConfigured, setBackendConfigured] = useState<boolean | null>(null);
const [migrationResult, setMigrationResult] = useState<{ cleared: string[] } | null>(null);
const [refreshing, setRefreshing] = useState(false);
const save = () => {
setSentinelCredentials(clientId.trim(), clientSecret.trim());
setDirty(false);
setStatus({
ok: true,
msg: `Credentials saved to browser ${storageMode === 'session' ? 'session' : 'local'} storage.`,
});
};
useEffect(() => {
// One-time legacy browser-key wipe. Idempotent — does nothing on a
// fresh install. We do NOT silently POST any browser-stored values
// to the backend; operators who relied on them re-enter once in the
// API Keys panel. Doing the wipe regardless ensures pre-#298 secrets
// don't linger in localStorage indefinitely.
setMigrationResult(migrateLegacySentinelBrowserKeys());
const testConnection = async () => {
setTesting(true);
setStatus(null);
// Check whether the backend has SENTINEL_CLIENT_ID/SECRET set.
void checkBackendSentinelStatus().then(setBackendConfigured);
}, []);
const refresh = async () => {
setRefreshing(true);
try {
const resp = await fetch(`${API_BASE}/api/sentinel/token`, {
method: 'POST',
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
body: new URLSearchParams({
client_id: clientId.trim(),
client_secret: clientSecret.trim(),
}),
});
if (resp.ok) {
setStatus({ ok: true, msg: 'Connected — token acquired successfully.' });
} else {
const text = await resp.text().catch(() => '');
setStatus({ ok: false, msg: `Auth failed (${resp.status}): ${text.slice(0, 120)}` });
}
} catch (err) {
const msg =
typeof err === 'object' && err !== null && 'message' in err
? String((err as { message?: string }).message)
: 'unknown';
setStatus({ ok: false, msg: `Network error: ${msg}` });
// refreshSentinelStatus() invalidates the module-level cache so the
// next check actually hits the backend instead of returning the
// memoized value. Lazy-imported so SSR/tests don't choke.
const { refreshSentinelStatus } = await import('@/lib/sentinelHub');
refreshSentinelStatus();
const ok = await checkBackendSentinelStatus();
setBackendConfigured(ok);
} finally {
setTesting(false);
setRefreshing(false);
}
};
const clear = () => {
clearSentinelCredentials();
setClientId('');
setClientSecret('');
setDirty(false);
setStatus({ ok: true, msg: 'Credentials cleared.' });
};
const inputCls =
'w-full bg-[var(--bg-primary)]/60 border border-[var(--border-primary)] px-3 py-2 text-[11px] font-mono text-[var(--text-secondary)] outline-none focus:border-purple-500 placeholder:text-[var(--text-muted)]/50 transition-colors';
const statusColor =
backendConfigured === null
? 'text-[var(--text-muted)]'
: backendConfigured
? 'text-green-400'
: 'text-yellow-400';
const statusLabel =
backendConfigured === null
? 'CHECKING…'
: backendConfigured
? 'CONFIGURED ON BACKEND'
: 'NOT CONFIGURED';
return (
<div className="flex-1 flex flex-col overflow-y-auto styled-scrollbar">
@@ -2733,106 +2735,73 @@ function SentinelTab() {
</p>
<p>
<span className="text-purple-400 font-bold">STEP 3:</span>{' '}
Paste both values in the fields below, hit{' '}
<span className="text-cyan-400">SAVE</span>, then{' '}
<span className="text-cyan-400">TEST CONNECTION</span> to verify.
That&apos;s it!
Paste both values into the <span className="text-cyan-400">API Keys</span> panel
under <span className="text-white">SENTINEL_CLIENT_ID</span> and{' '}
<span className="text-white">SENTINEL_CLIENT_SECRET</span>, then hit Save.
The backend uses them to mint short-lived tokens your browser never sees
the secret again.
</p>
</div>
</div>
</div>
</div>
{/* Credential Inputs */}
<div className="p-4 space-y-3">
<div>
<label className="text-[13px] font-mono text-[var(--text-muted)] tracking-widest mb-1 block">
CLIENT ID
</label>
<input
type="text"
value={clientId}
onChange={(e) => {
setClientId(e.target.value);
setDirty(true);
}}
placeholder="sh-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
spellCheck={false}
autoComplete="off"
className={inputCls}
/>
{/* Backend status */}
<div className="mx-4 mt-3 p-3 border border-[var(--border-primary)] bg-[var(--bg-primary)]/30">
<div className="flex items-center justify-between mb-2">
<span className="text-[13px] font-mono text-[var(--text-muted)] tracking-widest">
BACKEND STATUS
</span>
<span className={`text-[11px] font-mono font-bold ${statusColor}`}>
{statusLabel}
</span>
</div>
<div>
<label className="text-[13px] font-mono text-[var(--text-muted)] tracking-widest mb-1 block">
CLIENT SECRET
</label>
<input
type={showSecret ? 'text' : 'password'}
value={clientSecret}
onChange={(e) => {
setClientSecret(e.target.value);
setDirty(true);
}}
placeholder="Paste client secret here..."
spellCheck={false}
autoComplete="new-password"
className={inputCls}
/>
<p className="text-[13px] text-[var(--text-muted)] font-mono leading-relaxed">
{backendConfigured === false
? 'Sentinel credentials are not yet set in the backend .env. Open the API Keys panel to enter them — the tile overlay and Sentinel-2 Intel Card will work as soon as both fields are saved.'
: backendConfigured === true
? 'Sentinel credentials are configured on the backend. The dashboard fetches tokens automatically; your browser does not handle the secret.'
: 'Checking backend configuration…'}
</p>
<div className="mt-3 flex items-center gap-2">
<button
type="button"
onClick={() => setShowSecret((current) => !current)}
className="mt-2 inline-flex items-center gap-1.5 text-[13px] font-mono text-[var(--text-muted)] hover:text-[var(--text-secondary)] transition-colors"
onClick={onGoToApiKeys}
className="flex-1 px-4 py-2 bg-purple-500/20 border border-purple-500/40 text-purple-400 hover:bg-purple-500/30 transition-colors text-sm font-mono flex items-center justify-center gap-1.5"
>
{showSecret ? <EyeOff size={10} /> : <Eye size={10} />}
{showSecret ? 'HIDE SECRET' : 'SHOW SECRET'}
OPEN API KEYS PANEL
</button>
<button
onClick={refresh}
disabled={refreshing}
className="px-3 py-2 border border-[var(--border-primary)] text-[var(--text-muted)] hover:text-cyan-400 hover:border-cyan-500/50 transition-all text-sm font-mono disabled:opacity-40"
title="Re-check backend status"
>
{refreshing ? 'CHECKING…' : 'REFRESH'}
</button>
</div>
</div>
{/* Status */}
{status && (
<div
className={`mx-4 mb-2 px-3 py-2 text-sm font-mono ${status.ok ? 'text-green-400 bg-green-950/20 border border-green-900/30' : 'text-red-400 bg-red-950/20 border border-red-900/30'}`}
>
{status.msg}
{/* Migration notice (only if we actually cleared anything) */}
{migrationResult && migrationResult.cleared.length > 0 && (
<div className="mx-4 mt-3 px-3 py-2 text-sm font-mono text-cyan-400 bg-cyan-950/20 border border-cyan-900/30">
<p className="font-bold mb-1">LEGACY BROWSER CREDENTIALS CLEARED</p>
<p className="text-[13px] leading-relaxed text-[var(--text-muted)]">
Found and removed pre-#298 Sentinel credentials from browser storage
({migrationResult.cleared.join(', ')}). Re-enter them in the API Keys panel
above; they&apos;ll be stored server-side from now on and never sent back to
the browser.
</p>
</div>
)}
{/* Actions */}
{/* Footer + Usage Meter */}
<div className="p-4 border-t border-[var(--border-primary)]/80 mt-auto">
<div className="flex items-center gap-2">
<button
onClick={save}
disabled={!dirty}
className="flex-1 px-4 py-2 bg-purple-500/20 border border-purple-500/40 text-purple-400 hover:bg-purple-500/30 transition-colors text-sm font-mono flex items-center justify-center gap-1.5 disabled:opacity-30 disabled:cursor-not-allowed"
>
<Save size={10} />
SAVE
</button>
<button
onClick={testConnection}
disabled={testing || !clientId || !clientSecret}
className="flex-1 px-4 py-2 bg-cyan-500/20 border border-cyan-500/40 text-cyan-400 hover:bg-cyan-500/30 transition-colors text-sm font-mono flex items-center justify-center gap-1.5 disabled:opacity-30 disabled:cursor-not-allowed"
>
{testing ? 'TESTING...' : 'TEST CONNECTION'}
</button>
<button
onClick={clear}
className="px-3 py-2 border border-[var(--border-primary)] text-[var(--text-muted)] hover:text-red-400 hover:border-red-500/50 hover:bg-red-950/10 transition-all text-sm font-mono flex items-center gap-1.5"
title="Clear credentials"
>
<Trash2 size={10} />
</button>
</div>
{/* Usage Meter */}
<UsageMeter />
<div className="mt-2 p-2 border border-[var(--border-primary)]/40 bg-[var(--bg-primary)]/30">
<p className="text-[13px] text-[var(--text-muted)] font-mono leading-relaxed">
Credentials stay in browser-only storage and never touch ShadowBroker servers.
{storageMode === 'session'
? ' Current privacy mode keeps them in session storage only.'
: ' Current privacy mode keeps them in local storage for persistence.'}
Credentials are stored in the backend <span className="text-cyan-400">.env</span>{' '}
and never sent to the browser. The tile proxy mints short-lived OAuth tokens
on demand using those values.
</p>
</div>
</div>
@@ -0,0 +1,85 @@
/**
* useAisUpstreamHealth polls /api/health and exposes AIS proxy connectivity.
*
* Background: AISStream's WebSocket server went fully offline 2026-05-23 (TCP
* timeouts at stream.aisstream.io). The backend kept reconnecting in a tight
* loop and the ships layer silently went empty. Users had no signal that the
* problem was upstream, not their config. This hook surfaces the state so a
* banner can explain "AIS upstream is offline" instead of letting users
* wonder.
*
* The poll interval is intentionally relaxed (30s) this is a low-urgency UX
* signal, not a real-time data feed. Backend already escalates top_status to
* "degraded" when AIS is configured-but-disconnected.
*/
import { useEffect, useRef, useState } from 'react';
import { API_BASE } from '@/lib/api';
export interface AisUpstreamHealth {
/** True when we've received a vessel message in the last ~60s. */
connected: boolean;
/** Seconds since the last vessel message; null when we've never seen one. */
lastMsgAgeSeconds: number | null;
/**
* True when the SPKI-pinned fallback is in effect (issue #258).
* Data still flows in this mode it's a separate, less urgent signal
* than ``connected``.
*/
degradedTls: boolean;
/** How many times the proxy has been spawned (sustained growth without
* ``connected`` means upstream is dead and we're respawning in a loop). */
proxySpawnCount: number;
/** Whether the operator has configured an API key. When false, the banner
* shouldn't fire because "AIS is off" is the intended state. The backend
* signals this via the ``connected`` flag being false AND no msg ever
* seen we approximate it by requiring at least one spawn before
* declaring an outage. */
aisEnabled: boolean;
}
const POLL_INTERVAL_MS = 30_000;
export function useAisUpstreamHealth(): AisUpstreamHealth | null {
const [health, setHealth] = useState<AisUpstreamHealth | null>(null);
const cancelledRef = useRef(false);
useEffect(() => {
cancelledRef.current = false;
const fetchHealth = async () => {
try {
const res = await fetch(`${API_BASE}/api/health`, { cache: 'no-store' });
if (!res.ok) return;
const body = await res.json();
if (cancelledRef.current) return;
const proxy = body?.ais_proxy ?? {};
// ``proxy_spawn_count > 0`` is the cheapest "AIS is enabled" check:
// if the backend never spawned the proxy (no API key, opt-out env)
// we shouldn't ever show the outage banner. Once the proxy has
// spawned at least once we know the operator wants AIS data.
const spawns = Number(proxy.proxy_spawn_count ?? 0);
setHealth({
connected: Boolean(proxy.connected),
lastMsgAgeSeconds:
proxy.last_msg_age_seconds == null
? null
: Number(proxy.last_msg_age_seconds),
degradedTls: Boolean(proxy.degraded_tls),
proxySpawnCount: spawns,
aisEnabled: spawns > 0,
});
} catch {
// Backend unreachable — separate problem. Banner not relevant.
}
};
void fetchHealth();
const interval = setInterval(() => void fetchHealth(), POLL_INTERVAL_MS);
return () => {
cancelledRef.current = true;
clearInterval(interval);
};
}, []);
return health;
}
+119 -68
View File
@@ -1,77 +1,137 @@
/**
* Sentinel Hub (Copernicus CDSE) client-side token management & Process API tile fetcher.
* Sentinel Hub (Copernicus CDSE) client-side token + Process API tile fetcher.
*
* Credentials are stored in browser-controlled storage only. In privacy/session
* mode they stay session-scoped; otherwise they persist in local storage. Token
* exchange is proxied through the ShadowBroker backend (/api/sentinel/token) to
* avoid CORS blocks from the Copernicus identity provider. Credentials are
* forwarded, never stored server-side.
* Issue #298 (tg12): Credentials are now stored server-side in the backend
* ``.env`` (managed through the existing ``/api/settings/api-keys`` flow,
* same as every other third-party API key). The browser no longer holds
* ``client_id`` / ``client_secret`` in localStorage or sessionStorage and
* no longer forwards them in proxy requests.
*
* Uses the Process API with inline evalscripts no Instance ID / Configuration needed.
* Old browser-storage keys (``sb_sentinel_client_id`` / ``sb_sentinel_client_secret``
* / ``sb_sentinel_instance_id``) are migrated out by ``SettingsPanel`` on
* first mount after the upgrade see ``migrateLegacySentinelBrowserKeys()``
* exported below.
*/
import { API_BASE } from '@/lib/api';
import {
getSensitiveBrowserItem,
getSensitiveBrowserStorageMode,
removeSensitiveBrowserItem,
setSensitiveBrowserItem,
} from '@/lib/privacyBrowserStorage';
// Token exchange proxied through our backend (Copernicus blocks browser CORS)
// Token exchange proxied through our backend (Copernicus blocks browser CORS).
const TOKEN_PROXY_URL = `${API_BASE}/api/sentinel/token`;
// browser-storage keys
const LS_CLIENT_ID = 'sb_sentinel_client_id';
const LS_CLIENT_SECRET = 'sb_sentinel_client_secret';
// In-memory token cache (never persisted)
let cachedToken: string | null = null;
let tokenExpiry = 0;
// Dedup: only one in-flight token request at a time
let _tokenPromise: Promise<string | null> | null = null;
// ─── Credential helpers ────────────────────────────────────────────────────
// In-memory cache of "does the backend have Sentinel credentials configured?"
// so the rest of the UI can short-circuit tile load attempts without a server
// round-trip per tile. Refreshed by callers via `refreshSentinelStatus()`.
let _backendCredentialsConfigured: boolean | null = null;
let _backendStatusPromise: Promise<boolean> | null = null;
export function getSentinelCredentials(): {
clientId: string;
clientSecret: string;
} {
if (typeof window === 'undefined') return { clientId: '', clientSecret: '' };
return {
clientId: getSensitiveBrowserItem(LS_CLIENT_ID) || '',
clientSecret: getSensitiveBrowserItem(LS_CLIENT_SECRET) || '',
};
// ─── Credential status (server-side) ───────────────────────────────────────
/**
* Ask the backend whether Sentinel credentials are configured in ``.env``.
* Caches the result in memory; call ``refreshSentinelStatus()`` after the
* operator saves new API keys in the settings panel.
*
* Returns ``false`` on network errors so the UI fails safely (no broken
* tile requests). Never returns the secret itself that stays server-side.
*/
export async function checkBackendSentinelStatus(): Promise<boolean> {
if (_backendCredentialsConfigured !== null) return _backendCredentialsConfigured;
if (_backendStatusPromise) return _backendStatusPromise;
_backendStatusPromise = (async () => {
try {
const resp = await fetch(`${API_BASE}/api/settings/api-keys`, {
headers: { Accept: 'application/json' },
});
if (!resp.ok) return false;
const list = await resp.json();
// /api/settings/api-keys returns an array of { id, env_key, is_set, ... }
const ids = new Set(['sentinel_client_id', 'sentinel_client_secret']);
const configured = Array.isArray(list)
&& list.filter((row: { id?: string; is_set?: boolean }) =>
row && row.id && ids.has(row.id) && row.is_set === true,
).length === 2;
_backendCredentialsConfigured = configured;
return configured;
} catch {
_backendCredentialsConfigured = false;
return false;
} finally {
_backendStatusPromise = null;
}
})();
return _backendStatusPromise;
}
export function setSentinelCredentials(clientId: string, clientSecret: string): void {
setSensitiveBrowserItem(LS_CLIENT_ID, clientId);
setSensitiveBrowserItem(LS_CLIENT_SECRET, clientSecret);
// Invalidate cached token when credentials change
/** Invalidate the cached status — call this after the API Keys panel saves. */
export function refreshSentinelStatus(): void {
_backendCredentialsConfigured = null;
// Drop any cached token too — credentials may have changed.
cachedToken = null;
tokenExpiry = 0;
}
export function clearSentinelCredentials(): void {
removeSensitiveBrowserItem(LS_CLIENT_ID);
removeSensitiveBrowserItem(LS_CLIENT_SECRET);
// Also remove legacy instance ID if present
removeSensitiveBrowserItem('sb_sentinel_instance_id');
if (typeof window !== 'undefined') {
localStorage.removeItem('sb_sentinel_instance_id');
sessionStorage.removeItem('sb_sentinel_instance_id');
}
cachedToken = null;
tokenExpiry = 0;
}
export function getSentinelCredentialStorageMode(): 'local' | 'session' {
return getSensitiveBrowserStorageMode();
/**
* Synchronous getter returns the last known status without a network call.
* Returns ``null`` until ``checkBackendSentinelStatus()`` has run at least once.
*/
export function getCachedSentinelStatus(): boolean | null {
return _backendCredentialsConfigured;
}
/**
* Back-compat shim. Pre-#298 callers asked ``hasSentinelCredentials()`` to
* decide whether to render the Sentinel layer / open the API key prompt.
* The credential now lives server-side, so this is just the cached
* server-status check. Returns ``false`` until the first
* ``checkBackendSentinelStatus()`` resolves (callers should kick that off
* once at app startup see ``page.tsx`` mount effect).
*/
export function hasSentinelCredentials(): boolean {
const { clientId, clientSecret } = getSentinelCredentials();
return Boolean(clientId && clientSecret);
return _backendCredentialsConfigured === true;
}
/**
* One-time migration helper: clear the legacy browser-storage keys that
* pre-#298 versions used to persist Sentinel credentials. Idempotent and
* safe to call on every page load does nothing if no keys are present.
*
* Called by ``SettingsPanel`` on mount. We do NOT auto-POST the legacy
* browser values to the backend, because doing so would silently migrate
* a secret across a trust boundary without operator consent. Operators
* who relied on browser-stored credentials will re-enter them once in
* the API Keys panel, and the legacy keys get wiped here.
*/
export function migrateLegacySentinelBrowserKeys(): { cleared: string[] } {
if (typeof window === 'undefined') return { cleared: [] };
const legacy = [
'sb_sentinel_client_id',
'sb_sentinel_client_secret',
'sb_sentinel_instance_id',
];
const cleared: string[] = [];
for (const key of legacy) {
try {
if (window.localStorage?.getItem(key) !== null) {
window.localStorage.removeItem(key);
cleared.push(key);
}
} catch { /* ignore quota / privacy mode errors */ }
try {
if (window.sessionStorage?.getItem(key) !== null) {
window.sessionStorage.removeItem(key);
if (!cleared.includes(key)) cleared.push(key);
}
} catch { /* ignore */ }
}
return { cleared };
}
// ─── OAuth2 token ──────────────────────────────────────────────────────────
@@ -79,14 +139,16 @@ export function hasSentinelCredentials(): boolean {
/**
* Fetch an OAuth2 access token using the client_credentials grant.
* Caches in memory; auto-refreshes 30 s before expiry.
*
* The request body NO LONGER carries client_id/secret the backend
* resolves credentials from its ``.env`` via the API Keys flow. The
* server-side proxy still accepts body credentials for legacy callers,
* but the dashboard does not supply them.
*/
export function getSentinelToken(): Promise<string | null> {
// Return cached token if still valid (with 30 s margin)
if (cachedToken && Date.now() < tokenExpiry - 30_000) return Promise.resolve(cachedToken);
const { clientId, clientSecret } = getSentinelCredentials();
if (!clientId || !clientSecret) return Promise.resolve(null);
// Dedup: reuse in-flight request so 20 tiles don't each trigger a token fetch
if (_tokenPromise) return _tokenPromise;
@@ -94,11 +156,9 @@ export function getSentinelToken(): Promise<string | null> {
try {
const resp = await fetch(TOKEN_PROXY_URL, {
method: 'POST',
// Backend resolves credentials from env. Empty body = "use server-side".
headers: { 'Content-Type': 'application/x-www-form-urlencoded' },
body: new URLSearchParams({
client_id: clientId,
client_secret: clientSecret,
}),
body: new URLSearchParams({}),
});
if (!resp.ok) {
@@ -131,6 +191,8 @@ const TILE_PROXY_URL = `${API_BASE}/api/sentinel/tile`;
/**
* Fetch a single 256×256 tile via backend proxy to Sentinel Hub Process API.
* Returns a PNG ArrayBuffer or null on failure.
*
* Body no longer carries client_id/secret the backend uses .env values.
*/
export async function fetchSentinelTile(
z: number,
@@ -139,21 +201,10 @@ export async function fetchSentinelTile(
preset: string,
date: string,
): Promise<ArrayBuffer | null> {
const { clientId, clientSecret } = getSentinelCredentials();
if (!clientId || !clientSecret) return null;
const resp = await fetch(TILE_PROXY_URL, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
client_id: clientId,
client_secret: clientSecret,
preset,
date,
z,
x,
y,
}),
body: JSON.stringify({ preset, date, z, x, y }),
});
if (!resp.ok) return null;