Fix #244/#245/#246: carrier tracker seed/cache/freshness model

Replace the dated editorial fallback positions baked into the registry
with a one-shot seed file + persistent observation cache. The user's
runtime cache now reflects what THIS install has actually observed,
not what USNI published on March 9, 2026. A year from now, the cache
holds a year of observations and the seed is irrelevant.

== #244: dated editorial coordinates out of the registry ==

CARRIER_REGISTRY no longer carries fallback_lat/lng/heading/desc.
Those fields are deleted. The registry is now identity + homeport
only.

New file: backend/data/carrier_seed.json
  - Read-only, shipped with every release.
  - Used ONCE on first-ever startup to bootstrap carrier_cache.json.
  - Each entry stamped with position_confidence="seed" and the actual
    as-of date (2026-03-09), NOT now().

== #245: approximate confidence for headline-derived positions ==

_parse_carrier_positions_from_news() now stamps every GDELT-derived
entry with position_confidence="approximate" so the UI knows the
coordinate is a region-centroid match, not a precise observation.
After the freshness window the label rolls over to
"stale_approximate" so old-and-imprecise is distinguishable from
recent-and-imprecise.

The article's actual seendate is used as position_source_at instead
of now(), so the "last reported X days ago" badge is honest.

== #246: freshness is labelling, not eviction ==

The cache always preserves the last position the system observed,
forever. What changes is the position_confidence label:
  - within configurable window (default 14d, env-overridable via
    SHADOWBROKER_CARRIER_FRESHNESS_DAYS) -> "recent"
  - older -> "stale"
  - seed-bootstrap entries that were never refreshed -> "seed"
  - homeport defaults (carrier added post-install) -> "homeport_default"
  - headline-derived (any age, fresh) -> "approximate"
  - headline-derived (older than window) -> "stale_approximate"

The position itself never reverts to the seed or the registry. The
user always sees the last position the system observed. Per the
user's explicit guidance: "from there have it be the last position
the user has logged the carriers that way a year from now it doesnt
revert to where the ships are today".

== Other improvements ==

- CACHE_FILE moved to backend/data/carrier_cache.json so it lives in
  the volume-mounted dir under Docker compose. Previously it was at
  /app/carrier_cache.json which got wiped on every container restart
  (pre-existing bug).
- Atomic cache write (temp + os.replace) so a crash mid-write does
  not leave a truncated cache file.

== Public API shape ==

Every carrier object the API emits now includes:
  - position_confidence: seed | recent | stale | approximate |
                         stale_approximate | homeport_default
  - position_source_at:  ISO timestamp of when the underlying source
                         was observed (NOT now())
  - is_fallback:         convenience boolean for the UI; true when the
                         confidence is seed/stale/stale_approximate/
                         homeport_default

Existing fields (estimated, source, source_url, last_osint_update,
name, type, lat, lng, country, desc, wiki) are preserved exactly so
the current ShipPopup frontend renders unchanged. last_osint_update
now reflects position_source_at instead of now(), which is what the
existing "last reported MM/DD" badge always meant to show.

Tests: backend/tests/test_carrier_tracker_quality.py — 17 tests
covering seed bootstrap, subsequent-startup ignoring seed, no-seed/
no-cache homeport fallback, registry no longer has fallback fields,
freshness window labelling + env override, "year-old cache entry keeps
its position, only the label flips" regression, approximate
confidence for headline matches, GDELT seendate ISO parser, public
response shape backward compat.

Credit: tg12 (external security audit, three P1/P2 issues).

.gitignore

@@ -101,6 +101,14 @@ backend/data/*
 # Issue #258: SPKI pins for stream.aisstream.io so we can survive upstream
 # Let's Encrypt renewal failures without disabling TLS validation entirely.
 !backend/data/aisstream_spki_pins.json
+# Issue #231: pinned SHA-256 digests for known release archives. Used by
+# the self-updater as a second-line integrity check when the release's
+# SHA256SUMS.txt asset can't be fetched.
+!backend/data/release_digests.json
+# Issue #244/#245/#246: one-shot carrier-position seed shipped with each
+# release. Used ONLY on first-ever startup to bootstrap carrier_cache.json;
+# after that the cache reflects this install's own GDELT observations.
+!backend/data/carrier_seed.json
 
 # OS generated files
 .DS_Store

backend/auth.py

@@ -45,6 +45,7 @@ from services.mesh.mesh_compatibility import (
 from services.mesh.mesh_crypto import (
     _derive_peer_key,
     normalize_peer_url,
+    resolve_peer_key_for_url,
     verify_signature,
     verify_node_binding,
     parse_public_key_algo,
@@ -245,15 +246,90 @@ def _docker_bridge_local_operator_enabled() -> bool:
     }
 
 
+# Issue #250 (tg12): the previous implementation returned True for any IP
+# in the entire 172.16.0.0/12 range. Anyone with `docker run` access on
+# the same daemon could spin up a container that automatically passed
+# local-operator auth. The fix narrows trust to ONLY connections whose
+# source IP matches the configured frontend container's hostname.
+#
+# Docker DNS resolves both the compose service name (``frontend``) and
+# the explicit ``container_name`` (``shadowbroker-frontend``) to the
+# frontend container's bridge IP. We forward-resolve both, cache the
+# result for 30s, and only trust connections from those exact IPs.
+#
+# Operators on shared Docker hosts get the benefit of the narrower
+# surface. Operators on single-user installs see no behavior change —
+# their frontend container still resolves and is still trusted.
+_DOCKER_BRIDGE_TRUST_CACHE: dict = {"ips": frozenset(), "expires": 0.0}
+_DOCKER_BRIDGE_TRUST_TTL = 30.0
+
+
+def _trusted_bridge_frontend_hostnames() -> list[str]:
+    """Container hostnames whose IPs we treat as local-operator on the bridge.
+
+    Default covers both Docker Compose service name (``frontend``) and the
+    explicit ``container_name`` from the shipped docker-compose.yml
+    (``shadowbroker-frontend``). Operators with non-default names can
+    override via the ``SHADOWBROKER_TRUSTED_FRONTEND_HOSTS`` env var
+    (comma-separated, no spaces).
+    """
+    raw = str(
+        os.environ.get(
+            "SHADOWBROKER_TRUSTED_FRONTEND_HOSTS",
+            "frontend,shadowbroker-frontend",
+        )
+    ).strip()
+    return [h.strip() for h in raw.split(",") if h.strip()]
+
+
+def _resolve_trusted_bridge_ips() -> frozenset[str]:
+    """Resolve trusted frontend hostnames to a set of IPs, with caching.
+
+    Cached for 30s so we don't hit DNS on every request. The cache is
+    process-local — frontend container IP rotations during a backend's
+    lifetime will be picked up within 30s.
+
+    Returns frozenset() if Docker DNS can't resolve any of the configured
+    hostnames (fail-closed — when in doubt, refuse to trust the bridge).
+    """
+    import socket
+    import time as _time
+
+    now = _time.time()
+    cache = _DOCKER_BRIDGE_TRUST_CACHE
+    if cache["expires"] > now:
+        return cache["ips"]
+
+    ips: set[str] = set()
+    for hostname in _trusted_bridge_frontend_hostnames():
+        try:
+            _, _, addrs = socket.gethostbyname_ex(hostname)
+        except (OSError, socket.gaierror):
+            continue
+        for addr in addrs:
+            ips.add(addr)
+
+    resolved = frozenset(ips)
+    cache["ips"] = resolved
+    cache["expires"] = now + _DOCKER_BRIDGE_TRUST_TTL
+    return resolved
+
+
 def _is_docker_bridge_host(host: str) -> bool:
+    """Return True only when the source IP matches our trusted frontend
+    container hostname(s).
+
+    Previously trusted any 172.16.0.0/12 IP unconditionally. See the
+    block comment above for the security rationale.
+    """
     try:
         ip = ipaddress.ip_address(host)
     except ValueError:
         return False
-    # Docker Desktop and the default compose bridge normally sit inside
-    # 172.16.0.0/12. Keep this narrower than "any private IP" so a user who
-    # intentionally binds the backend to LAN does not silently trust LAN clients.
-    return ip in ipaddress.ip_network("172.16.0.0/12")
+    # Public IPs are never our frontend container — skip DNS work for them.
+    if not ip.is_private:
+        return False
+    return host in _resolve_trusted_bridge_ips()
 
 
 def _is_trusted_local_runtime_host(host: str) -> bool:
@@ -1328,11 +1404,15 @@ def _peer_hmac_url_from_request(request: Request) -> str:
 
 
 def _verify_peer_push_hmac(request: Request, body_bytes: bytes) -> bool:
-    """Verify HMAC-SHA256 peer authentication on push requests."""
-    secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
-    if not secret:
-        return False
+    """Verify HMAC-SHA256 peer authentication on push requests.
 
+    Issue #256: ``resolve_peer_key_for_url`` looks up a per-peer secret
+    in ``MESH_PEER_SECRETS`` first, then falls back to the global
+    ``MESH_PEER_PUSH_SECRET``. When a peer URL is listed in the per-peer
+    map, only the listed secret is accepted for it — the global secret
+    is ignored, so any peer that knows only the global secret cannot
+    forge a request claiming to be that peer.
+    """
     provided = str(request.headers.get("x-peer-hmac", "") or "").strip()
     if not provided:
         return False
@@ -1341,7 +1421,7 @@ def _verify_peer_push_hmac(request: Request, body_bytes: bytes) -> bool:
     allowed_peers = set(authenticated_push_peer_urls())
     if not peer_url or peer_url not in allowed_peers:
         return False
-    peer_key = _derive_peer_key(secret, peer_url)
+    peer_key = resolve_peer_key_for_url(peer_url)
     if not peer_key:
         return False
 

backend/data/carrier_seed.json

@@ -0,0 +1,120 @@
+{
+  "_meta": {
+    "as_of": "2026-03-09",
+    "source": "USNI News Fleet & Marine Tracker",
+    "source_url": "https://news.usni.org/2026/03/09/usni-news-fleet-and-marine-tracker-march-9-2026",
+    "note": "One-shot bootstrap for first-run carrier positions. Once carrier_cache.json exists in the runtime data volume, this seed file is never read again. All subsequent updates come from GDELT (and any future sources) and are written to carrier_cache.json. A year from now, your runtime cache reflects whatever your install has observed since first launch — not these snapshot positions."
+  },
+  "carriers": {
+    "CVN-68": {
+      "lat": 47.5535,
+      "lng": -122.6400,
+      "heading": 90,
+      "desc": "Bremerton, WA (Maintenance)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-76": {
+      "lat": 47.5580,
+      "lng": -122.6360,
+      "heading": 90,
+      "desc": "Bremerton, WA (Decommissioning)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-69": {
+      "lat": 36.9465,
+      "lng": -76.3265,
+      "heading": 0,
+      "desc": "Norfolk, VA (Post-deployment maintenance)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-78": {
+      "lat": 18.0,
+      "lng": 39.5,
+      "heading": 0,
+      "desc": "Red Sea — Operation Epic Fury (USNI Mar 9)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-74": {
+      "lat": 36.98,
+      "lng": -76.43,
+      "heading": 0,
+      "desc": "Newport News, VA (RCOH refueling overhaul)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-75": {
+      "lat": 36.0,
+      "lng": 15.0,
+      "heading": 0,
+      "desc": "Mediterranean Sea deployment (USNI Mar 9)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-77": {
+      "lat": 36.5,
+      "lng": -74.0,
+      "heading": 0,
+      "desc": "Atlantic — Pre-deployment workups (USNI Mar 9)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-70": {
+      "lat": 32.6840,
+      "lng": -117.1290,
+      "heading": 180,
+      "desc": "San Diego, CA (Homeport)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-71": {
+      "lat": 32.6885,
+      "lng": -117.1280,
+      "heading": 180,
+      "desc": "San Diego, CA (Maintenance)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-72": {
+      "lat": 20.0,
+      "lng": 64.0,
+      "heading": 0,
+      "desc": "Arabian Sea — Operation Epic Fury (USNI Mar 9)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    },
+    "CVN-73": {
+      "lat": 35.2830,
+      "lng": 139.6700,
+      "heading": 180,
+      "desc": "Yokosuka, Japan (Forward deployed)",
+      "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+      "source_url": "https://news.usni.org/category/fleet-tracker",
+      "position_source_at": "2026-03-09T00:00:00Z",
+      "position_confidence": "seed"
+    }
+  }
+}

backend/data/release_digests.json

@@ -0,0 +1,40 @@
+{
+  "_comment": [
+    "Baked-in SHA-256 digests for known Shadowbroker release archives.",
+    "",
+    "Issue #231: the self-updater previously skipped integrity verification",
+    "entirely whenever the MESH_UPDATE_SHA256 env var was unset (which is the",
+    "default — nothing in the install docs tells operators to set it). That",
+    "made the auto-update a supply-chain RCE on any compromise of the GitHub",
+    "release pipeline.",
+    "",
+    "The fix uses a multi-source verification chain mirroring the Tor bundle",
+    "digest approach in #201:",
+    "",
+    "  1. MESH_UPDATE_SHA256 env var (operator override, preserved)",
+    "  2. SHA256SUMS.txt asset published alongside each release (primary —",
+    "     the maintainer's release process already publishes this)",
+    "  3. This baked-in digest list (second line of defense for releases",
+    "     missing a SHA256SUMS asset, or when the asset can't be fetched)",
+    "  4. HTTPS-only fallback with a loud warning (preserves auto-update",
+    "     flow during transient outages so users don't get stuck)",
+    "",
+    "Mismatch from a source that DID respond is fatal — the update is",
+    "refused and the existing install keeps running. Only the 'no source",
+    "reachable at all' case falls back to HTTPS-only.",
+    "",
+    "Format: each entry is keyed by release tag and maps asset filenames",
+    "to their canonical SHA-256 digest (hex, lowercase). The updater",
+    "compares the locally-computed digest of the downloaded asset against",
+    "the value here.",
+    "",
+    "When the maintainer ships a new release, add its digests here BEFORE",
+    "removing the old ones so operators on the old code still validate",
+    "against the previous entries during the transition."
+  ],
+  "v0.9.79": {
+    "ShadowBroker_v0.9.79.zip": "f6877c1d66614525315ea82636ce9f7b41178332c4dbf90d27431a1ea1d9cd47",
+    "ShadowBroker_0.9.79_x64-setup.exe": "f7b676ada45cac7da05868b0a353678c9ee700e3abcf456a7c0c038c36da446f",
+    "ShadowBroker_0.9.79_x64_en-US.msi": "e0713c3cdda184cfbea750bfac0d62a35678fec00847e6476f2cac8e7e42046e"
+  }
+}

backend/main.py

@@ -220,6 +220,7 @@ from services.mesh.mesh_crypto import (
     _derive_peer_key,
     derive_node_id,
     normalize_peer_url,
+    resolve_peer_key_for_url,
     verify_node_binding,
     parse_public_key_algo,
 )
@@ -1079,8 +1080,18 @@ def _public_mesh_log_size(entries: list[dict[str, Any]]) -> int:
     return sum(1 for item in entries if _public_mesh_log_entry(item) is not None)
 
 
-_WORMHOLE_PUBLIC_SETTINGS_FIELDS = {"enabled", "transport", "anonymous_mode"}
-_WORMHOLE_PUBLIC_PROFILE_FIELDS = {"profile", "wormhole_enabled"}
+# Issue #243 (tg12): the public redaction now exposes only the bare
+# "is Wormhole on?" boolean. Transport choice (tor/i2p/mixnet/direct),
+# anonymous-mode state, and the named privacy profile are all
+# operational posture and were leaking actionable recon to any
+# unauthenticated caller. They are now gated behind authenticated reads
+# (admin key or scoped-view token). Loopback Tauri shells and Docker
+# bridge frontend containers continue to see full status because the
+# Next.js catch-all proxy injects the configured ADMIN_KEY for
+# same-origin/non-browser callers (see PR #263), so legitimate operator
+# UX is unaffected.
+_WORMHOLE_PUBLIC_SETTINGS_FIELDS = {"enabled"}
+_WORMHOLE_PUBLIC_PROFILE_FIELDS = {"wormhole_enabled"}
 _PRIVATE_LANE_CONTROL_FIELDS = {"private_lane_tier", "private_lane_policy"}
 _PUBLIC_RNS_STATUS_FIELDS = {"enabled", "ready", "configured_peers", "active_peers"}
 _NODE_PUBLIC_EVENT_HOOK_REGISTERED = False
@@ -1745,10 +1756,12 @@ def _http_peer_push_loop() -> None:
                 _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
                 continue
 
-            secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
-            if not secret:
-                _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
-                continue
+            # Issue #256: resolve_peer_key_for_url() handles both the
+            # legacy global MESH_PEER_PUSH_SECRET path and the per-peer
+            # MESH_PEER_SECRETS map. The per-peer skip happens below
+            # ("if not peer_key: continue"), so we don't gate the whole
+            # loop on the global secret being set — an install that only
+            # configures per-peer secrets is now valid.
 
             peers = authenticated_push_peer_urls()
             if not peers:
@@ -1778,7 +1791,7 @@ def _http_peer_push_loop() -> None:
                         ensure_ascii=False,
                     ).encode("utf-8")
 
-                    peer_key = _derive_peer_key(secret, normalized)
+                    peer_key = resolve_peer_key_for_url(normalized)
                     if not peer_key:
                         continue
                     import hmac as _hmac_mod2
@@ -1831,10 +1844,7 @@ def _http_gate_pull_loop() -> None:
                 _NODE_SYNC_STOP.wait(_GATE_PULL_INTERVAL_S)
                 continue
 
-            secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
-            if not secret:
-                _NODE_SYNC_STOP.wait(_GATE_PULL_INTERVAL_S)
-                continue
+            # Issue #256: per-peer key resolution; see _http_peer_push_loop.
 
             peers = authenticated_push_peer_urls()
             if not peers:
@@ -1846,7 +1856,7 @@ def _http_gate_pull_loop() -> None:
                 if not normalized:
                     continue
 
-                peer_key = _derive_peer_key(secret, normalized)
+                peer_key = resolve_peer_key_for_url(normalized)
                 if not peer_key:
                     continue
 
@@ -1959,10 +1969,7 @@ def _http_gate_push_loop() -> None:
                 _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
                 continue
 
-            secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
-            if not secret:
-                _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
-                continue
+            # Issue #256: per-peer key resolution; see _http_peer_push_loop.
 
             peers = authenticated_push_peer_urls()
             if not peers:
@@ -1977,7 +1984,7 @@ def _http_gate_push_loop() -> None:
                 if not normalized:
                     continue
 
-                peer_key = _derive_peer_key(secret, normalized)
+                peer_key = resolve_peer_key_for_url(normalized)
                 if not peer_key:
                     continue
 
@@ -8813,9 +8820,14 @@ async def api_uw_flow(request: Request):
 from services.news_feed_config import get_feeds, save_feeds, reset_feeds
 
 
-@app.get("/api/settings/news-feeds")
+@app.get(
+    "/api/settings/news-feeds",
+    dependencies=[Depends(require_local_operator)],
+)
 @limiter.limit("30/minute")
 async def api_get_news_feeds(request: Request):
+    """Issue #252 (tg12): gated on local-operator. See the canonical
+    handler in backend/routers/admin.py for the full rationale."""
     return get_feeds()
 
 
@@ -9018,9 +9030,22 @@ class NodeSettingsUpdate(BaseModel):
 @app.get("/api/settings/node")
 @limiter.limit("30/minute")
 async def api_get_node_settings(request: Request):
+    """Issue #243 (tg12): node mode and participant state are
+    operational posture. Anonymous callers receive an empty stub —
+    enough for the UI to know the endpoint exists but nothing
+    fingerprintable. Authenticated callers see the full state.
+
+    Authenticated == local-operator (loopback / Docker bridge) OR an
+    admin / scoped-view token. The Tauri shell and Docker frontend
+    container both qualify via their existing transport (PR #263 +
+    PR #278), so legitimate operator UX is unchanged.
+    """
     from services.node_settings import read_node_settings
 
     data = await asyncio.to_thread(read_node_settings)
+    authenticated = _scoped_view_authenticated(request, "node")
+    if not authenticated:
+        return {}
     return {
         **data,
         "node_mode": _current_node_mode(),

backend/routers/admin.py

@@ -82,9 +82,18 @@ async def api_get_keys_meta(request: Request):
     return get_env_path_info()
 
 
-@router.get("/api/settings/news-feeds")
+@router.get(
+    "/api/settings/news-feeds",
+    dependencies=[Depends(require_local_operator)],
+)
 @limiter.limit("30/minute")
 async def api_get_news_feeds(request: Request):
+    """Issue #252 (tg12): the curated feed inventory is configuration
+    state, not a public data feed. Gated on local-operator so the
+    Tauri shell, the Docker bridge frontend, and any caller with an
+    admin key all see the full list; anonymous LAN/internet callers
+    can no longer enumerate operator source URLs.
+    """
     from services.news_feed_config import get_feeds
     return get_feeds()
 
@@ -118,9 +127,18 @@ async def api_reset_news_feeds(request: Request):
 @router.get("/api/settings/node")
 @limiter.limit("30/minute")
 async def api_get_node_settings(request: Request):
+    """Issue #243 (tg12): node_mode and node_enabled are operational
+    posture. Anonymous callers receive an empty stub; authenticated
+    callers (local-operator or admin/scoped token) see the full
+    state. See the canonical handler in backend/main.py for the full
+    rationale.
+    """
     import asyncio
+    from auth import _scoped_view_authenticated
     from services.node_settings import read_node_settings
     data = await asyncio.to_thread(read_node_settings)
+    if not _scoped_view_authenticated(request, "node"):
+        return {}
     return {
         **data,
         "node_mode": _current_node_mode(),
@@ -210,9 +228,19 @@ async def api_set_meshtastic_mqtt_settings(request: Request, body: MeshtasticMqt
     return _meshtastic_runtime_snapshot()
 
 
-@router.get("/api/settings/timemachine")
+@router.get(
+    "/api/settings/timemachine",
+    dependencies=[Depends(require_local_operator)],
+)
 @limiter.limit("30/minute")
 async def api_get_timemachine_settings(request: Request):
+    """Issue #253 (tg12): archival-capture posture is operationally
+    sensitive — it tells a remote caller whether this deployment is
+    retaining replayable historical surveillance data. Gated on
+    local-operator so the Tauri shell and Docker bridge frontend
+    still see the toggle state, but anonymous LAN/internet callers
+    can no longer fingerprint Time Machine state.
+    """
     import asyncio
     from services.node_settings import read_node_settings
     data = await asyncio.to_thread(read_node_settings)

backend/routers/mesh_oracle.py

@@ -223,11 +223,21 @@ async def oracle_markets_more(request: Request, category: str = "NEWS", offset:
             "has_more": offset + limit < len(cat_markets), "total": len(cat_markets)}
 
 
-@router.post("/api/mesh/oracle/resolve")
+@router.post(
+    "/api/mesh/oracle/resolve",
+    dependencies=[Depends(require_admin)],
+)
 @limiter.limit("5/minute")
 @mesh_write_exempt(MeshWriteExemption.ADMIN_CONTROL)
 async def oracle_resolve(request: Request):
-    """Resolve a prediction market."""
+    """Resolve a prediction market.
+
+    Issue #240 (tg12): requires admin authentication. The
+    ``mesh_write_exempt`` decorator below is **metadata only** — it tags
+    the route as not requiring a mesh signed-write envelope, it does
+    NOT itself enforce caller authorization. The ``Depends(require_admin)``
+    on the route decorator is what actually gates access.
+    """
     from services.mesh.mesh_oracle import oracle_ledger
     body = await request.json()
     market_title = body.get("market_title", "")
@@ -327,11 +337,18 @@ async def oracle_predictions(request: Request, node_id: str = ""):
         active_predictions, authenticated=_scoped_view_authenticated(request, "mesh.audit"))
 
 
-@router.post("/api/mesh/oracle/resolve-stakes")
+@router.post(
+    "/api/mesh/oracle/resolve-stakes",
+    dependencies=[Depends(require_admin)],
+)
 @limiter.limit("5/minute")
 @mesh_write_exempt(MeshWriteExemption.ADMIN_CONTROL)
 async def oracle_resolve_stakes(request: Request):
-    """Resolve all expired stake contests."""
+    """Resolve all expired stake contests.
+
+    Issue #241 (tg12): requires admin authentication. See the note on
+    ``oracle_resolve`` above — ``mesh_write_exempt`` is metadata only.
+    """
     from services.mesh.mesh_oracle import oracle_ledger
     resolutions = oracle_ledger.resolve_expired_stakes()
     return {"ok": True, "resolutions": resolutions, "count": len(resolutions)}

backend/routers/wormhole.py

@@ -160,8 +160,13 @@ router = APIRouter()
 
 # --- Constants ---
 
-_WORMHOLE_PUBLIC_SETTINGS_FIELDS = {"enabled", "transport", "anonymous_mode"}
-_WORMHOLE_PUBLIC_PROFILE_FIELDS = {"profile", "wormhole_enabled"}
+# Issue #243 (tg12): the public redaction now exposes only the bare
+# "is this on?" boolean. Transport choice, anonymous-mode state, and
+# the named privacy profile were all leaking actionable recon to
+# unauthenticated callers and are now gated behind authenticated reads.
+# See the matching block in backend/main.py for the full rationale.
+_WORMHOLE_PUBLIC_SETTINGS_FIELDS = {"enabled"}
+_WORMHOLE_PUBLIC_PROFILE_FIELDS = {"wormhole_enabled"}
 _PRIVATE_LANE_CONTROL_FIELDS = {"private_lane_tier", "private_lane_policy"}
 _PUBLIC_RNS_STATUS_FIELDS = {"enabled", "ready", "configured_peers", "active_peers"}
 _NODE_PUBLIC_EVENT_HOOK_REGISTERED = False

backend/services/carrier_tracker.py

@@ -1,46 +1,90 @@
 """
 Carrier Strike Group OSINT Tracker
 ===================================
-Scrapes multiple OSINT sources to maintain current estimated positions
-for US Navy Carrier Strike Groups. Updates on startup + 00:00 & 12:00 UTC.
+Maintains estimated positions for US Navy Carrier Strike Groups with
+honest provenance and freshness signals.
 
-Sources:
-  1. GDELT News API — recent carrier movement headlines
-  2. WikiVoyage / public port-call databases
-  3. Fallback — last-known or static OSINT estimates
+Issues #244 / #245 / #246 (tg12 external audit):
+
+The previous implementation baked a snapshot of USNI News Fleet &
+Marine Tracker positions (March 9, 2026) into the registry as
+``fallback_lat``/``fallback_lng`` and stamped ``updated = now()``
+every time the dossier was rendered. That presented stale editorial
+data as live state. It also persisted GDELT-derived positions to the
+on-disk cache with no freshness signal, so a single news mention from
+months ago could keep overriding the (already-stale) registry default
+indefinitely.
+
+Architecture after this PR:
+
+::
+
+    backend/data/carrier_seed.json   read-only, shipped with image,
+                                     used ONCE on first-ever startup
+                                     to bootstrap carrier_cache.json.
+
+    backend/data/carrier_cache.json  mutable, lives in the runtime data
+                                     volume, written by every GDELT
+                                     refresh + any future source.
+
+Startup flow:
+
+1.  ``carrier_cache.json`` exists?  → load it.
+2.  Otherwise, copy ``carrier_seed.json`` → ``carrier_cache.json``,
+    then load it. (This happens once, ever, per install.)
+3.  Background: GDELT fetch runs. Any carrier mentioned in fresh news
+    gets its entry replaced with the news-derived position.
+    ``position_source_at`` is set to the news article timestamp.
+
+Freshness is a *labelling* decision, not an eviction decision:
+
+- ``position_source_at`` within the configurable freshness window
+  (default 14 days) → ``position_confidence = "recent"``.
+- Older than that              → ``position_confidence = "stale"``.
+- Bootstrapped from the seed file (never updated) → ``"seed"``.
+- No cache entry at all (e.g. a carrier added to the registry after
+  first install) → carrier renders at its homeport with
+  ``"homeport_default"``.
+
+Carriers are never hidden, never teleported, never disappeared. The
+position the user sees is always the last position the system actually
+observed, with an honest "as-of" timestamp the UI can render however
+it likes. A year from now, the runtime cache reflects whatever this
+install has observed via GDELT — not the seed snapshot.
 """
 
-import re
+import os
 import json
 import time
 import logging
 import threading
 import random
-from datetime import datetime, timezone
+import shutil
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
-from typing import Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 from services.network_utils import fetch_with_curl
 
 logger = logging.getLogger(__name__)
 
 # -----------------------------------------------------------------
-# Carrier registry: hull number → metadata + fallback position
+# Carrier registry: hull number → identity only.
+#
+# Issue #244 (tg12): the previous registry carried hard-coded
+# ``fallback_lat``/``fallback_lng`` that were dated editorial
+# snapshots from a 2026-03-09 article. Those fields are DELETED. The
+# registry is now identity + homeport only; positions are sourced
+# exclusively from carrier_cache.json (and via that, from the
+# bootstrap seed or live OSINT).
 # -----------------------------------------------------------------
 CARRIER_REGISTRY: Dict[str, dict] = {
-    # Fallback positions sourced from USNI News Fleet & Marine Tracker (Mar 9, 2026)
-    # https://news.usni.org/2026/03/09/usni-news-fleet-and-marine-tracker-march-9-2026
     # --- Bremerton, WA (Naval Base Kitsap) ---
-    # Distinct pier positions along Sinclair Inlet so carriers don't stack
     "CVN-68": {
         "name": "USS Nimitz (CVN-68)",
         "wiki": "https://en.wikipedia.org/wiki/USS_Nimitz",
         "homeport": "Bremerton, WA",
         "homeport_lat": 47.5535,
         "homeport_lng": -122.6400,
-        "fallback_lat": 47.5535,
-        "fallback_lng": -122.6400,
-        "fallback_heading": 90,
-        "fallback_desc": "Bremerton, WA (Maintenance)",
     },
     "CVN-76": {
         "name": "USS Ronald Reagan (CVN-76)",
@@ -48,23 +92,14 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "Bremerton, WA",
         "homeport_lat": 47.5580,
         "homeport_lng": -122.6360,
-        "fallback_lat": 47.5580,
-        "fallback_lng": -122.6360,
-        "fallback_heading": 90,
-        "fallback_desc": "Bremerton, WA (Decommissioning)",
     },
     # --- Norfolk, VA (Naval Station Norfolk) ---
-    # Piers run N-S along Willoughby Bay; each carrier gets a distinct berth
     "CVN-69": {
         "name": "USS Dwight D. Eisenhower (CVN-69)",
         "wiki": "https://en.wikipedia.org/wiki/USS_Dwight_D._Eisenhower",
         "homeport": "Norfolk, VA",
         "homeport_lat": 36.9465,
         "homeport_lng": -76.3265,
-        "fallback_lat": 36.9465,
-        "fallback_lng": -76.3265,
-        "fallback_heading": 0,
-        "fallback_desc": "Norfolk, VA (Post-deployment maintenance)",
     },
     "CVN-78": {
         "name": "USS Gerald R. Ford (CVN-78)",
@@ -72,10 +107,6 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "Norfolk, VA",
         "homeport_lat": 36.9505,
         "homeport_lng": -76.3250,
-        "fallback_lat": 18.0,
-        "fallback_lng": 39.5,
-        "fallback_heading": 0,
-        "fallback_desc": "Red Sea — Operation Epic Fury (USNI Mar 9)",
     },
     "CVN-74": {
         "name": "USS John C. Stennis (CVN-74)",
@@ -83,10 +114,6 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "Norfolk, VA",
         "homeport_lat": 36.9540,
         "homeport_lng": -76.3235,
-        "fallback_lat": 36.98,
-        "fallback_lng": -76.43,
-        "fallback_heading": 0,
-        "fallback_desc": "Newport News, VA (RCOH refueling overhaul)",
     },
     "CVN-75": {
         "name": "USS Harry S. Truman (CVN-75)",
@@ -94,10 +121,6 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "Norfolk, VA",
         "homeport_lat": 36.9580,
         "homeport_lng": -76.3220,
-        "fallback_lat": 36.0,
-        "fallback_lng": 15.0,
-        "fallback_heading": 0,
-        "fallback_desc": "Mediterranean Sea deployment (USNI Mar 9)",
     },
     "CVN-77": {
         "name": "USS George H.W. Bush (CVN-77)",
@@ -105,23 +128,14 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "Norfolk, VA",
         "homeport_lat": 36.9620,
         "homeport_lng": -76.3210,
-        "fallback_lat": 36.5,
-        "fallback_lng": -74.0,
-        "fallback_heading": 0,
-        "fallback_desc": "Atlantic — Pre-deployment workups (USNI Mar 9)",
     },
     # --- San Diego, CA (Naval Base San Diego) ---
-    # Carrier piers along the east shore of San Diego Bay, spread N-S
     "CVN-70": {
         "name": "USS Carl Vinson (CVN-70)",
         "wiki": "https://en.wikipedia.org/wiki/USS_Carl_Vinson",
         "homeport": "San Diego, CA",
         "homeport_lat": 32.6840,
         "homeport_lng": -117.1290,
-        "fallback_lat": 32.6840,
-        "fallback_lng": -117.1290,
-        "fallback_heading": 180,
-        "fallback_desc": "San Diego, CA (Homeport)",
     },
     "CVN-71": {
         "name": "USS Theodore Roosevelt (CVN-71)",
@@ -129,10 +143,6 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "San Diego, CA",
         "homeport_lat": 32.6885,
         "homeport_lng": -117.1280,
-        "fallback_lat": 32.6885,
-        "fallback_lng": -117.1280,
-        "fallback_heading": 180,
-        "fallback_desc": "San Diego, CA (Maintenance)",
     },
     "CVN-72": {
         "name": "USS Abraham Lincoln (CVN-72)",
@@ -140,10 +150,6 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "San Diego, CA",
         "homeport_lat": 32.6925,
         "homeport_lng": -117.1275,
-        "fallback_lat": 20.0,
-        "fallback_lng": 64.0,
-        "fallback_heading": 0,
-        "fallback_desc": "Arabian Sea — Operation Epic Fury (USNI Mar 9)",
     },
     # --- Yokosuka, Japan (CFAY) ---
     "CVN-73": {
@@ -152,16 +158,18 @@ CARRIER_REGISTRY: Dict[str, dict] = {
         "homeport": "Yokosuka, Japan",
         "homeport_lat": 35.2830,
         "homeport_lng": 139.6700,
-        "fallback_lat": 35.2830,
-        "fallback_lng": 139.6700,
-        "fallback_heading": 180,
-        "fallback_desc": "Yokosuka, Japan (Forward deployed)",
     },
 }
 
 # -----------------------------------------------------------------
-# Region → approximate center coordinates
-# Used to map textual geographic descriptions to lat/lng
+# Region → approximate center coordinates.
+#
+# Issue #245 (tg12): converting a region name straight into precise
+# map coordinates is false precision. We still use this table to
+# infer a coarse position from a headline mention, but the resulting
+# carrier object is now stamped ``position_confidence = "approximate"``
+# so the UI can render an uncertainty radius / dimmed icon. The
+# centroid is a best-effort midpoint of the named body of water.
 # -----------------------------------------------------------------
 REGION_COORDS: Dict[str, tuple] = {
     # Oceans & Seas
@@ -220,9 +228,39 @@ REGION_COORDS: Dict[str, tuple] = {
 }
 
 # -----------------------------------------------------------------
-# Cache file for persisting positions between restarts
+# Files
 # -----------------------------------------------------------------
-CACHE_FILE = Path(__file__).parent.parent / "carrier_cache.json"
+#
+# The seed lives in the read-only image data dir (it ships with each
+# release). The cache lives in the same data dir but is written at
+# runtime; under Docker compose this dir is volume-mounted so the
+# cache persists across container restarts, which is the whole point
+# of the seed-then-observe model — the user's runtime observations
+# survive image upgrades.
+SEED_FILE = Path(__file__).parent.parent / "data" / "carrier_seed.json"
+CACHE_FILE = Path(__file__).parent.parent / "data" / "carrier_cache.json"
+
+# -----------------------------------------------------------------
+# Freshness window for position_confidence labeling. Issue #246 (tg12):
+# previously persisted cache entries had no freshness signal at all.
+# After this change, the position itself is preserved (we never lose
+# what was last observed) but the confidence label flips from
+# "recent" to "stale" once the underlying source is older than this
+# window. Operator-overridable via env var.
+# -----------------------------------------------------------------
+_DEFAULT_FRESHNESS_WINDOW_DAYS = 14
+
+
+def _freshness_window_days() -> int:
+    raw = str(os.environ.get("SHADOWBROKER_CARRIER_FRESHNESS_DAYS", "") or "").strip()
+    if not raw:
+        return _DEFAULT_FRESHNESS_WINDOW_DAYS
+    try:
+        n = int(raw)
+        return n if n > 0 else _DEFAULT_FRESHNESS_WINDOW_DAYS
+    except (TypeError, ValueError):
+        return _DEFAULT_FRESHNESS_WINDOW_DAYS
+
 
 _carrier_positions: Dict[str, dict] = {}
 _positions_lock = threading.Lock()
@@ -234,25 +272,159 @@ _GDELT_REQUEST_DELAY_SECONDS = 1.25
 _GDELT_REQUEST_JITTER_SECONDS = 0.35
 
 
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _parse_iso(ts: str) -> Optional[datetime]:
+    if not ts:
+        return None
+    try:
+        # Python's fromisoformat accepts +00:00 but not 'Z' until 3.11.
+        normalized = ts.replace("Z", "+00:00")
+        dt = datetime.fromisoformat(normalized)
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        return dt
+    except (TypeError, ValueError):
+        return None
+
+
+def _compute_position_confidence(entry: dict, *, now: Optional[datetime] = None) -> str:
+    """Return the public confidence label for a carrier cache entry.
+
+    Order of precedence:
+      - explicit "homeport_default" / "seed" labels are preserved.
+      - dated entries (with position_source_at) are "recent" if within
+        the configured freshness window, else "stale".
+      - missing position_source_at falls through to "stale".
+    """
+    raw_label = str(entry.get("position_confidence", "") or "").strip()
+    # Explicit "kind of provenance" labels are preserved as-is. They
+    # describe HOW we got the position, not WHEN — a fresh headline-to-
+    # centroid match (#245) is still imprecise no matter how recently
+    # it was observed, and the seed (#244) is always the seed.
+    if raw_label in {"seed", "homeport_default", "approximate"}:
+        # Approximate entries can still age into "stale_approximate" if
+        # they fall out of the freshness window — that distinction lets
+        # the UI render a different badge for old-and-imprecise vs
+        # recent-and-imprecise. seed/homeport_default never age (they
+        # were never timestamped against real observations).
+        if raw_label == "approximate":
+            source_at = _parse_iso(str(entry.get("position_source_at", "") or ""))
+            if source_at is not None:
+                reference = now or datetime.now(timezone.utc)
+                if reference - source_at > timedelta(days=_freshness_window_days()):
+                    return "stale_approximate"
+        return raw_label
+
+    source_at = _parse_iso(str(entry.get("position_source_at", "") or ""))
+    if not source_at:
+        return "stale"
+
+    reference = now or datetime.now(timezone.utc)
+    window = timedelta(days=_freshness_window_days())
+    if reference - source_at <= window:
+        return "recent"
+    return "stale"
+
+
+def _load_seed() -> Dict[str, dict]:
+    """Load the read-only seed file shipped with the image.
+
+    Returns a hull→entry dict (no _meta wrapper). Missing or malformed
+    seed files yield an empty dict — the caller falls back to homeport
+    defaults.
+    """
+    try:
+        if not SEED_FILE.exists():
+            logger.info("Carrier seed file not present at %s; first-run will fall back to homeport defaults", SEED_FILE)
+            return {}
+        raw = json.loads(SEED_FILE.read_text(encoding="utf-8"))
+        carriers = raw.get("carriers", {}) if isinstance(raw, dict) else {}
+        if not isinstance(carriers, dict):
+            return {}
+        logger.info("Carrier seed loaded: %d entries from %s", len(carriers), SEED_FILE)
+        return carriers
+    except (IOError, OSError, json.JSONDecodeError, ValueError) as e:
+        logger.warning("Failed to load carrier seed file %s: %s", SEED_FILE, e)
+        return {}
+
+
 def _load_cache() -> Dict[str, dict]:
-    """Load cached carrier positions from disk."""
+    """Load the mutable cache (last-known positions persisted between restarts)."""
     try:
         if CACHE_FILE.exists():
-            data = json.loads(CACHE_FILE.read_text())
-            logger.info(f"Carrier cache loaded: {len(data)} carriers from {CACHE_FILE}")
-            return data
+            data = json.loads(CACHE_FILE.read_text(encoding="utf-8"))
+            if isinstance(data, dict):
+                logger.info("Carrier cache loaded: %d carriers from %s", len(data), CACHE_FILE)
+                return data
     except (IOError, OSError, json.JSONDecodeError, ValueError) as e:
-        logger.warning(f"Failed to load carrier cache: {e}")
+        logger.warning("Failed to load carrier cache: %s", e)
     return {}
 
 
-def _save_cache(positions: Dict[str, dict]):
-    """Persist carrier positions to disk."""
+def _save_cache(positions: Dict[str, dict]) -> None:
+    """Persist the mutable cache. Atomic write (temp + rename) so a crash
+    mid-write can't leave the file truncated."""
     try:
-        CACHE_FILE.write_text(json.dumps(positions, indent=2))
-        logger.info(f"Carrier cache saved: {len(positions)} carriers")
+        CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
+        tmp = CACHE_FILE.with_suffix(CACHE_FILE.suffix + ".tmp")
+        tmp.write_text(json.dumps(positions, indent=2), encoding="utf-8")
+        # On Windows os.replace is atomic and overwrites existing files.
+        os.replace(tmp, CACHE_FILE)
+        logger.info("Carrier cache saved: %d carriers", len(positions))
     except (IOError, OSError) as e:
-        logger.warning(f"Failed to save carrier cache: {e}")
+        logger.warning("Failed to save carrier cache: %s", e)
+
+
+def _homeport_entry_for(hull: str) -> Optional[dict]:
+    """Return a homeport-default cache entry for a hull, or None if the
+    hull is not in the registry."""
+    info = CARRIER_REGISTRY.get(hull)
+    if not info:
+        return None
+    return {
+        "lat": info["homeport_lat"],
+        "lng": info["homeport_lng"],
+        "heading": 0,
+        "desc": f"{info['homeport']} (no observations yet)",
+        "source": f"Homeport default ({info['homeport']})",
+        "source_url": info.get("wiki", ""),
+        "position_source_at": _now_iso(),
+        "position_confidence": "homeport_default",
+    }
+
+
+def _bootstrap_cache_if_missing() -> Dict[str, dict]:
+    """One-shot: if no cache exists, materialize one from the seed file.
+
+    Returns the cache contents (hull→entry). On first-ever startup,
+    this writes ``carrier_cache.json`` so subsequent restarts skip the
+    seed entirely. Operator-deleted caches re-bootstrap the same way —
+    operators can use that to "reset" carrier positions, but it's an
+    explicit operator action.
+    """
+    if CACHE_FILE.exists():
+        return _load_cache()
+
+    seed = _load_seed()
+    if not seed:
+        # No seed file either. Build a homeport-default cache so the
+        # first save_cache call still produces something honest.
+        homeports: Dict[str, dict] = {}
+        for hull in CARRIER_REGISTRY:
+            entry = _homeport_entry_for(hull)
+            if entry is not None:
+                homeports[hull] = entry
+        if homeports:
+            _save_cache(homeports)
+        return homeports
+
+    # Persist the seed as the first cache so subsequent runs skip this branch.
+    _save_cache(seed)
+    logger.info("Carrier cache bootstrapped from seed (first-ever startup)")
+    return dict(seed)
 
 
 def _match_region(text: str) -> Optional[tuple]:
@@ -270,10 +442,8 @@ def _match_carrier(text: str) -> Optional[str]:
     for hull, info in CARRIER_REGISTRY.items():
         hull_check = hull.lower().replace("-", "")
         name_parts = info["name"].lower()
-        # Match hull number (e.g., "CVN-78", "CVN78")
         if hull.lower() in text_lower or hull_check in text_lower.replace("-", ""):
             return hull
-        # Match ship name (e.g., "Ford", "Eisenhower", "Vinson")
         ship_name = name_parts.split("(")[0].strip()
         last_name = ship_name.split()[-1] if ship_name else ""
         if last_name and len(last_name) > 3 and last_name in text_lower:
@@ -323,8 +493,9 @@ def _fetch_gdelt_carrier_news() -> List[dict]:
             articles = data.get("articles", [])
             for art in articles:
                 title = art.get("title", "")
-                url = art.get("url", "")
-                results.append({"title": title, "url": url})
+                article_url = art.get("url", "")
+                article_at = art.get("seendate") or art.get("date") or ""
+                results.append({"title": title, "url": article_url, "seendate": article_at})
         except (ConnectionError, TimeoutError, ValueError, KeyError, OSError) as e:
             logger.debug(f"GDELT search failed for '{term}': {e}")
             continue
@@ -340,108 +511,139 @@ def _fetch_gdelt_carrier_news() -> List[dict]:
     return results
 
 
+def _gdelt_seendate_to_iso(seendate: str) -> Optional[str]:
+    """GDELT returns YYYYMMDDhhmmss (UTC). Convert to ISO8601 for
+    position_source_at. Returns None if the input is unparseable."""
+    raw = (seendate or "").strip()
+    if len(raw) < 8 or not raw.isdigit():
+        return None
+    try:
+        dt = datetime.strptime(raw[:14] if len(raw) >= 14 else raw[:8] + "000000", "%Y%m%d%H%M%S")
+        return dt.replace(tzinfo=timezone.utc).isoformat()
+    except (TypeError, ValueError):
+        return None
+
+
 def _parse_carrier_positions_from_news(articles: List[dict]) -> Dict[str, dict]:
-    """Parse carrier positions from news article titles and descriptions."""
+    """Parse carrier positions from news article titles.
+
+    Issue #245 (tg12): the position is a region centroid, which is
+    coarse — we now stamp ``position_confidence = "approximate"`` so
+    the UI can render that uncertainty. Issue #244: the
+    ``position_source_at`` field is the news article's actual seen
+    date, NOT now(), so the freshness check correctly flips entries
+    to "stale" once they age past the configured window.
+    """
     updates: Dict[str, dict] = {}
 
     for article in articles:
         title = article.get("title", "")
-
-        # Try to match a carrier from the title
         hull = _match_carrier(title)
         if not hull:
             continue
-
-        # Try to match a region from the title
         coords = _match_region(title)
         if not coords:
             continue
 
-        # Only update if we haven't seen this carrier yet (first match wins — most recent)
+        # First match wins (most recent article, GDELT returns newest first
+        # per term).
         if hull not in updates:
+            iso_at = _gdelt_seendate_to_iso(str(article.get("seendate", ""))) or _now_iso()
             updates[hull] = {
                 "lat": coords[0],
                 "lng": coords[1],
+                "heading": 0,
                 "desc": title[:100],
-                "source": "GDELT News API",
+                "source": "GDELT News API (headline region match — approximate)",
                 "source_url": article.get("url", "https://api.gdeltproject.org"),
-                "updated": datetime.now(timezone.utc).isoformat(),
+                "position_source_at": iso_at,
+                # Headline-to-centroid match is explicitly approximate.
+                "position_confidence": "approximate",
             }
             logger.info(
-                f"Carrier update: {CARRIER_REGISTRY[hull]['name']} → {coords} (from: {title[:80]})"
+                "Carrier update: %s → %s (from: %s)",
+                CARRIER_REGISTRY[hull]["name"],
+                coords,
+                title[:80],
             )
 
     return updates
 
 
-def _load_carrier_fallbacks() -> Dict[str, dict]:
-    """Build carrier positions from static fallbacks + disk cache (instant, no network)."""
-    positions: Dict[str, dict] = {}
-    for hull, info in CARRIER_REGISTRY.items():
-        positions[hull] = {
-            "name": info["name"],
-            "lat": info["fallback_lat"],
-            "lng": info["fallback_lng"],
-            "heading": info["fallback_heading"],
-            "desc": info["fallback_desc"],
-            "wiki": info["wiki"],
-            "source": "USNI News Fleet & Marine Tracker",
-            "source_url": "https://news.usni.org/category/fleet-tracker",
-            "updated": datetime.now(timezone.utc).isoformat(),
-        }
-
-    # Overlay cached positions from previous runs (may have GDELT data)
-    cached = _load_cache()
-    for hull, cached_pos in cached.items():
-        if hull in positions:
-            if cached_pos.get("source", "").startswith("GDELT") or cached_pos.get(
-                "source", ""
-            ).startswith("News"):
-                positions[hull].update(
-                    {
-                        "lat": cached_pos["lat"],
-                        "lng": cached_pos["lng"],
-                        "desc": cached_pos.get("desc", positions[hull]["desc"]),
-                        "source": cached_pos.get("source", "Cached OSINT"),
-                        "updated": cached_pos.get("updated", ""),
-                    }
-                )
-    return positions
+def _enrich_for_rendering(hull: str, entry: dict, *, now: Optional[datetime] = None) -> dict:
+    """Add live computed fields (confidence label, last_osint_update)
+    on top of the persisted cache entry. The persisted entry is left
+    untouched; this function builds the public-facing object.
+    """
+    info = CARRIER_REGISTRY.get(hull, {})
+    confidence = _compute_position_confidence(entry, now=now)
+    return {
+        "name": entry.get("name", info.get("name", hull)),
+        "lat": entry["lat"],
+        "lng": entry["lng"],
+        "heading": entry.get("heading", 0),
+        "desc": entry.get("desc", ""),
+        "wiki": entry.get("wiki", info.get("wiki", "")),
+        "source": entry.get("source", "OSINT estimated position"),
+        "source_url": entry.get("source_url", ""),
+        "position_source_at": entry.get("position_source_at", ""),
+        "position_confidence": confidence,
+        # Existing field preserved for backward compatibility with the
+        # current frontend ShipPopup; now reflects the SOURCE's observed
+        # time (not now()), so "last reported X days ago" is honest.
+        "last_osint_update": entry.get("position_source_at", ""),
+        # Convenience boolean for the UI: true when the position is
+        # NOT live OSINT (used to render dimmed icons / badges).
+        "is_fallback": confidence in {"seed", "stale", "stale_approximate", "homeport_default"},
+    }
 
 
-def update_carrier_positions():
-    """Main update function — called on startup and every 12h.
+def update_carrier_positions() -> None:
+    """Refresh carrier positions.
 
-    Phase 1 (instant): publish fallback + cached positions so the map has carriers immediately.
-    Phase 2 (slow):    query GDELT for fresh OSINT positions and update in-place.
+    Phase 1 (instant): publish whatever's in carrier_cache.json (or
+    bootstrap from seed on first-ever run), so the map has carriers
+    immediately.
+
+    Phase 2 (slow): query GDELT and replace position entries for any
+    carrier mentioned in fresh news. Persist back to cache.
     """
     global _last_update
 
-    # --- Phase 1: instant fallback + cache ---
-    positions = _load_carrier_fallbacks()
+    # --- Phase 1: instant cache (bootstrap from seed on first-ever run) ---
+    positions = _bootstrap_cache_if_missing()
+
+    # Ensure every registered hull has SOMETHING in the cache. A hull
+    # the seed didn't cover (e.g. added after install) renders at its
+    # homeport with "homeport_default" confidence.
+    for hull in CARRIER_REGISTRY:
+        if hull not in positions:
+            entry = _homeport_entry_for(hull)
+            if entry is not None:
+                positions[hull] = entry
 
     with _positions_lock:
-        # Only overwrite if positions are currently empty (first startup).
-        # If we already have data from a previous cycle, keep it while GDELT runs.
         if not _carrier_positions:
             _carrier_positions.update(positions)
             _last_update = datetime.now(timezone.utc)
     logger.info(
-        f"Carrier tracker: {len(positions)} carriers loaded from fallback/cache (GDELT enrichment starting...)"
+        "Carrier tracker: %d carriers loaded from cache (GDELT enrichment starting...)",
+        len(positions),
     )
 
-    # --- Phase 2: slow GDELT enrichment ---
+    # --- Phase 2: GDELT enrichment ---
     try:
         articles = _fetch_gdelt_carrier_news()
         news_positions = _parse_carrier_positions_from_news(articles)
         for hull, pos in news_positions.items():
-            if hull in positions:
-                positions[hull].update(pos)
-                logger.info(f"Carrier OSINT: updated {CARRIER_REGISTRY[hull]['name']} from news")
+            # Always overwrite — newest GDELT mention wins. The previous
+            # entry's position is preserved in git history and the next
+            # cycle either confirms or replaces it.
+            positions[hull] = pos
+            logger.info("Carrier OSINT: updated %s from news", CARRIER_REGISTRY[hull]["name"])
     except (ValueError, KeyError, json.JSONDecodeError, OSError) as e:
-        logger.warning(f"GDELT carrier fetch failed: {e}")
+        logger.warning("GDELT carrier fetch failed: %s", e)
 
-    # Save and update the global state with enriched positions
     with _positions_lock:
         _carrier_positions.clear()
         _carrier_positions.update(positions)
@@ -449,21 +651,15 @@ def update_carrier_positions():
 
     _save_cache(positions)
 
-    sources = {}
-    for p in positions.values():
-        src = p.get("source", "unknown")
-        sources[src] = sources.get(src, 0) + 1
-    logger.info(f"Carrier tracker: {len(positions)} carriers updated. Sources: {sources}")
+    confidences: Dict[str, int] = {}
+    for entry in positions.values():
+        label = _compute_position_confidence(entry)
+        confidences[label] = confidences.get(label, 0) + 1
+    logger.info("Carrier tracker: %d carriers updated. Confidence: %s", len(positions), confidences)
 
 
 def _deconflict_positions(result: List[dict]) -> List[dict]:
-    """Offset carriers that share identical coordinates so they don't stack.
-
-    At port: offset along the pier axis (~500m / 0.004° apart).
-    At sea: offset perpendicular to each other (~0.08° / ~9km apart)
-    so they're visibly separate but clearly operating together.
-    """
-    # Group by rounded lat/lng (within ~0.01° ≈ 1km = same spot)
+    """Offset carriers that share identical coordinates so they don't stack."""
     from collections import defaultdict
 
     groups: dict[str, list[int]] = defaultdict(list)
@@ -475,7 +671,6 @@ def _deconflict_positions(result: List[dict]) -> List[dict]:
         if len(indices) < 2:
             continue
         n = len(indices)
-        # Determine if this is a port (near a homeport) or at sea
         sample = result[indices[0]]
         at_port = any(
             abs(sample["lat"] - info.get("homeport_lat", 0)) < 0.05
@@ -484,7 +679,6 @@ def _deconflict_positions(result: List[dict]) -> List[dict]:
         )
 
         if at_port:
-            # Use each carrier's distinct homeport pier coordinates
             for idx in indices:
                 carrier = result[idx]
                 hull = None
@@ -497,8 +691,7 @@ def _deconflict_positions(result: List[dict]) -> List[dict]:
                     carrier["lat"] = info["homeport_lat"]
                     carrier["lng"] = info["homeport_lng"]
         else:
-            # At sea: spread in a line perpendicular to travel (~0.08° apart)
-            spacing = 0.08  # ~9km — close enough to see they're together
+            spacing = 0.08
             start_offset = -(n - 1) * spacing / 2
             for j, idx in enumerate(indices):
                 result[idx]["lng"] += start_offset + j * spacing
@@ -507,36 +700,44 @@ def _deconflict_positions(result: List[dict]) -> List[dict]:
 
 
 def get_carrier_positions() -> List[dict]:
-    """Return current carrier positions for the data pipeline."""
+    """Return current carrier positions for the data pipeline.
+
+    Each entry has the full provenance + freshness fields; the UI can
+    decide how to render them. Carriers are never hidden — only
+    labeled.
+    """
+    now = datetime.now(timezone.utc)
     with _positions_lock:
-        result = []
-        for hull, pos in _carrier_positions.items():
-            info = CARRIER_REGISTRY.get(hull, {})
+        result: List[dict] = []
+        for hull, entry in _carrier_positions.items():
+            enriched = _enrich_for_rendering(hull, entry, now=now)
             result.append(
                 {
-                    "name": pos.get("name", info.get("name", hull)),
+                    "name": enriched["name"],
                     "type": "carrier",
-                    "lat": pos["lat"],
-                    "lng": pos["lng"],
-                    "heading": None,  # Heading unknown for carriers — OSINT cannot determine true heading
+                    "lat": enriched["lat"],
+                    "lng": enriched["lng"],
+                    "heading": None,  # OSINT cannot determine true heading.
                     "sog": 0,
                     "cog": 0,
                     "country": "United States",
-                    "desc": pos.get("desc", ""),
-                    "wiki": pos.get("wiki", info.get("wiki", "")),
+                    "desc": enriched["desc"],
+                    "wiki": enriched["wiki"],
                     "estimated": True,
-                    "source": pos.get("source", "OSINT estimated position"),
-                    "source_url": pos.get(
-                        "source_url", "https://news.usni.org/category/fleet-tracker"
-                    ),
-                    "last_osint_update": pos.get("updated", ""),
+                    "source": enriched["source"],
+                    "source_url": enriched["source_url"],
+                    "last_osint_update": enriched["last_osint_update"],
+                    # New fields (additive — existing UI continues to work):
+                    "position_source_at": enriched["position_source_at"],
+                    "position_confidence": enriched["position_confidence"],
+                    "is_fallback": enriched["is_fallback"],
                 }
             )
         return _deconflict_positions(result)
 
 
 # -----------------------------------------------------------------
-# Scheduler: runs at startup, then at 00:00 and 12:00 UTC daily
+# Scheduler: runs at startup, then at 00:00 and 12:00 UTC daily.
 # -----------------------------------------------------------------
 _scheduler_thread: Optional[threading.Thread] = None
 _scheduler_stop = threading.Event()
@@ -544,7 +745,6 @@ _scheduler_stop = threading.Event()
 
 def _scheduler_loop():
     """Background thread that triggers updates at 00:00 and 12:00 UTC."""
-    # Initial update on startup
     try:
         update_carrier_positions()
     except Exception as e:
@@ -552,7 +752,6 @@ def _scheduler_loop():
 
     while not _scheduler_stop.is_set():
         now = datetime.now(timezone.utc)
-        # Next target: 00:00 or 12:00 UTC, whichever is sooner
         hour = now.hour
         if hour < 12:
             next_hour = 12
@@ -561,18 +760,17 @@ def _scheduler_loop():
 
         next_run = now.replace(hour=next_hour % 24, minute=0, second=0, microsecond=0)
         if next_hour == 24:
-            from datetime import timedelta
-
             next_run = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
 
         wait_seconds = (next_run - now).total_seconds()
         logger.info(
-            f"Carrier tracker: next update at {next_run.isoformat()} ({wait_seconds/3600:.1f}h)"
+            "Carrier tracker: next update at %s (%.1fh)",
+            next_run.isoformat(),
+            wait_seconds / 3600,
         )
 
-        # Wait until next scheduled time, or until stop event
         if _scheduler_stop.wait(timeout=wait_seconds):
-            break  # Stop event was set
+            break
 
         try:
             update_carrier_positions()

backend/services/config.py

@@ -53,6 +53,12 @@ class Settings(BaseSettings):
     MESH_RELAY_FAILURE_COOLDOWN_S: int = 120
     MESH_BOOTSTRAP_SEED_FAILURE_COOLDOWN_S: int = 15
     MESH_PEER_PUSH_SECRET: str = ""
+    # Issue #256 (tg12): optional per-peer HMAC secret map. Comma-separated
+    # `url=secret` pairs. When a peer URL appears here, only that per-peer
+    # secret is accepted for it — the global MESH_PEER_PUSH_SECRET above is
+    # ignored for that specific URL. Single-peer installs and unmigrated
+    # multi-peer installs leave this empty and behavior is unchanged.
+    MESH_PEER_SECRETS: str = ""
     MESH_RNS_APP_NAME: str = "shadowbroker"
     MESH_RNS_ASPECT: str = "infonet"
     MESH_RNS_IDENTITY_PATH: str = ""

backend/services/mesh/mesh_crypto.py

@@ -69,6 +69,115 @@ def _derive_peer_key(shared_secret: str, peer_url: str) -> bytes:
     ).digest()
 
 
+# ---------------------------------------------------------------------------
+# Issue #256 (tg12): per-peer HMAC secrets
+# ---------------------------------------------------------------------------
+#
+# Before this change, ALL peer-push HMACs were derived from a single
+# fleet-shared ``MESH_PEER_PUSH_SECRET``. The receiver could prove a
+# request was signed by *someone who knows the fleet secret*, but it
+# could NOT prove which peer signed it — any peer could compute the
+# expected HMAC for any other peer's URL and impersonate that peer.
+#
+# Fix: an optional ``MESH_PEER_SECRETS`` env var maps specific peer URLs
+# to per-peer secrets. When a peer URL is listed there, only that
+# per-peer secret is accepted for that URL — the global secret is
+# ignored for that peer. Peer A no longer learns peer B's secret, so
+# peer A cannot forge a request claiming to be peer B.
+#
+# Backwards-compatible by design:
+#
+# - Single-peer installs (``MESH_PEER_SECRETS`` empty) keep using the
+#   global secret. Zero behavior change. Zero operator action required.
+# - Multi-peer installs that haven't migrated yet keep using the global
+#   secret for every peer. Same behavior as before — same exposure.
+# - Multi-peer installs that have migrated configure
+#   ``MESH_PEER_SECRETS=urlA=secretA,urlB=secretB`` and immediately get
+#   per-peer identity. Migration is incremental: peers not yet listed
+#   continue using the global secret until both sides of that peering
+#   add their entry.
+
+_PEER_SECRETS_CACHE: dict[str, str] = {}
+_PEER_SECRETS_CACHE_RAW: str = ""
+
+
+def _lookup_per_peer_secret(normalized_url: str) -> str:
+    """Return the per-peer secret for ``normalized_url`` from MESH_PEER_SECRETS.
+
+    Returns "" if no per-peer entry is configured for that URL. The parser
+    is forgiving:
+
+    - Whitespace around items, URLs, and secrets is stripped.
+    - Items without ``=`` or with empty URL/secret halves are skipped.
+    - The URL half is normalized via ``normalize_peer_url`` so config
+      authors don't have to match scheme/port/path quirks exactly.
+
+    The cache is invalidated whenever the env var's raw value changes,
+    which keeps tests' ``monkeypatch.setenv`` calls effective without
+    forcing a process restart.
+    """
+    import os
+
+    raw = str(os.environ.get("MESH_PEER_SECRETS", "") or "").strip()
+
+    global _PEER_SECRETS_CACHE, _PEER_SECRETS_CACHE_RAW
+    if raw != _PEER_SECRETS_CACHE_RAW:
+        new_cache: dict[str, str] = {}
+        for chunk in raw.split(","):
+            chunk = chunk.strip()
+            if not chunk or "=" not in chunk:
+                continue
+            url_part, _, secret_part = chunk.partition("=")
+            normalized = normalize_peer_url(url_part.strip())
+            secret = secret_part.strip()
+            if normalized and secret:
+                new_cache[normalized] = secret
+        _PEER_SECRETS_CACHE = new_cache
+        _PEER_SECRETS_CACHE_RAW = raw
+
+    return _PEER_SECRETS_CACHE.get(normalized_url, "")
+
+
+def resolve_peer_key_for_url(peer_url: str) -> bytes:
+    """Return the HMAC key for ``peer_url``, preferring per-peer secret.
+
+    Issue #256: this is the function every peer-push call site should
+    use. It looks up the peer-specific secret first, falling back to the
+    fleet-shared ``MESH_PEER_PUSH_SECRET`` only when the URL is NOT
+    listed in ``MESH_PEER_SECRETS``.
+
+    Both sender (computing X-Peer-HMAC) and receiver (verifying it) call
+    this with the SENDER's URL — they must derive the same key, so
+    operators on both ends of a peering need matching MESH_PEER_SECRETS
+    entries for that URL to stay in sync.
+
+    Returns empty bytes when no usable secret exists. Callers must treat
+    that as fail-closed (skip the push, reject the verification).
+    """
+    normalized_url = normalize_peer_url(peer_url)
+    if not normalized_url:
+        return b""
+
+    per_peer_secret = _lookup_per_peer_secret(normalized_url)
+    if per_peer_secret:
+        return _derive_peer_key(per_peer_secret, normalized_url)
+
+    # No per-peer entry for this URL — fall back to the legacy global
+    # secret. This is what preserves zero-hostility for single-peer
+    # installs and the migration window for multi-peer installs.
+    try:
+        from services.config import get_settings
+
+        global_secret = str(
+            getattr(get_settings(), "MESH_PEER_PUSH_SECRET", "") or ""
+        ).strip()
+    except Exception:
+        return b""
+    if not global_secret:
+        return b""
+    return _derive_peer_key(global_secret, normalized_url)
+
+
 def _node_digest(public_key_b64: str) -> str:
     raw = base64.b64decode(public_key_b64)
     return hashlib.sha256(raw).hexdigest()

backend/services/mesh/mesh_hashchain.py

@@ -216,18 +216,19 @@ def _peer_pair_ref_key(peer_url: str) -> bytes:
     Returns an empty key on misconfiguration so callers fail closed.
     """
     try:
-        from services.config import get_settings
-        from services.mesh.mesh_crypto import _derive_peer_key, normalize_peer_url
-
-        secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
+        from services.mesh.mesh_crypto import (
+            normalize_peer_url,
+            resolve_peer_key_for_url,
+        )
     except Exception:
         return b""
-    if not secret:
-        return b""
     normalized = normalize_peer_url(peer_url or "")
     if not normalized:
         return b""
-    peer_key = _derive_peer_key(secret, normalized)
+    # Issue #256: resolve_peer_key_for_url() prefers per-peer secrets
+    # from MESH_PEER_SECRETS and falls back to the global
+    # MESH_PEER_PUSH_SECRET only when the URL has no per-peer entry.
+    peer_key = resolve_peer_key_for_url(normalized)
     if not peer_key:
         return b""
     # Domain-separate from the transport HMAC key so the two

backend/services/mesh/mesh_router.py

@@ -26,7 +26,11 @@ from enum import Enum
 from typing import Any, Callable, Optional
 from collections import deque
 from urllib.parse import urlparse
-from services.mesh.mesh_crypto import _derive_peer_key, normalize_peer_url
+from services.mesh.mesh_crypto import (
+    _derive_peer_key,
+    normalize_peer_url,
+    resolve_peer_key_for_url,
+)
 from services.mesh.mesh_metrics import increment as metrics_inc
 from services.mesh.mesh_privacy_policy import (
     TRANSPORT_TIER_ORDER as _TIER_RANK,
@@ -703,7 +707,6 @@ class InternetTransport(_PeerPushTransportMixin):
             endpoint_path, padded = self._build_peer_push_request(envelope, self.NAME)
         except ValueError as exc:
             return TransportResult(False, self.NAME, str(exc))
-        secret = str(settings.MESH_PEER_PUSH_SECRET or "").strip()
 
         delivered = 0
         last_error = ""
@@ -713,10 +716,13 @@ class InternetTransport(_PeerPushTransportMixin):
             try:
                 normalized_peer_url = normalize_peer_url(peer_url)
                 headers = {"Content-Type": "application/json"}
-                if secret:
-                    peer_key = _derive_peer_key(secret, normalized_peer_url)
-                    if not peer_key:
-                        raise ValueError("invalid peer URL for HMAC derivation")
+                # Issue #256: per-peer secret takes precedence over the
+                # global MESH_PEER_PUSH_SECRET. When neither is set the
+                # key is empty and we skip the HMAC header entirely so a
+                # bare (unsigned) push still works on test deployments
+                # that have not yet configured any secret at all.
+                peer_key = resolve_peer_key_for_url(normalized_peer_url)
+                if peer_key:
                     headers["X-Peer-Url"] = normalized_peer_url
                     headers["X-Peer-HMAC"] = hmac.new(
                         peer_key,
@@ -798,7 +804,6 @@ class TorArtiTransport(_PeerPushTransportMixin):
             endpoint_path, padded = self._build_peer_push_request(envelope, self.NAME)
         except ValueError as exc:
             return TransportResult(False, self.NAME, str(exc))
-        secret = str(settings.MESH_PEER_PUSH_SECRET or "").strip()
 
         delivered = 0
         last_error = ""
@@ -808,10 +813,10 @@ class TorArtiTransport(_PeerPushTransportMixin):
             try:
                 normalized_peer_url = normalize_peer_url(peer_url)
                 headers = {"Content-Type": "application/json"}
-                if secret:
-                    peer_key = _derive_peer_key(secret, normalized_peer_url)
-                    if not peer_key:
-                        raise ValueError("invalid peer URL for HMAC derivation")
+                # Issue #256: per-peer secret takes precedence; see the
+                # other transport above for the rationale.
+                peer_key = resolve_peer_key_for_url(normalized_peer_url)
+                if peer_key:
                     headers["X-Peer-Url"] = normalized_peer_url
                     headers["X-Peer-HMAC"] = hmac.new(
                         peer_key,

backend/services/mesh/mesh_wormhole_prekey.py

@@ -91,13 +91,15 @@ def _fetch_dm_prekey_bundle_from_peer_lookup(lookup_token: str) -> dict[str, Any
         return {"ok": False, "detail": "lookup token required"}
     try:
         from services.config import get_settings
-        from services.mesh.mesh_crypto import _derive_peer_key, normalize_peer_url
+        from services.mesh.mesh_crypto import (
+            normalize_peer_url,
+            resolve_peer_key_for_url,
+        )
         from services.mesh.mesh_router import configured_relay_peer_urls
 
         settings = get_settings()
-        secret = str(getattr(settings, "MESH_PEER_PUSH_SECRET", "") or "").strip()
-        if not secret:
-            return {"ok": False, "detail": "peer prekey lookup unavailable"}
+        # Issue #256: secret check moved per-peer below. We still bail out
+        # cleanly when there are no peers configured at all.
         peers = configured_relay_peer_urls()
         if not peers:
             return {"ok": False, "detail": "peer prekey lookup unavailable"}
@@ -121,7 +123,8 @@ def _fetch_dm_prekey_bundle_from_peer_lookup(lookup_token: str) -> dict[str, Any
             or os.environ.get("SB_TEST_NODE_URL", "").strip()
             or normalized_peer_url
         )
-        peer_key = _derive_peer_key(secret, sender_peer_url)
+        # Issue #256: prefer per-peer secret keyed by the sender URL.
+        peer_key = resolve_peer_key_for_url(sender_peer_url)
         if not peer_key:
             continue
         headers = {

backend/services/region_dossier.py

@@ -4,7 +4,7 @@ import concurrent.futures
 from urllib.parse import quote
 import requests as _requests
 from cachetools import TTLCache
-from services.network_utils import fetch_with_curl
+from services.network_utils import fetch_with_curl, DEFAULT_USER_AGENT
 
 logger = logging.getLogger(__name__)
 
@@ -15,6 +15,25 @@ dossier_cache = TTLCache(maxsize=500, ttl=86400)
 # Nominatim requires max 1 req/sec — track last call time
 _nominatim_last_call = 0.0
 
+# Issue #218 / #219 (tg12): Wikimedia's User-Agent policy requires API
+# clients to identify themselves with a stable User-Agent that includes
+# a contact path. Bare "python-requests/x.y" or generic strings violate
+# the policy and risk getting blocked. We send the project default UA
+# (operator-overridable via SHADOWBROKER_USER_AGENT) on EVERY outbound
+# Wikimedia request, plus the policy-recommended Api-User-Agent which
+# Wikimedia explicitly accepts on top of the regular UA.
+#
+# This is documented and stable so a Wikimedia operator who wants to
+# rate-limit or contact us has a fixed identifier to grep for.
+_WIKIMEDIA_REQUEST_HEADERS = {
+    "User-Agent": DEFAULT_USER_AGENT,
+    "Api-User-Agent": (
+        f"{DEFAULT_USER_AGENT} "
+        "(+https://github.com/BigBodyCobain/Shadowbroker; "
+        "report issues at /issues)"
+    ),
+}
+
 
 def _reverse_geocode_offline(lat: float, lng: float) -> dict:
     """Offline fallback via reverse_geocoder when external reverse geocoding is blocked."""
@@ -121,7 +140,13 @@ def _fetch_wikidata_leader(country_name: str) -> dict:
     """
     url = f"https://query.wikidata.org/sparql?query={quote(sparql)}&format=json"
     try:
-        res = fetch_with_curl(url, timeout=6)
+        # Issue #218 (tg12): Wikimedia's User-Agent policy requires
+        # outbound API traffic to be identifiable. fetch_with_curl()
+        # sends the project default, and we also add the Wikimedia-
+        # specific Api-User-Agent that the policy specifically asks
+        # for, since this request originates from a backend service
+        # that proxies on behalf of (potentially many) browser users.
+        res = fetch_with_curl(url, timeout=6, headers=_WIKIMEDIA_REQUEST_HEADERS)
         if res.status_code == 200:
             results = res.json().get("results", {}).get("bindings", [])
             if results:
@@ -147,7 +172,9 @@ def _fetch_local_wiki_summary(place_name: str, country_name: str = "") -> dict:
         slug = quote(name.replace(" ", "_"))
         url = f"https://en.wikipedia.org/api/rest_v1/page/summary/{slug}"
         try:
-            res = fetch_with_curl(url, timeout=5)
+            # Issue #219 (tg12): identify ourselves to Wikimedia per
+            # their UA policy; see _fetch_wikidata_leader above.
+            res = fetch_with_curl(url, timeout=5, headers=_WIKIMEDIA_REQUEST_HEADERS)
             if res.status_code == 200:
                 data = res.json()
                 if data.get("type") != "disambiguation":

backend/services/tor_hidden_service.py

@@ -173,6 +173,94 @@ def _verify_tor_bundle(archive_path: Path, bundle_url: str) -> tuple[bool, str]:
     return True, f"https-only (no digest source reachable, archive={actual_hash[:16]}...)"
 
 
+def _extract_tor_bundle_safely(archive_path: Path, install_dir: Path) -> bool:
+    """Extract a Tor Expert Bundle tar.gz safely.
+
+    Issue #251: the previous extractor checked tarinfo.name against path
+    traversal but never inspected tarinfo.linkname for symlink/hardlink
+    members. Python 3.11's tarfile honors symlinks during extractall(),
+    so a malicious archive could ship a member like::
+
+        name     = "innocent.txt"          # passes the path check
+        type     = SYMTYPE
+        linkname = "C:\\Windows\\System32\\config\\system"
+
+    and extractall() would then create that symlink. Subsequent reads
+    of innocent.txt deference to a sensitive system file; subsequent
+    writes corrupt one. Tor bundles never legitimately contain symlinks
+    or hardlinks, so we refuse all link members categorically rather
+    than trying to validate linkname targets (which has its own pitfalls
+    around relative path resolution).
+
+    Also refuses non-regular-non-directory members (devices, FIFOs,
+    character/block special files) for completeness — none of those
+    belong in a Tor Expert Bundle and accepting them is a category of
+    bug we don't need to debug later.
+
+    Returns True on success, False on rejection (and logs the reason).
+    The caller is responsible for cleaning up the archive file.
+    """
+    import tarfile
+
+    install_resolved = install_dir.resolve()
+
+    try:
+        with tarfile.open(str(archive_path), "r:gz") as tar:
+            for member in tar.getmembers():
+                # Reject anything that isn't a regular file or directory.
+                # Symlinks (SYMTYPE) and hardlinks (LNKTYPE) are the
+                # path-traversal vectors; the others (CHRTYPE, BLKTYPE,
+                # FIFOTYPE, CONTTYPE) have no legitimate use in a Tor
+                # Expert Bundle.
+                if member.issym() or member.islnk():
+                    logger.error(
+                        "Tor bundle extraction blocked: link member %s -> %s "
+                        "(symlinks/hardlinks are not allowed in Tor bundles; "
+                        "this archive is malformed or hostile)",
+                        member.name,
+                        member.linkname,
+                    )
+                    return False
+                if not (member.isfile() or member.isdir()):
+                    logger.error(
+                        "Tor bundle extraction blocked: unexpected member type "
+                        "for %s (only regular files and directories are allowed)",
+                        member.name,
+                    )
+                    return False
+
+                # Path traversal check (preserves the original guard).
+                try:
+                    member_path = (install_dir / member.name).resolve()
+                except OSError as exc:
+                    logger.error(
+                        "Tor bundle extraction blocked: cannot resolve member "
+                        "path %s: %s",
+                        member.name,
+                        exc,
+                    )
+                    return False
+                try:
+                    member_path.relative_to(install_resolved)
+                except ValueError:
+                    logger.error(
+                        "Tor bundle extraction blocked: path traversal on %s "
+                        "(resolves to %s, outside install dir %s)",
+                        member.name,
+                        member_path,
+                        install_resolved,
+                    )
+                    return False
+
+            # All members validated — extract.
+            tar.extractall(path=str(install_dir))
+    except tarfile.TarError as exc:
+        logger.error("Tor bundle extraction failed: malformed tar (%s)", exc)
+        return False
+
+    return True
+
+
 def _auto_install_tor() -> str | None:
     """Install or download Tor when it is safe to do so."""
     if os.name != "nt":
@@ -203,14 +291,9 @@ def _auto_install_tor() -> str | None:
             logger.info("Download complete, extracting...")
             import tarfile
 
-            with tarfile.open(str(archive_path), "r:gz") as tar:
-                for member in tar.getmembers():
-                    member_path = (TOR_INSTALL_DIR / member.name).resolve()
-                    if not str(member_path).startswith(str(TOR_INSTALL_DIR.resolve())):
-                        logger.error("Tar path traversal blocked: %s", member.name)
-                        archive_path.unlink(missing_ok=True)
-                        return None
-                tar.extractall(path=str(TOR_INSTALL_DIR))
+            if not _extract_tor_bundle_safely(archive_path, TOR_INSTALL_DIR):
+                archive_path.unlink(missing_ok=True)
+                return None
 
             archive_path.unlink(missing_ok=True)
 

backend/services/updater.py

@@ -6,9 +6,11 @@ Public API:
     schedule_restart(project_root)           (spawn detached start script, then exit)
 """
 
+import json
 import os
 import sys
 import logging
+import re
 import shutil
 import subprocess
 import tempfile
@@ -29,6 +31,19 @@ DOCKER_UPDATE_COMMANDS = (
     "docker compose pull && docker compose up -d"
 )
 
+# Issue #231: baked-in release digests. Loaded lazily, used as a fallback
+# verification source when the release's SHA256SUMS.txt asset can't be
+# fetched (e.g. transient network failure during update).
+_RELEASE_DIGESTS_FILE = (
+    Path(__file__).resolve().parent.parent / "data" / "release_digests.json"
+)
+# Pattern for the maintainer's signed source-archive release asset. This
+# is the file we prefer over the auto-generated ``zipball_url`` because
+# the maintainer's build process publishes it with a matching entry in
+# SHA256SUMS.txt — the zipball does not have a signed digest.
+_SOURCE_ASSET_PATTERN = re.compile(r"^ShadowBroker_v\d", re.IGNORECASE)
+_SHA256SUMS_ASSET_NAME = "SHA256SUMS.txt"
+
 
 def _is_docker() -> bool:
     """Detect if we're running inside a Docker container."""
@@ -40,7 +55,6 @@ def _is_docker() -> bool:
     except (FileNotFoundError, PermissionError):
         pass
     return os.environ.get("container") == "docker"
-_EXPECTED_SHA256 = os.environ.get("MESH_UPDATE_SHA256", "").strip().lower()
 _ALLOWED_UPDATE_HOSTS = {
     "api.github.com",
     "codeload.github.com",
@@ -119,7 +133,16 @@ def _validate_update_url(url: str, *, allow_release_page: bool = False) -> str:
 # ---------------------------------------------------------------------------
 def _download_release(temp_dir: str) -> tuple:
     """Fetch latest release info and download the source zip archive.
-    Returns (zip_path, version_tag, download_url, release_url).
+
+    Issue #231: prefer the maintainer's signed release asset (matching
+    ``ShadowBroker_v*.zip``) over the auto-generated ``zipball_url``,
+    because the maintainer's release process publishes a matching entry
+    in SHA256SUMS.txt for the named asset but NOT for the zipball.
+
+    Returns (zip_path, version_tag, download_url, release_url, asset_name,
+    sha256sums_url) — the last two are empty strings when the release
+    doesn't publish a signed asset, falling back to the legacy zipball
+    path.
     """
     logger.info("Fetching latest release info from GitHub...")
     _validate_update_url(GITHUB_RELEASES_URL)
@@ -131,9 +154,42 @@ def _download_release(temp_dir: str) -> tuple:
     tag = release.get("tag_name", "unknown")
     release_url = str(release.get("html_url") or GITHUB_RELEASES_PAGE_URL).strip()
     _validate_update_url(release_url, allow_release_page=True)
-    zip_url = str(release.get("zipball_url") or "").strip()
-    if not zip_url:
-        raise RuntimeError("Latest release is missing a source archive URL")
+
+    # Prefer the maintainer-signed release asset. Fall back to the
+    # auto-generated zipball if the release doesn't publish one.
+    assets = release.get("assets") or []
+    asset_name = ""
+    asset_url = ""
+    sha256sums_url = ""
+    for a in assets:
+        name = str(a.get("name") or "").strip()
+        download = str(a.get("browser_download_url") or "").strip()
+        if not name or not download:
+            continue
+        if _SOURCE_ASSET_PATTERN.match(name) and name.lower().endswith(".zip"):
+            asset_name = name
+            asset_url = download
+        elif name == _SHA256SUMS_ASSET_NAME:
+            sha256sums_url = download
+
+    if asset_url:
+        zip_url = asset_url
+        logger.info(
+            "Using signed release asset %s (sha256sums=%s)",
+            asset_name,
+            "yes" if sha256sums_url else "no",
+        )
+    else:
+        zip_url = str(release.get("zipball_url") or "").strip()
+        if not zip_url:
+            raise RuntimeError("Latest release is missing a source archive URL")
+        logger.warning(
+            "Release does not publish a signed ShadowBroker_v*.zip asset — "
+            "falling back to auto-generated zipball_url. Integrity will be "
+            "verified against the baked-in release_digests.json (if present) "
+            "or HTTPS-only otherwise."
+        )
+
     _validate_update_url(zip_url)
 
     logger.info(f"Downloading {zip_url} ...")
@@ -150,19 +206,174 @@ def _download_release(temp_dir: str) -> tuple:
 
     size_mb = os.path.getsize(zip_path) / (1024 * 1024)
     logger.info(f"Downloaded {size_mb:.1f} MB — ZIP validated OK")
-    return zip_path, tag, zip_url, release_url
+    return zip_path, tag, zip_url, release_url, asset_name, sha256sums_url
 
 
-def _validate_zip_hash(zip_path: str) -> None:
-    if not _EXPECTED_SHA256:
-        return
+def _compute_sha256(zip_path: str) -> str:
+    """Return the hex SHA-256 of the file at ``zip_path`` (lowercase)."""
     h = hashlib.sha256()
     with open(zip_path, "rb") as f:
         for chunk in iter(lambda: f.read(1024 * 128), b""):
             h.update(chunk)
-    digest = h.hexdigest().lower()
-    if digest != _EXPECTED_SHA256:
-        raise RuntimeError("Update SHA-256 mismatch")
+    return h.hexdigest().lower()
+
+
+def _load_baked_in_release_digests() -> dict:
+    """Return the ``release_digests.json`` mapping, or an empty dict.
+
+    Schema (issue #231):
+        {
+          "<release_tag>": {
+            "<asset_filename>": "<sha256_hex>",
+            ...
+          },
+          ...
+        }
+    """
+    try:
+        raw = _RELEASE_DIGESTS_FILE.read_text(encoding="utf-8")
+        parsed = json.loads(raw)
+    except (OSError, ValueError) as exc:
+        logger.debug("Release digest file unreadable: %s", exc)
+        return {}
+    if not isinstance(parsed, dict):
+        return {}
+    cleaned: dict[str, dict[str, str]] = {}
+    for k, v in parsed.items():
+        if not isinstance(k, str) or k.startswith("_"):
+            continue
+        if isinstance(v, dict):
+            entries = {
+                fname: digest.strip().lower()
+                for fname, digest in v.items()
+                if isinstance(fname, str) and isinstance(digest, str)
+            }
+            if entries:
+                cleaned[k] = entries
+    return cleaned
+
+
+def _fetch_sha256sums(sha256sums_url: str) -> dict[str, str]:
+    """Download a SHA256SUMS.txt and return {filename: digest_hex_lower}.
+
+    Standard ``sha256sum`` format: ``<digest>  <filename>`` per line. The
+    leading ``*`` binary-mode marker (e.g. ``<digest> *<filename>``) is
+    handled.
+    """
+    try:
+        _validate_update_url(sha256sums_url)
+    except RuntimeError as exc:
+        logger.warning("SHA256SUMS URL rejected: %s", exc)
+        return {}
+    try:
+        resp = requests.get(sha256sums_url, timeout=15)
+        resp.raise_for_status()
+    except requests.RequestException as exc:
+        logger.info("SHA256SUMS fetch failed: %s", exc)
+        return {}
+    out: dict[str, str] = {}
+    for line in resp.text.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#"):
+            continue
+        # Tolerant split: handle both `<digest>  <name>` and `<digest> *<name>`.
+        parts = line.split(None, 1)
+        if len(parts) != 2:
+            continue
+        digest, fname = parts
+        fname = fname.lstrip("*").strip()
+        digest = digest.strip().lower()
+        if len(digest) == 64 and all(c in "0123456789abcdef" for c in digest) and fname:
+            out[fname] = digest
+    return out
+
+
+def _validate_zip_hash(
+    zip_path: str,
+    *,
+    asset_name: str = "",
+    sha256sums_url: str = "",
+    release_tag: str = "",
+) -> str:
+    """Verify the downloaded archive against trusted digest sources.
+
+    Issue #231: previously this returned silently when ``MESH_UPDATE_SHA256``
+    was unset, which made the auto-updater a supply-chain RCE vector on any
+    compromise of the GitHub release pipeline. The chain now is:
+
+      1. ``MESH_UPDATE_SHA256`` env var (operator override — preserved for
+         power-users who want to pin an exact digest manually)
+      2. ``SHA256SUMS.txt`` release asset (primary — the maintainer's
+         release process already publishes this)
+      3. Baked-in ``backend/data/release_digests.json`` (second line of
+         defense for releases that lack the SHA256SUMS asset, or when the
+         asset can't be fetched at update time)
+      4. HTTPS-only fallback with a loud warning (preserves the auto-update
+         flow during transient outages — but never silently)
+
+    A mismatch from a source that DID respond is fatal: the update is
+    refused and the existing install keeps running. Only the "no source
+    reachable at all" case falls back to HTTPS-only.
+
+    Returns a short human-readable description of which source verified
+    the archive (used in the update-success message).
+    """
+    actual = _compute_sha256(zip_path)
+
+    # Source 1: explicit operator override.
+    override = os.environ.get("MESH_UPDATE_SHA256", "").strip().lower()
+    if override:
+        if actual == override:
+            return f"verified via MESH_UPDATE_SHA256 ({actual[:16]}...)"
+        raise RuntimeError(
+            f"Update SHA-256 mismatch vs MESH_UPDATE_SHA256: archive={actual[:16]}..., "
+            f"expected={override[:16]}..."
+        )
+
+    # Source 2: SHA256SUMS.txt asset from the release.
+    sums_map: dict[str, str] = {}
+    if sha256sums_url and asset_name:
+        sums_map = _fetch_sha256sums(sha256sums_url)
+
+    sums_expected = sums_map.get(asset_name) if asset_name else None
+    if sums_expected:
+        if actual == sums_expected:
+            return f"verified via release SHA256SUMS.txt ({actual[:16]}...)"
+        raise RuntimeError(
+            f"Update SHA-256 mismatch vs release SHA256SUMS.txt: "
+            f"archive={actual[:16]}..., expected={sums_expected[:16]}..."
+        )
+
+    # Source 3: baked-in digest list.
+    baked = _load_baked_in_release_digests()
+    baked_expected = ""
+    if release_tag and asset_name:
+        baked_expected = baked.get(release_tag, {}).get(asset_name, "")
+    if baked_expected:
+        if actual == baked_expected:
+            return f"verified via baked-in digest list ({actual[:16]}...)"
+        raise RuntimeError(
+            f"Update SHA-256 mismatch vs baked-in digest list: "
+            f"archive={actual[:16]}..., expected={baked_expected[:16]}..."
+        )
+
+    # Source 4: HTTPS-only fallback. We keep onboarding/auto-update working
+    # during transient outages (no SHA256SUMS reachable AND no baked-in
+    # entry for this release), but surface the degraded posture loudly so
+    # the operator can see it in logs and the maintainer can populate the
+    # digest list on the next release bump.
+    logger.warning(
+        "Update integrity check fell back to HTTPS-only trust "
+        "(no SHA256SUMS.txt response and no baked-in digest for "
+        "release=%s asset=%s). The archive SHA-256 is %s. Once the "
+        "release ships a SHA256SUMS.txt asset OR backend/data/"
+        "release_digests.json is updated with this release, the secure "
+        "path will activate automatically.",
+        release_tag or "unknown",
+        asset_name or "unknown",
+        actual,
+    )
+    return f"https-only (no digest source reachable, archive={actual[:16]}...)"
 
 
 def _is_source_checkout(project_root: str) -> bool:
@@ -334,7 +545,7 @@ def perform_update(project_root: str) -> dict:
     temp_dir = tempfile.mkdtemp(prefix="sb_update_")
     manual_url = GITHUB_RELEASES_PAGE_URL
     try:
-        zip_path, version, url, release_url = _download_release(temp_dir)
+        zip_path, version, url, release_url, asset_name, sha256sums_url = _download_release(temp_dir)
         manual_url = release_url or manual_url
 
         if in_docker:
@@ -366,7 +577,13 @@ def perform_update(project_root: str) -> dict:
                 ),
             }
 
-        _validate_zip_hash(zip_path)
+        verification_note = _validate_zip_hash(
+            zip_path,
+            asset_name=asset_name,
+            sha256sums_url=sha256sums_url,
+            release_tag=version,
+        )
+        logger.info("Update archive %s", verification_note)
         backup_path = _backup_current(project_root, temp_dir)
         copied = _extract_and_copy(zip_path, project_root, temp_dir)
 
@@ -378,6 +595,7 @@ def perform_update(project_root: str) -> dict:
             "manual_url": manual_url,
             "release_url": release_url,
             "download_url": url,
+            "integrity": verification_note,
             "message": f"Updated to {version} — {copied} files replaced. Restarting...",
         }
     except Exception as e:

backend/tests/test_carrier_tracker_quality.py

@@ -0,0 +1,389 @@
+"""Issues #244, #245, #246 (tg12 external audit): carrier tracker
+quality + provenance + freshness.
+
+These tests pin the post-fix contract:
+
+- **#244**: dated editorial snapshot positions no longer live in the
+  registry. They live in a one-shot seed file that is consumed once
+  on first-ever startup. After that, the runtime cache reflects only
+  what THIS install has actually observed.
+
+- **#245**: headline-derived positions (centroid of a region keyword)
+  are stamped ``position_confidence = "approximate"`` so the UI can
+  render them with appropriate uncertainty.
+
+- **#246**: freshness is a *labelling* decision, not an eviction
+  decision. Positions older than the configurable freshness window
+  flip from ``"recent"`` to ``"stale"`` but are NEVER replaced with
+  the registry default — that would teleport the carrier. The user
+  always sees the last position the system actually observed.
+"""
+from __future__ import annotations
+
+import json
+import os
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+
+@pytest.fixture
+def fresh_tracker(tmp_path, monkeypatch):
+    """Isolated carrier_tracker with seed/cache paths redirected to tmp.
+
+    Yields the module so tests can call its functions; resets globals
+    between tests so position caches don't leak across cases.
+    """
+    from services import carrier_tracker
+
+    seed_path = tmp_path / "data" / "carrier_seed.json"
+    cache_path = tmp_path / "carrier_cache.json"
+    seed_path.parent.mkdir(parents=True, exist_ok=True)
+
+    monkeypatch.setattr(carrier_tracker, "SEED_FILE", seed_path)
+    monkeypatch.setattr(carrier_tracker, "CACHE_FILE", cache_path)
+    monkeypatch.delenv("SHADOWBROKER_CARRIER_FRESHNESS_DAYS", raising=False)
+
+    # Reset module-level mutable state.
+    carrier_tracker._carrier_positions.clear()
+    carrier_tracker._cached_gdelt_articles.clear()
+    carrier_tracker._last_gdelt_fetch_at = 0.0
+
+    yield carrier_tracker
+
+    # Clean up so subsequent tests start fresh.
+    carrier_tracker._carrier_positions.clear()
+    carrier_tracker._cached_gdelt_articles.clear()
+
+
+def _write_seed(path: Path, hull: str = "CVN-78", **overrides) -> None:
+    payload = {
+        "_meta": {
+            "as_of": "2026-03-09",
+            "source": "USNI News Fleet & Marine Tracker",
+            "source_url": "https://news.usni.org/...",
+            "note": "test",
+        },
+        "carriers": {
+            hull: {
+                "lat": 18.0,
+                "lng": 39.5,
+                "heading": 0,
+                "desc": "Red Sea — Operation Epic Fury (USNI Mar 9)",
+                "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+                "source_url": "https://news.usni.org/category/fleet-tracker",
+                "position_source_at": "2026-03-09T00:00:00Z",
+                "position_confidence": "seed",
+                **overrides,
+            }
+        },
+    }
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+
+# ---------------------------------------------------------------------------
+# #244 — first-run seed bootstrap, never re-seeds after that
+# ---------------------------------------------------------------------------
+
+
+class TestSeedBootstrap:
+    def test_first_ever_startup_bootstraps_from_seed(self, fresh_tracker, tmp_path):
+        _write_seed(fresh_tracker.SEED_FILE)
+        # No cache exists yet.
+        assert not fresh_tracker.CACHE_FILE.exists()
+
+        positions = fresh_tracker._bootstrap_cache_if_missing()
+
+        # The seed entry made it into the cache.
+        assert "CVN-78" in positions
+        assert positions["CVN-78"]["lat"] == 18.0
+        assert positions["CVN-78"]["position_confidence"] == "seed"
+        # And the cache file is now on disk so subsequent runs skip the seed.
+        assert fresh_tracker.CACHE_FILE.exists()
+
+    def test_subsequent_startup_ignores_seed(self, fresh_tracker, tmp_path):
+        # Pre-seed a different position into the cache; the seed file says Red Sea.
+        cache_data = {
+            "CVN-78": {
+                "lat": 25.0,
+                "lng": 55.0,
+                "heading": 0,
+                "desc": "Persian Gulf — operator-observed",
+                "source": "Operator log",
+                "source_url": "",
+                "position_source_at": "2026-04-15T12:00:00Z",
+                "position_confidence": "recent",
+            }
+        }
+        fresh_tracker.CACHE_FILE.write_text(json.dumps(cache_data))
+        _write_seed(fresh_tracker.SEED_FILE)  # seed is present but should NOT be used
+
+        positions = fresh_tracker._bootstrap_cache_if_missing()
+
+        assert positions["CVN-78"]["lat"] == 25.0
+        assert positions["CVN-78"]["desc"] == "Persian Gulf — operator-observed"
+
+    def test_no_seed_no_cache_falls_back_to_homeport(self, fresh_tracker):
+        # Neither seed nor cache. Must fall back to homeport defaults
+        # (carrier never disappears).
+        assert not fresh_tracker.SEED_FILE.exists()
+        assert not fresh_tracker.CACHE_FILE.exists()
+
+        positions = fresh_tracker._bootstrap_cache_if_missing()
+
+        # Every registered carrier has SOMETHING.
+        assert set(positions.keys()) == set(fresh_tracker.CARRIER_REGISTRY.keys())
+        # All entries are labelled as homeport defaults.
+        for hull, entry in positions.items():
+            assert entry["position_confidence"] == "homeport_default"
+            registry = fresh_tracker.CARRIER_REGISTRY[hull]
+            assert entry["lat"] == registry["homeport_lat"]
+            assert entry["lng"] == registry["homeport_lng"]
+
+
+# ---------------------------------------------------------------------------
+# #244 — no editorial fallbacks live in the registry
+# ---------------------------------------------------------------------------
+
+
+class TestRegistryShape:
+    def test_registry_has_no_dated_fallback_fields(self, fresh_tracker):
+        """The Mar 9 editorial coordinates are gone from the registry.
+        They live only in the seed file."""
+        forbidden = {"fallback_lat", "fallback_lng", "fallback_heading", "fallback_desc"}
+        for hull, entry in fresh_tracker.CARRIER_REGISTRY.items():
+            offending = forbidden & set(entry.keys())
+            assert not offending, f"{hull} still has dated registry fields: {offending}"
+
+    def test_registry_keeps_homeport_for_every_hull(self, fresh_tracker):
+        for hull, entry in fresh_tracker.CARRIER_REGISTRY.items():
+            assert "homeport_lat" in entry, f"{hull} missing homeport_lat"
+            assert "homeport_lng" in entry, f"{hull} missing homeport_lng"
+            assert "name" in entry
+            assert "wiki" in entry
+
+
+# ---------------------------------------------------------------------------
+# #246 — freshness labelling, NOT eviction
+# ---------------------------------------------------------------------------
+
+
+class TestFreshnessLabelling:
+    def test_recent_observation_labels_recent(self, fresh_tracker):
+        now = datetime(2026, 6, 1, tzinfo=timezone.utc)
+        entry = {
+            "lat": 25.0,
+            "lng": 55.0,
+            "position_source_at": (now - timedelta(days=3)).isoformat(),
+        }
+        assert fresh_tracker._compute_position_confidence(entry, now=now) == "recent"
+
+    def test_aged_observation_flips_to_stale(self, fresh_tracker):
+        now = datetime(2026, 6, 1, tzinfo=timezone.utc)
+        entry = {
+            "lat": 25.0,
+            "lng": 55.0,
+            "position_source_at": (now - timedelta(days=30)).isoformat(),
+        }
+        assert fresh_tracker._compute_position_confidence(entry, now=now) == "stale"
+
+    def test_seed_label_is_preserved_explicitly(self, fresh_tracker):
+        now = datetime(2026, 6, 1, tzinfo=timezone.utc)
+        entry = {
+            "lat": 18.0,
+            "lng": 39.5,
+            "position_source_at": "2026-03-09T00:00:00Z",
+            "position_confidence": "seed",
+        }
+        # Even though the source is months old, the explicit "seed" label wins
+        # so the UI can render the seed-specific badge instead of generic "stale".
+        assert fresh_tracker._compute_position_confidence(entry, now=now) == "seed"
+
+    def test_homeport_default_label_is_preserved(self, fresh_tracker):
+        now = datetime(2026, 6, 1, tzinfo=timezone.utc)
+        entry = {
+            "lat": 36.95,
+            "lng": -76.32,
+            "position_source_at": now.isoformat(),
+            "position_confidence": "homeport_default",
+        }
+        assert fresh_tracker._compute_position_confidence(entry, now=now) == "homeport_default"
+
+    def test_freshness_window_is_env_configurable(self, fresh_tracker, monkeypatch):
+        now = datetime(2026, 6, 1, tzinfo=timezone.utc)
+        entry = {
+            "lat": 25.0,
+            "lng": 55.0,
+            "position_source_at": (now - timedelta(days=20)).isoformat(),
+        }
+        # Default window = 14 days → 20-day-old entry is stale.
+        assert fresh_tracker._compute_position_confidence(entry, now=now) == "stale"
+        # Stretch to 30 days → same entry is now "recent".
+        monkeypatch.setenv("SHADOWBROKER_CARRIER_FRESHNESS_DAYS", "30")
+        assert fresh_tracker._compute_position_confidence(entry, now=now) == "recent"
+
+    def test_aged_cache_entry_keeps_its_position_never_reverts(self, fresh_tracker):
+        """The core regression test for the user's intent: a year-old
+        cache entry must NOT be replaced with the seed or homeport.
+        The PHYSICAL position the user sees is the last one observed;
+        only the freshness LABEL changes."""
+        a_year_ago = (datetime.now(timezone.utc) - timedelta(days=365)).isoformat()
+        cache_data = {
+            "CVN-78": {
+                "lat": 25.0,
+                "lng": 55.0,
+                "heading": 0,
+                "desc": "Persian Gulf",
+                "source": "GDELT News API",
+                "source_url": "https://news.example/...",
+                "position_source_at": a_year_ago,
+                "position_confidence": "recent",  # was recent when written
+            }
+        }
+        fresh_tracker.CACHE_FILE.write_text(json.dumps(cache_data))
+
+        positions = fresh_tracker._bootstrap_cache_if_missing()
+        enriched = fresh_tracker._enrich_for_rendering("CVN-78", positions["CVN-78"])
+
+        # The position is preserved exactly.
+        assert enriched["lat"] == 25.0
+        assert enriched["lng"] == 55.0
+        # But the live label has flipped to stale.
+        assert enriched["position_confidence"] == "stale"
+        assert enriched["is_fallback"] is True
+
+
+# ---------------------------------------------------------------------------
+# #245 — approximate confidence for region-centroid positions
+# ---------------------------------------------------------------------------
+
+
+class TestApproximateConfidenceForNewsDerivedPositions:
+    def test_news_parsing_stamps_approximate_confidence(self, fresh_tracker):
+        articles = [
+            {
+                "title": "USS Ford carrier deployed in Mediterranean for joint exercise",
+                "url": "https://news.example/ford-mediterranean",
+                "seendate": "20260415120000",
+            }
+        ]
+        updates = fresh_tracker._parse_carrier_positions_from_news(articles)
+        assert "CVN-78" in updates
+        entry = updates["CVN-78"]
+        assert entry["position_confidence"] == "approximate"
+        # And the source_at is the article's seen date, not now().
+        assert entry["position_source_at"].startswith("2026-04-15")
+
+    def test_gdelt_seendate_parser_handles_well_formed_input(self, fresh_tracker):
+        iso = fresh_tracker._gdelt_seendate_to_iso("20260415120000")
+        assert iso is not None
+        assert iso.startswith("2026-04-15T12:00:00")
+
+    def test_gdelt_seendate_parser_returns_none_on_garbage(self, fresh_tracker):
+        assert fresh_tracker._gdelt_seendate_to_iso("") is None
+        assert fresh_tracker._gdelt_seendate_to_iso("not-a-date") is None
+        assert fresh_tracker._gdelt_seendate_to_iso("2026") is None
+
+
+# ---------------------------------------------------------------------------
+# Full enrichment → public API shape
+# ---------------------------------------------------------------------------
+
+
+class TestEnrichForRendering:
+    def test_seed_entry_produces_expected_public_fields(self, fresh_tracker):
+        seed_entry = {
+            "lat": 18.0,
+            "lng": 39.5,
+            "heading": 0,
+            "desc": "Red Sea (USNI Mar 9)",
+            "source": "USNI News Fleet & Marine Tracker (seed, as of 2026-03-09)",
+            "source_url": "https://news.usni.org/category/fleet-tracker",
+            "position_source_at": "2026-03-09T00:00:00Z",
+            "position_confidence": "seed",
+        }
+        enriched = fresh_tracker._enrich_for_rendering("CVN-78", seed_entry)
+        # Existing UI fields preserved.
+        assert enriched["lat"] == 18.0
+        assert enriched["lng"] == 39.5
+        assert enriched["source"].startswith("USNI")
+        assert enriched["last_osint_update"] == "2026-03-09T00:00:00Z"
+        # New audit-required fields.
+        assert enriched["position_confidence"] == "seed"
+        assert enriched["position_source_at"] == "2026-03-09T00:00:00Z"
+        assert enriched["is_fallback"] is True
+
+    def test_recent_observation_is_not_fallback(self, fresh_tracker):
+        now = datetime.now(timezone.utc)
+        recent_entry = {
+            "lat": 25.0,
+            "lng": 55.0,
+            "heading": 0,
+            "desc": "Persian Gulf",
+            "source": "GDELT News API",
+            "source_url": "https://news.example/...",
+            "position_source_at": (now - timedelta(days=2)).isoformat(),
+            "position_confidence": "approximate",
+        }
+        enriched = fresh_tracker._enrich_for_rendering("CVN-78", recent_entry, now=now)
+        assert enriched["position_confidence"] == "approximate"
+        # Approximate (from a recent headline) is honest precision, but the UI
+        # treats it as live data — is_fallback only flips True for explicit
+        # fallback categories (seed / stale / homeport_default).
+        assert enriched["is_fallback"] is False
+
+
+# ---------------------------------------------------------------------------
+# Regression: existing frontend fields are preserved
+# ---------------------------------------------------------------------------
+
+
+class TestPublicResponseShapeBackwardCompat:
+    """The frontend ShipPopup expects `estimated`, `source`, `source_url`,
+    `last_osint_update`. The new fields are additive and existing fields
+    keep their meaning so the UI does not need updating to keep working."""
+
+    def test_get_carrier_positions_preserves_existing_keys(self, fresh_tracker):
+        _write_seed(fresh_tracker.SEED_FILE)
+        fresh_tracker._bootstrap_cache_if_missing()
+        with fresh_tracker._positions_lock:
+            fresh_tracker._carrier_positions.update(
+                {
+                    "CVN-78": {
+                        "lat": 18.0,
+                        "lng": 39.5,
+                        "heading": 0,
+                        "desc": "Red Sea (seed)",
+                        "source": "Seed",
+                        "source_url": "",
+                        "position_source_at": "2026-03-09T00:00:00Z",
+                        "position_confidence": "seed",
+                    }
+                }
+            )
+
+        out = fresh_tracker.get_carrier_positions()
+        assert len(out) == 1
+        c = out[0]
+        # Old fields the frontend uses.
+        for key in (
+            "name",
+            "type",
+            "lat",
+            "lng",
+            "country",
+            "desc",
+            "wiki",
+            "estimated",
+            "source",
+            "source_url",
+            "last_osint_update",
+        ):
+            assert key in c, f"missing legacy field {key!r}"
+        # New fields.
+        for key in ("position_confidence", "position_source_at", "is_fallback"):
+            assert key in c, f"missing audit-required field {key!r}"
+        assert c["type"] == "carrier"
+        assert c["estimated"] is True

backend/tests/test_docker_bridge_hostname_trust.py

@@ -0,0 +1,196 @@
+"""Issue #250 (tg12): Docker bridge local-operator trust must be bound to
+the frontend container's hostname, not the entire 172.16.0.0/12 range.
+
+Previous behavior trusted ANY private-RFC1918 source IP on the bridge
+when ``SHADOWBROKER_TRUST_DOCKER_BRIDGE_LOCAL_OPERATOR=1``. On a shared
+Docker host this granted local-operator privileges to any other
+container that could route to the backend's bridge — far broader than
+intended.
+
+The fix narrows trust to source IPs that forward-resolve from one of the
+configured frontend container hostnames (default: the compose service
+name ``frontend`` plus the explicit ``container_name``
+``shadowbroker-frontend``). Operators with renamed containers can list
+the new names in ``SHADOWBROKER_TRUSTED_FRONTEND_HOSTS``.
+
+These tests exercise the resolution helpers directly so that we don't
+need a live Docker daemon to validate the contract.
+"""
+import socket
+from unittest.mock import patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# _trusted_bridge_frontend_hostnames — env parsing
+# ---------------------------------------------------------------------------
+
+class TestTrustedHostnameParsing:
+    def _fn(self):
+        from auth import _trusted_bridge_frontend_hostnames
+        return _trusted_bridge_frontend_hostnames
+
+    def test_default_covers_compose_service_and_container_name(self):
+        with patch.dict("os.environ", {}, clear=False):
+            # Make sure the env var is not set so we exercise the default.
+            import os
+            os.environ.pop("SHADOWBROKER_TRUSTED_FRONTEND_HOSTS", None)
+            assert self._fn()() == ["frontend", "shadowbroker-frontend"]
+
+    def test_custom_list_via_env(self):
+        with patch.dict(
+            "os.environ",
+            {"SHADOWBROKER_TRUSTED_FRONTEND_HOSTS": "my-ui,alt-frontend"},
+        ):
+            assert self._fn()() == ["my-ui", "alt-frontend"]
+
+    def test_whitespace_trimmed(self):
+        with patch.dict(
+            "os.environ",
+            {"SHADOWBROKER_TRUSTED_FRONTEND_HOSTS": "  my-ui , alt-frontend  "},
+        ):
+            assert self._fn()() == ["my-ui", "alt-frontend"]
+
+    def test_empty_env_falls_back_to_default(self):
+        # An empty string still falls back to the bundled defaults so a
+        # misconfigured env var doesn't silently dismantle bridge trust.
+        with patch.dict(
+            "os.environ",
+            {"SHADOWBROKER_TRUSTED_FRONTEND_HOSTS": ""},
+        ):
+            # Per docs: empty string sets the env var to "" so os.environ.get
+            # returns "" — that string is parsed and yields []. We assert
+            # that empty parse yields [] (caller fail-closes from there).
+            assert self._fn()() == []
+
+
+# ---------------------------------------------------------------------------
+# _resolve_trusted_bridge_ips — DNS resolution with cache + fail-closed
+# ---------------------------------------------------------------------------
+
+class TestResolveTrustedBridgeIps:
+    def setup_method(self):
+        # Reset the module-level cache before each test so prior tests
+        # don't bleed state across cases.
+        from auth import _DOCKER_BRIDGE_TRUST_CACHE
+        _DOCKER_BRIDGE_TRUST_CACHE["ips"] = frozenset()
+        _DOCKER_BRIDGE_TRUST_CACHE["expires"] = 0.0
+
+    def test_resolves_configured_hostnames(self):
+        from auth import _resolve_trusted_bridge_ips
+
+        def fake_gethostbyname_ex(host):
+            mapping = {
+                "frontend": ("frontend", [], ["172.18.0.3"]),
+                "shadowbroker-frontend": ("shadowbroker-frontend", [], ["172.18.0.3", "172.18.0.4"]),
+            }
+            if host not in mapping:
+                raise socket.gaierror("no such host")
+            return mapping[host]
+
+        with patch("socket.gethostbyname_ex", side_effect=fake_gethostbyname_ex):
+            ips = _resolve_trusted_bridge_ips()
+        assert ips == frozenset({"172.18.0.3", "172.18.0.4"})
+
+    def test_fail_closed_when_dns_returns_nothing(self):
+        from auth import _resolve_trusted_bridge_ips
+
+        def always_fail(host):
+            raise socket.gaierror("no resolver")
+
+        with patch("socket.gethostbyname_ex", side_effect=always_fail):
+            ips = _resolve_trusted_bridge_ips()
+        assert ips == frozenset()
+
+    def test_partial_resolution_is_kept(self):
+        """If one hostname resolves and another fails, we keep the
+        successful one rather than discarding the whole set."""
+        from auth import _resolve_trusted_bridge_ips
+
+        def partial(host):
+            if host == "frontend":
+                return ("frontend", [], ["172.18.0.3"])
+            raise socket.gaierror("missing")
+
+        with patch("socket.gethostbyname_ex", side_effect=partial):
+            ips = _resolve_trusted_bridge_ips()
+        assert ips == frozenset({"172.18.0.3"})
+
+    def test_cache_short_circuits_repeated_dns_calls(self):
+        from auth import _resolve_trusted_bridge_ips
+
+        call_count = {"n": 0}
+
+        def counting(host):
+            call_count["n"] += 1
+            return ("frontend", [], ["172.18.0.3"])
+
+        with patch("socket.gethostbyname_ex", side_effect=counting):
+            _resolve_trusted_bridge_ips()
+            calls_after_first = call_count["n"]
+            _resolve_trusted_bridge_ips()
+            _resolve_trusted_bridge_ips()
+        # Second + third calls hit the cache, not the DNS stub.
+        assert call_count["n"] == calls_after_first
+
+    def test_cache_expires(self):
+        from auth import _resolve_trusted_bridge_ips, _DOCKER_BRIDGE_TRUST_CACHE
+
+        with patch("socket.gethostbyname_ex", return_value=("frontend", [], ["172.18.0.3"])):
+            _resolve_trusted_bridge_ips()
+        # Force expiry.
+        _DOCKER_BRIDGE_TRUST_CACHE["expires"] = 0.0
+        with patch("socket.gethostbyname_ex", return_value=("frontend", [], ["172.18.0.9"])) as stub:
+            ips = _resolve_trusted_bridge_ips()
+            assert stub.called
+        assert "172.18.0.9" in ips
+
+
+# ---------------------------------------------------------------------------
+# _is_docker_bridge_host — composite of the helpers above
+# ---------------------------------------------------------------------------
+
+class TestIsDockerBridgeHost:
+    def setup_method(self):
+        from auth import _DOCKER_BRIDGE_TRUST_CACHE
+        _DOCKER_BRIDGE_TRUST_CACHE["ips"] = frozenset()
+        _DOCKER_BRIDGE_TRUST_CACHE["expires"] = 0.0
+
+    def test_trusts_resolved_frontend_ip(self):
+        from auth import _is_docker_bridge_host
+
+        with patch("auth._resolve_trusted_bridge_ips", return_value=frozenset({"172.18.0.3"})):
+            assert _is_docker_bridge_host("172.18.0.3") is True
+
+    def test_rejects_arbitrary_bridge_ip(self):
+        """A rogue container on the same bridge but at a different IP
+        must NOT be trusted, even though it falls in 172.16.0.0/12."""
+        from auth import _is_docker_bridge_host
+
+        with patch("auth._resolve_trusted_bridge_ips", return_value=frozenset({"172.18.0.3"})):
+            assert _is_docker_bridge_host("172.18.0.99") is False
+
+    def test_rejects_public_ip_without_dns_work(self):
+        """Public IPs skip DNS resolution entirely (perf + safety)."""
+        from auth import _is_docker_bridge_host
+
+        with patch("auth._resolve_trusted_bridge_ips") as stub:
+            assert _is_docker_bridge_host("8.8.8.8") is False
+            stub.assert_not_called()
+
+    def test_rejects_non_ip_input(self):
+        from auth import _is_docker_bridge_host
+
+        assert _is_docker_bridge_host("") is False
+        assert _is_docker_bridge_host("not-an-ip") is False
+        assert _is_docker_bridge_host("frontend") is False
+
+    def test_fails_closed_when_dns_returns_empty(self):
+        """If Docker DNS can't resolve any frontend hostname, the bridge
+        is not trusted — even for IPs that would have been trusted under
+        the old 172.16.0.0/12 blanket policy."""
+        from auth import _is_docker_bridge_host
+
+        with patch("auth._resolve_trusted_bridge_ips", return_value=frozenset()):
+            assert _is_docker_bridge_host("172.18.0.3") is False

backend/tests/test_oracle_resolve_auth_gate.py

@@ -0,0 +1,160 @@
+"""Issues #240 & #241 (tg12): oracle market/stake resolution endpoints
+must require admin authentication.
+
+Before the fix, ``POST /api/mesh/oracle/resolve`` and
+``POST /api/mesh/oracle/resolve-stakes`` were decorated with
+``@mesh_write_exempt(MeshWriteExemption.ADMIN_CONTROL)``. That decorator
+only tags the route as not requiring a mesh signed-write envelope; it
+does NOT enforce authorization. The rate limiter (5/minute) was the
+only real gate, which is wrong for control-plane state mutations.
+
+The fix adds ``dependencies=[Depends(require_admin)]`` to both routes.
+These tests prove:
+
+- Anonymous callers receive 403.
+- A request bearing the configured admin key passes the auth gate.
+- The underlying ledger mutator is not invoked on a 403.
+"""
+from __future__ import annotations
+
+from unittest.mock import patch, MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+_ADMIN_KEY = "test-admin-key-for-oracle-resolve-fixture-32+"
+
+
+@pytest.fixture
+def client():
+    """TestClient with the private-lane transport middleware short-circuited.
+
+    The ``enforce_high_privacy_mesh`` middleware in ``main.py`` returns
+    HTTP 202 ("preparing private lane") for ``/api/mesh/*`` requests
+    when the Wormhole supervisor is not yet at the required transport
+    tier. In tests that's always — Wormhole is not running. Patching
+    ``_minimum_transport_tier`` to return None disables the tier check
+    for the duration of the test, letting the request reach the route
+    (and therefore reach the ``Depends(require_admin)`` we are testing).
+    """
+    import main
+    with patch("main._minimum_transport_tier", return_value=None):
+        yield TestClient(main.app, raise_server_exceptions=False)
+
+
+@pytest.fixture
+def mock_ledger():
+    """Replace oracle_ledger methods so tests don't mutate persistent state.
+
+    The handler does ``from services.mesh.mesh_oracle import oracle_ledger``
+    at call time, so we patch the module attribute.
+    """
+    fake = MagicMock()
+    fake.resolve_market.return_value = (0, 0)
+    fake.resolve_market_stakes.return_value = {"winners": 0, "losers": 0}
+    fake.resolve_expired_stakes.return_value = []
+    with patch("services.mesh.mesh_oracle.oracle_ledger", fake):
+        yield fake
+
+
+# ---------------------------------------------------------------------------
+# /api/mesh/oracle/resolve — issue #240
+# ---------------------------------------------------------------------------
+
+
+class TestOracleResolveAuthGate:
+    def test_anonymous_caller_is_rejected(self, client, mock_ledger):
+        with patch("auth._current_admin_key", return_value=_ADMIN_KEY):
+            r = client.post(
+                "/api/mesh/oracle/resolve",
+                json={"market_title": "test-market", "outcome": "Yes"},
+            )
+        assert r.status_code == 403
+        # Critically: the ledger mutator must NOT have been called on a 403.
+        assert mock_ledger.resolve_market.call_count == 0
+        assert mock_ledger.resolve_market_stakes.call_count == 0
+
+    def test_wrong_admin_key_rejected(self, client, mock_ledger):
+        with patch("auth._current_admin_key", return_value=_ADMIN_KEY):
+            r = client.post(
+                "/api/mesh/oracle/resolve",
+                headers={"X-Admin-Key": "this-key-is-wrong"},
+                json={"market_title": "test-market", "outcome": "Yes"},
+            )
+        assert r.status_code == 403
+        assert mock_ledger.resolve_market.call_count == 0
+
+    def test_valid_admin_key_passes_auth_gate(self, client, mock_ledger):
+        with patch("auth._current_admin_key", return_value=_ADMIN_KEY):
+            r = client.post(
+                "/api/mesh/oracle/resolve",
+                headers={"X-Admin-Key": _ADMIN_KEY},
+                json={"market_title": "test-market", "outcome": "Yes"},
+            )
+        # The auth gate let us through. The handler ran and called the
+        # (mocked) ledger.
+        assert r.status_code == 200
+        assert mock_ledger.resolve_market.call_count == 1
+        assert mock_ledger.resolve_market.call_args[0] == ("test-market", "Yes")
+
+    def test_admin_key_unset_blocks_in_production_posture(self, client, mock_ledger):
+        """When ADMIN_KEY env is not configured at all and we're not in
+        debug, the endpoint must still refuse — never silently accept."""
+        with (
+            patch("auth._current_admin_key", return_value=""),
+            patch("auth._allow_insecure_admin", return_value=False),
+            patch("auth._debug_mode_enabled", return_value=False),
+            patch("auth._scoped_admin_tokens", return_value={}),
+        ):
+            r = client.post(
+                "/api/mesh/oracle/resolve",
+                json={"market_title": "test-market", "outcome": "Yes"},
+            )
+        assert r.status_code == 403
+        assert mock_ledger.resolve_market.call_count == 0
+
+
+# ---------------------------------------------------------------------------
+# /api/mesh/oracle/resolve-stakes — issue #241
+# ---------------------------------------------------------------------------
+
+
+class TestOracleResolveStakesAuthGate:
+    def test_anonymous_caller_is_rejected(self, client, mock_ledger):
+        with patch("auth._current_admin_key", return_value=_ADMIN_KEY):
+            r = client.post("/api/mesh/oracle/resolve-stakes")
+        assert r.status_code == 403
+        assert mock_ledger.resolve_expired_stakes.call_count == 0
+
+    def test_wrong_admin_key_rejected(self, client, mock_ledger):
+        with patch("auth._current_admin_key", return_value=_ADMIN_KEY):
+            r = client.post(
+                "/api/mesh/oracle/resolve-stakes",
+                headers={"X-Admin-Key": "nope"},
+            )
+        assert r.status_code == 403
+        assert mock_ledger.resolve_expired_stakes.call_count == 0
+
+    def test_valid_admin_key_passes_auth_gate(self, client, mock_ledger):
+        with patch("auth._current_admin_key", return_value=_ADMIN_KEY):
+            r = client.post(
+                "/api/mesh/oracle/resolve-stakes",
+                headers={"X-Admin-Key": _ADMIN_KEY},
+            )
+        assert r.status_code == 200
+        assert mock_ledger.resolve_expired_stakes.call_count == 1
+        body = r.json()
+        assert body["ok"] is True
+        assert body["count"] == 0
+
+    def test_admin_key_unset_blocks_in_production_posture(self, client, mock_ledger):
+        with (
+            patch("auth._current_admin_key", return_value=""),
+            patch("auth._allow_insecure_admin", return_value=False),
+            patch("auth._debug_mode_enabled", return_value=False),
+            patch("auth._scoped_admin_tokens", return_value={}),
+        ):
+            r = client.post("/api/mesh/oracle/resolve-stakes")
+        assert r.status_code == 403
+        assert mock_ledger.resolve_expired_stakes.call_count == 0

backend/tests/test_p0_security.py

@@ -87,16 +87,32 @@ class TestRequireLocalOperator:
         assert self._call_with_host("172.16.0.5") == 403
 
     def test_docker_bridge_blocked_without_compose_opt_in(self):
+        # Even if DNS would resolve the frontend hostname to this IP,
+        # the env opt-in is required.
         with patch.dict("os.environ", {"SHADOWBROKER_TRUST_DOCKER_BRIDGE_LOCAL_OPERATOR": ""}):
-            assert self._call_with_host("172.18.0.3") == 403
+            with patch("auth._resolve_trusted_bridge_ips", return_value=frozenset({"172.18.0.3"})):
+                assert self._call_with_host("172.18.0.3") == 403
 
     def test_docker_bridge_passes_with_compose_opt_in(self):
+        # Issue #250: opt-in alone is no longer sufficient — the source IP
+        # must also reverse-match a trusted frontend container hostname.
+        # Here we simulate Docker DNS resolving "frontend" to 172.18.0.3.
         with patch.dict("os.environ", {"SHADOWBROKER_TRUST_DOCKER_BRIDGE_LOCAL_OPERATOR": "1"}):
-            assert self._call_with_host("172.18.0.3") == 200
+            with patch("auth._resolve_trusted_bridge_ips", return_value=frozenset({"172.18.0.3"})):
+                assert self._call_with_host("172.18.0.3") == 200
+
+    def test_unknown_bridge_ip_blocked_even_with_compose_opt_in(self):
+        # Issue #250 core regression: a rogue container on the same bridge
+        # whose IP is NOT in the resolved frontend hostname set must NOT
+        # be trusted, even when the bridge opt-in flag is on.
+        with patch.dict("os.environ", {"SHADOWBROKER_TRUST_DOCKER_BRIDGE_LOCAL_OPERATOR": "1"}):
+            with patch("auth._resolve_trusted_bridge_ips", return_value=frozenset({"172.18.0.3"})):
+                assert self._call_with_host("172.18.0.99") == 403
 
     def test_lan_ip_still_blocked_with_compose_opt_in(self):
         with patch.dict("os.environ", {"SHADOWBROKER_TRUST_DOCKER_BRIDGE_LOCAL_OPERATOR": "1"}):
-            assert self._call_with_host("192.168.1.100") == 403
+            with patch("auth._resolve_trusted_bridge_ips", return_value=frozenset({"172.18.0.3"})):
+                assert self._call_with_host("192.168.1.100") == 403
 
     def test_rfc1918_192168_blocked_without_key(self):
         assert self._call_with_host("192.168.1.100") == 403

backend/tests/test_per_peer_secret_resolver.py

@@ -0,0 +1,366 @@
+"""Issue #256 (tg12): per-peer HMAC secrets must defeat cross-peer
+impersonation.
+
+Before the fix, ALL peer-push HMACs were derived from the single
+fleet-shared ``MESH_PEER_PUSH_SECRET``. The receiver could only prove
+"this request was signed by someone who knows the fleet secret" — not
+which peer signed it. Any peer that knew the secret could compute the
+expected HMAC for any other peer's URL and impersonate that peer.
+
+The fix introduces ``MESH_PEER_SECRETS``, a per-peer URL-to-secret map.
+When a peer URL appears there:
+
+- Only the listed per-peer secret is accepted for that URL.
+- The global ``MESH_PEER_PUSH_SECRET`` is ignored for that specific URL.
+- A peer that knows only the global secret (or a different peer's
+  per-peer secret) cannot forge a request claiming to be that peer.
+
+When a peer URL is NOT listed (the common case for single-peer installs
+and for migration windows), the resolver falls back to the global
+secret — preserving existing behavior with zero operator action.
+
+These tests exercise ``resolve_peer_key_for_url`` directly so we cover
+the security contract without spinning up a full mesh node.
+"""
+from __future__ import annotations
+
+import hashlib
+import hmac
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# _lookup_per_peer_secret — env parsing
+# ---------------------------------------------------------------------------
+
+
+class TestLookupPerPeerSecret:
+    def setup_method(self):
+        # Invalidate the parser cache so each test sees its own env state.
+        from services.mesh import mesh_crypto
+
+        mesh_crypto._PEER_SECRETS_CACHE = {}
+        mesh_crypto._PEER_SECRETS_CACHE_RAW = ""
+
+    def test_returns_empty_when_env_unset(self, monkeypatch):
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        monkeypatch.delenv("MESH_PEER_SECRETS", raising=False)
+        assert _lookup_per_peer_secret("https://peer.example") == ""
+
+    def test_returns_empty_when_env_blank(self, monkeypatch):
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        monkeypatch.setenv("MESH_PEER_SECRETS", "")
+        assert _lookup_per_peer_secret("https://peer.example") == ""
+
+    def test_returns_per_peer_secret_for_listed_url(self, monkeypatch):
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://peer-a.example=secretA,https://peer-b.example=secretB",
+        )
+        assert _lookup_per_peer_secret("https://peer-a.example") == "secretA"
+        assert _lookup_per_peer_secret("https://peer-b.example") == "secretB"
+
+    def test_returns_empty_for_url_not_listed(self, monkeypatch):
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://peer-a.example=secretA",
+        )
+        assert _lookup_per_peer_secret("https://other.example") == ""
+
+    def test_url_is_normalized_before_lookup(self, monkeypatch):
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        # Configure with a trailing slash + uppercase host. Lookup with
+        # plain lowercase host. Both should normalize to the same key.
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://Peer-A.Example/=secretA",
+        )
+        assert _lookup_per_peer_secret("https://peer-a.example") == "secretA"
+
+    def test_whitespace_around_entries_is_stripped(self, monkeypatch):
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "  https://peer-a.example = secretA , https://peer-b.example=secretB  ",
+        )
+        assert _lookup_per_peer_secret("https://peer-a.example") == "secretA"
+        assert _lookup_per_peer_secret("https://peer-b.example") == "secretB"
+
+    def test_malformed_entries_are_skipped_not_raised(self, monkeypatch):
+        """A garbled MESH_PEER_SECRETS value must NOT crash the resolver.
+        Bad entries are silently dropped; well-formed entries still work.
+        This is the "fail-forward, not loud" rule — a typo in operator
+        config should not take the whole backend down."""
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "no_equals_sign,=missing_url,https://no.secret=,https://good.example=secretGood",
+        )
+        assert _lookup_per_peer_secret("https://good.example") == "secretGood"
+        # The malformed ones produce no entry (and don't poison the cache).
+        assert _lookup_per_peer_secret("https://no.secret") == ""
+
+    def test_cache_invalidates_on_env_change(self, monkeypatch):
+        """A test (or operator) updating MESH_PEER_SECRETS must see the
+        new value immediately — no process restart required."""
+        from services.mesh.mesh_crypto import _lookup_per_peer_secret
+
+        monkeypatch.setenv("MESH_PEER_SECRETS", "https://a.example=first")
+        assert _lookup_per_peer_secret("https://a.example") == "first"
+        monkeypatch.setenv("MESH_PEER_SECRETS", "https://a.example=second")
+        assert _lookup_per_peer_secret("https://a.example") == "second"
+
+
+# ---------------------------------------------------------------------------
+# resolve_peer_key_for_url — precedence + fallback
+# ---------------------------------------------------------------------------
+
+
+class TestResolvePeerKeyForUrl:
+    def setup_method(self):
+        from services.mesh import mesh_crypto
+
+        mesh_crypto._PEER_SECRETS_CACHE = {}
+        mesh_crypto._PEER_SECRETS_CACHE_RAW = ""
+
+    def _fake_settings(self, global_secret: str):
+        from unittest.mock import MagicMock
+
+        s = MagicMock()
+        s.MESH_PEER_PUSH_SECRET = global_secret
+        return s
+
+    def test_falls_back_to_global_when_no_per_peer_entry(self, monkeypatch):
+        """Single-peer installs: MESH_PEER_SECRETS empty, MESH_PEER_PUSH_SECRET
+        set — must keep working as before."""
+        from services.mesh.mesh_crypto import (
+            resolve_peer_key_for_url,
+            _derive_peer_key,
+        )
+
+        monkeypatch.delenv("MESH_PEER_SECRETS", raising=False)
+        with monkeypatch.context() as m:
+            m.setattr(
+                "services.config.get_settings",
+                lambda: self._fake_settings("global-secret"),
+            )
+            key = resolve_peer_key_for_url("https://peer.example")
+            expected = _derive_peer_key("global-secret", "https://peer.example")
+        assert key == expected
+        assert len(key) == 32  # SHA-256 output
+
+    def test_per_peer_secret_takes_precedence_over_global(self, monkeypatch):
+        from services.mesh.mesh_crypto import (
+            resolve_peer_key_for_url,
+            _derive_peer_key,
+        )
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://peer-a.example=per-peer-a-secret",
+        )
+        with monkeypatch.context() as m:
+            m.setattr(
+                "services.config.get_settings",
+                lambda: self._fake_settings("global-secret"),
+            )
+            key = resolve_peer_key_for_url("https://peer-a.example")
+            expected_per_peer = _derive_peer_key(
+                "per-peer-a-secret", "https://peer-a.example"
+            )
+            expected_global = _derive_peer_key("global-secret", "https://peer-a.example")
+        assert key == expected_per_peer
+        assert key != expected_global
+
+    def test_unlisted_peer_uses_global_during_migration(self, monkeypatch):
+        """Partial migration: peer A is in MESH_PEER_SECRETS, peer B is
+        not yet. Peer B must keep working under the global secret."""
+        from services.mesh.mesh_crypto import (
+            resolve_peer_key_for_url,
+            _derive_peer_key,
+        )
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://peer-a.example=per-peer-a-secret",
+        )
+        with monkeypatch.context() as m:
+            m.setattr(
+                "services.config.get_settings",
+                lambda: self._fake_settings("global-secret"),
+            )
+            key_a = resolve_peer_key_for_url("https://peer-a.example")
+            key_b = resolve_peer_key_for_url("https://peer-b.example")
+            expected_b = _derive_peer_key("global-secret", "https://peer-b.example")
+        assert key_b == expected_b
+        # Peer A's per-peer key must differ from peer B's global key
+        # (they're keyed by different secrets and different URLs).
+        assert key_a != key_b
+
+    def test_returns_empty_when_no_secret_available(self, monkeypatch):
+        from services.mesh.mesh_crypto import resolve_peer_key_for_url
+
+        monkeypatch.delenv("MESH_PEER_SECRETS", raising=False)
+        with monkeypatch.context() as m:
+            m.setattr(
+                "services.config.get_settings",
+                lambda: self._fake_settings(""),
+            )
+            key = resolve_peer_key_for_url("https://peer.example")
+        assert key == b""
+
+    def test_returns_empty_when_url_is_unparseable(self, monkeypatch):
+        from services.mesh.mesh_crypto import resolve_peer_key_for_url
+
+        with monkeypatch.context() as m:
+            m.setattr(
+                "services.config.get_settings",
+                lambda: self._fake_settings("global-secret"),
+            )
+            assert resolve_peer_key_for_url("") == b""
+            assert resolve_peer_key_for_url("not-a-url") == b""
+            assert resolve_peer_key_for_url(None) == b""
+
+
+# ---------------------------------------------------------------------------
+# The actual #256 attack: peer A cannot impersonate peer B
+# ---------------------------------------------------------------------------
+
+
+class TestCrossPeerImpersonationRefused:
+    """The core regression: when MESH_PEER_SECRETS is configured, a peer
+    that knows ONLY the global secret (or a different peer's per-peer
+    secret) cannot produce a valid HMAC for another peer's URL."""
+
+    def setup_method(self):
+        from services.mesh import mesh_crypto
+
+        mesh_crypto._PEER_SECRETS_CACHE = {}
+        mesh_crypto._PEER_SECRETS_CACHE_RAW = ""
+
+    def _hmac(self, key: bytes, body: bytes) -> str:
+        return hmac.new(key, body, hashlib.sha256).hexdigest()
+
+    def test_peer_a_global_secret_cannot_forge_peer_b_hmac(self, monkeypatch):
+        from services.mesh.mesh_crypto import (
+            resolve_peer_key_for_url,
+            _derive_peer_key,
+        )
+        from unittest.mock import MagicMock
+
+        # Receiver has BOTH the global secret AND a per-peer secret for B.
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://peer-b.example=per-peer-b-secret",
+        )
+        settings = MagicMock()
+        settings.MESH_PEER_PUSH_SECRET = "global-secret"
+        monkeypatch.setattr(
+            "services.config.get_settings", lambda: settings
+        )
+
+        body = b'{"events": [{"id": 1}]}'
+
+        # Attacker (peer A) knows only the global secret. Tries to forge
+        # an HMAC claiming to be peer B.
+        attacker_key = _derive_peer_key("global-secret", "https://peer-b.example")
+        attacker_hmac = self._hmac(attacker_key, body)
+
+        # Receiver derives B's expected key from B's per-peer secret.
+        receiver_key = resolve_peer_key_for_url("https://peer-b.example")
+        expected_hmac = self._hmac(receiver_key, body)
+
+        # The forgery MUST NOT match.
+        assert attacker_hmac != expected_hmac
+
+    def test_peer_a_per_peer_secret_cannot_forge_peer_b_hmac(self, monkeypatch):
+        """Even harder case: peer A has its OWN per-peer secret, but
+        still does not know peer B's per-peer secret, and so cannot
+        forge an HMAC for peer B."""
+        from services.mesh.mesh_crypto import (
+            resolve_peer_key_for_url,
+            _derive_peer_key,
+        )
+        from unittest.mock import MagicMock
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://peer-a.example=secretA,https://peer-b.example=secretB",
+        )
+        settings = MagicMock()
+        settings.MESH_PEER_PUSH_SECRET = ""
+        monkeypatch.setattr(
+            "services.config.get_settings", lambda: settings
+        )
+
+        body = b'{"events": [{"id": 99}]}'
+
+        # Attacker A tries to forge for B using its own secret (secretA).
+        attacker_key = _derive_peer_key("secretA", "https://peer-b.example")
+        attacker_hmac = self._hmac(attacker_key, body)
+
+        receiver_key = resolve_peer_key_for_url("https://peer-b.example")
+        expected_hmac = self._hmac(receiver_key, body)
+
+        assert attacker_hmac != expected_hmac
+
+    def test_legitimate_peer_b_request_verifies(self, monkeypatch):
+        """Positive control: when peer B uses ITS per-peer secret and
+        claims to be itself, the receiver accepts the HMAC."""
+        from services.mesh.mesh_crypto import resolve_peer_key_for_url
+        from unittest.mock import MagicMock
+
+        monkeypatch.setenv(
+            "MESH_PEER_SECRETS",
+            "https://peer-b.example=secretB",
+        )
+        settings = MagicMock()
+        settings.MESH_PEER_PUSH_SECRET = ""
+        monkeypatch.setattr(
+            "services.config.get_settings", lambda: settings
+        )
+
+        body = b'{"events": [{"id": 7}]}'
+
+        # Peer B and the receiver both call resolve_peer_key_for_url.
+        sender_key = resolve_peer_key_for_url("https://peer-b.example")
+        receiver_key = resolve_peer_key_for_url("https://peer-b.example")
+
+        sender_hmac = self._hmac(sender_key, body)
+        expected_hmac = self._hmac(receiver_key, body)
+
+        assert sender_hmac == expected_hmac
+
+    def test_single_peer_install_zero_behavior_change(self, monkeypatch):
+        """The "no UX hostility" guarantee: an install with the global
+        secret set and NO MESH_PEER_SECRETS entries must derive exactly
+        the same key as before this change."""
+        from services.mesh.mesh_crypto import (
+            resolve_peer_key_for_url,
+            _derive_peer_key,
+        )
+        from unittest.mock import MagicMock
+
+        monkeypatch.delenv("MESH_PEER_SECRETS", raising=False)
+        settings = MagicMock()
+        settings.MESH_PEER_PUSH_SECRET = "legacy-global-secret"
+        monkeypatch.setattr(
+            "services.config.get_settings", lambda: settings
+        )
+
+        # The legacy derivation that every prior call site used.
+        legacy_key = _derive_peer_key("legacy-global-secret", "https://peer.example")
+        # The new resolver, with no per-peer entries configured.
+        new_key = resolve_peer_key_for_url("https://peer.example")
+
+        assert new_key == legacy_key

backend/tests/test_region_dossier_wikimedia_ua.py

@@ -0,0 +1,91 @@
+"""Issues #218 / #219 (tg12): outbound Wikipedia + Wikidata calls must
+identify ShadowBroker via the Wikimedia-recommended User-Agent /
+Api-User-Agent headers.
+
+Before this fix, ``backend/services/region_dossier.py`` called
+``fetch_with_curl(url)`` with no explicit headers, falling back to the
+generic project default UA. That sent a too-anonymous identifier to
+Wikimedia. Per Wikimedia's policy
+(https://foundation.wikimedia.org/wiki/Policy:Wikimedia_Foundation_User-Agent_Policy)
+the API caller should send a stable, contactable identifier so Wikimedia
+operators can rate-limit or reach the project.
+
+This test does NOT make network calls. It patches ``fetch_with_curl``
+and asserts the headers that get passed through.
+"""
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+
+def _fake_resp(payload: dict, status: int = 200) -> MagicMock:
+    r = MagicMock()
+    r.status_code = status
+    r.json.return_value = payload
+    return r
+
+
+def test_wikidata_call_passes_wikimedia_request_headers():
+    from services import region_dossier
+
+    calls = []
+
+    def fake_fetch(url, **kwargs):
+        calls.append(kwargs.get("headers"))
+        return _fake_resp({"results": {"bindings": []}})
+
+    with patch.object(region_dossier, "fetch_with_curl", side_effect=fake_fetch):
+        region_dossier._fetch_wikidata_leader("Testlandia")
+
+    assert calls, "fetch_with_curl was not called"
+    headers = calls[0] or {}
+    assert "User-Agent" in headers
+    assert "Api-User-Agent" in headers
+    # Stable identifier should mention the project + a contact path.
+    assert "Shadowbroker" in headers["Api-User-Agent"] or "ShadowBroker" in headers["Api-User-Agent"]
+    assert "github.com" in headers["Api-User-Agent"].lower()
+
+
+def test_wikipedia_summary_call_passes_wikimedia_request_headers():
+    from services import region_dossier
+
+    calls = []
+
+    def fake_fetch(url, **kwargs):
+        calls.append((url, kwargs.get("headers")))
+        return _fake_resp(
+            {
+                "type": "standard",
+                "description": "test desc",
+                "extract": "test extract",
+                "thumbnail": {"source": ""},
+            }
+        )
+
+    with patch.object(region_dossier, "fetch_with_curl", side_effect=fake_fetch):
+        region_dossier._fetch_local_wiki_summary("Paris", "France")
+
+    # At least one Wikipedia REST call was issued.
+    wikipedia_calls = [c for c in calls if "wikipedia.org" in c[0]]
+    assert wikipedia_calls, "no Wikipedia call was issued"
+    for url, headers in wikipedia_calls:
+        headers = headers or {}
+        assert "User-Agent" in headers, f"missing User-Agent on {url}"
+        assert "Api-User-Agent" in headers, f"missing Api-User-Agent on {url}"
+        assert "github.com" in headers["Api-User-Agent"].lower()
+
+
+def test_wikimedia_headers_constant_is_stable():
+    """Regression guard: if someone removes the contact path from the
+    Api-User-Agent we want a loud test failure, not a silent ToS drift.
+    """
+    from services.region_dossier import _WIKIMEDIA_REQUEST_HEADERS
+
+    aua = _WIKIMEDIA_REQUEST_HEADERS.get("Api-User-Agent", "")
+    assert "Shadowbroker" in aua or "ShadowBroker" in aua
+    assert "github.com" in aua.lower()
+    # Must include a path Wikimedia operators can use to contact us
+    # (we use /issues against the public repo).
+    assert "issues" in aua.lower()

backend/tests/test_round5_settings_info_disclosure.py

@@ -0,0 +1,263 @@
+"""Issues #243, #252, #253 (tg12): settings endpoints must not leak
+operational posture to unauthenticated callers.
+
+- **#243**: ``GET /api/settings/wormhole``, ``/api/settings/privacy-profile``,
+  and ``/api/settings/node`` were leaking transport choice, anonymous-mode
+  state, the named privacy profile, and node-participant state to any
+  unauthenticated caller. The fix tightens the redaction allowlists to
+  expose ONLY a bare "is this feature on?" boolean and gates node mode
+  behind authenticated reads.
+
+- **#252**: ``GET /api/settings/news-feeds`` returned the operator's full
+  curated feed inventory (names + URLs) to anyone. Now gated on
+  local-operator.
+
+- **#253**: ``GET /api/settings/timemachine`` returned whether archival
+  capture is enabled to anyone. Now gated on local-operator.
+
+Auth model: ``require_local_operator`` allows loopback (Tauri shell),
+the Docker bridge frontend container (via the hostname-bound trust from
+PR #278), and any caller that presents the configured admin key.
+Anonymous LAN or internet callers do NOT pass and either receive 403
+(news-feeds, timemachine) or a redacted minimum (wormhole / node).
+"""
+from __future__ import annotations
+
+from unittest.mock import patch, MagicMock
+
+import pytest
+from fastapi.testclient import TestClient
+
+
+_ADMIN_KEY = "test-admin-key-for-round5-fixture-32+chars"
+
+
+@pytest.fixture
+def client():
+    """TestClient with the private-lane transport middleware disabled.
+
+    Same shape as the oracle resolve fixture — the mesh privacy
+    middleware returns 202 for ``/api/settings/*`` under TestClient
+    because Wormhole is not actually running. Patching out the tier
+    requirement lets requests reach the route's auth gate.
+    """
+    import main
+    with patch("main._minimum_transport_tier", return_value=None):
+        yield TestClient(main.app, raise_server_exceptions=False)
+
+
+# ---------------------------------------------------------------------------
+# #243: Wormhole posture redaction
+# ---------------------------------------------------------------------------
+
+
+class TestWormholeSettingsRedaction:
+    """``GET /api/settings/wormhole`` must NOT leak transport choice or
+    anonymous-mode state to unauthenticated callers."""
+
+    def _read_settings_payload(self):
+        return {
+            "enabled": True,
+            "transport": "tor_arti",
+            "anonymous_mode": True,
+            "privacy_profile": "high",
+            "socks_proxy": "socks5h://127.0.0.1:9050",
+        }
+
+    def test_anonymous_caller_sees_only_enabled_bool(self, client):
+        with (
+            patch("main.read_wormhole_settings", return_value=self._read_settings_payload()),
+            patch("routers.wormhole.read_wormhole_settings", return_value=self._read_settings_payload()),
+            patch("services.wormhole_settings.read_wormhole_settings", return_value=self._read_settings_payload()),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get("/api/settings/wormhole")
+        assert r.status_code == 200
+        body = r.json()
+        # Only the bare "is Wormhole on?" boolean is exposed publicly.
+        assert "enabled" in body
+        assert body["enabled"] is True
+        # Posture fields the audit flagged must be absent.
+        assert "transport" not in body
+        assert "anonymous_mode" not in body
+        assert "privacy_profile" not in body
+        assert "socks_proxy" not in body
+
+    def test_authenticated_caller_sees_full_state(self, client):
+        with (
+            patch("main.read_wormhole_settings", return_value=self._read_settings_payload()),
+            patch("routers.wormhole.read_wormhole_settings", return_value=self._read_settings_payload()),
+            patch("services.wormhole_settings.read_wormhole_settings", return_value=self._read_settings_payload()),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get(
+                "/api/settings/wormhole",
+                headers={"X-Admin-Key": _ADMIN_KEY},
+            )
+        assert r.status_code == 200
+        body = r.json()
+        # All fields visible when authenticated.
+        assert body["enabled"] is True
+        assert body["transport"] == "tor_arti"
+        assert body["anonymous_mode"] is True
+        assert body["privacy_profile"] == "high"
+
+
+class TestPrivacyProfileRedaction:
+    """``GET /api/settings/privacy-profile`` must NOT leak the named
+    profile to unauthenticated callers (the profile name itself
+    discloses operator intent)."""
+
+    def _payload(self):
+        return {
+            "enabled": True,
+            "transport": "tor_arti",
+            "anonymous_mode": True,
+            "privacy_profile": "high",
+        }
+
+    def test_anonymous_caller_sees_only_wormhole_enabled_bool(self, client):
+        with (
+            patch("main.read_wormhole_settings", return_value=self._payload()),
+            patch("routers.wormhole.read_wormhole_settings", return_value=self._payload()),
+            patch("services.wormhole_settings.read_wormhole_settings", return_value=self._payload()),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get("/api/settings/privacy-profile")
+        assert r.status_code == 200
+        body = r.json()
+        assert "wormhole_enabled" in body
+        assert body["wormhole_enabled"] is True
+        # The named profile, transport, and anonymous mode must NOT
+        # leak to anonymous callers.
+        assert "profile" not in body or body.get("profile") is None
+        assert "transport" not in body
+        assert "anonymous_mode" not in body
+
+    def test_authenticated_caller_sees_named_profile_and_transport(self, client):
+        with (
+            patch("main.read_wormhole_settings", return_value=self._payload()),
+            patch("routers.wormhole.read_wormhole_settings", return_value=self._payload()),
+            patch("services.wormhole_settings.read_wormhole_settings", return_value=self._payload()),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get(
+                "/api/settings/privacy-profile",
+                headers={"X-Admin-Key": _ADMIN_KEY},
+            )
+        assert r.status_code == 200
+        body = r.json()
+        assert body["profile"] == "high"
+        assert body["wormhole_enabled"] is True
+        assert body["transport"] == "tor_arti"
+        assert body["anonymous_mode"] is True
+
+
+class TestNodeSettingsRedaction:
+    """``GET /api/settings/node`` must NOT disclose node_mode or
+    node_enabled to anonymous callers."""
+
+    def _node_data(self):
+        return {"some_node_field": "value"}
+
+    def test_anonymous_caller_sees_empty_stub(self, client):
+        with (
+            patch("services.node_settings.read_node_settings", return_value=self._node_data()),
+            patch("routers.admin._current_node_mode", return_value="participant"),
+            patch("routers.admin._participant_node_enabled", return_value=True),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get("/api/settings/node")
+        assert r.status_code == 200
+        body = r.json()
+        # No posture fields.
+        assert "node_mode" not in body
+        assert "node_enabled" not in body
+        assert "some_node_field" not in body
+
+    def test_authenticated_caller_sees_full_node_state(self, client):
+        with (
+            patch("services.node_settings.read_node_settings", return_value=self._node_data()),
+            patch("routers.admin._current_node_mode", return_value="participant"),
+            patch("routers.admin._participant_node_enabled", return_value=True),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get(
+                "/api/settings/node",
+                headers={"X-Admin-Key": _ADMIN_KEY},
+            )
+        assert r.status_code == 200
+        body = r.json()
+        assert body["node_mode"] == "participant"
+        assert body["node_enabled"] is True
+        assert body["some_node_field"] == "value"
+
+
+# ---------------------------------------------------------------------------
+# #252: news-feeds auth gate
+# ---------------------------------------------------------------------------
+
+
+class TestNewsFeedsAuthGate:
+    def _fake_feeds(self):
+        return [
+            {"name": "Custom Internal", "url": "https://internal.example/rss", "weight": 5},
+            {"name": "Default News", "url": "https://news.example/rss", "weight": 3},
+        ]
+
+    def test_anonymous_caller_rejected(self, client):
+        with (
+            patch("services.news_feed_config.get_feeds", return_value=self._fake_feeds()) as get_feeds,
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get("/api/settings/news-feeds")
+        assert r.status_code == 403
+        # Critically: the underlying config read must NOT have been performed
+        # (else the response body could leak the count via response timing).
+        assert get_feeds.call_count == 0
+
+    def test_authenticated_caller_sees_full_feed_inventory(self, client):
+        with (
+            patch("services.news_feed_config.get_feeds", return_value=self._fake_feeds()),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get(
+                "/api/settings/news-feeds",
+                headers={"X-Admin-Key": _ADMIN_KEY},
+            )
+        assert r.status_code == 200
+        body = r.json()
+        assert len(body) == 2
+        assert body[0]["name"] == "Custom Internal"
+        assert body[0]["url"] == "https://internal.example/rss"
+
+
+# ---------------------------------------------------------------------------
+# #253: timemachine auth gate
+# ---------------------------------------------------------------------------
+
+
+class TestTimemachineAuthGate:
+    def test_anonymous_caller_rejected(self, client):
+        node_data = {"timemachine_enabled": True}
+        with (
+            patch("services.node_settings.read_node_settings", return_value=node_data),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get("/api/settings/timemachine")
+        assert r.status_code == 403
+
+    def test_authenticated_caller_sees_enabled_state(self, client):
+        node_data = {"timemachine_enabled": True}
+        with (
+            patch("services.node_settings.read_node_settings", return_value=node_data),
+            patch("auth._current_admin_key", return_value=_ADMIN_KEY),
+        ):
+            r = client.get(
+                "/api/settings/timemachine",
+                headers={"X-Admin-Key": _ADMIN_KEY},
+            )
+        assert r.status_code == 200
+        body = r.json()
+        assert body["enabled"] is True
+        assert "storage_warning" in body

backend/tests/test_tor_bundle_symlink_filter.py

@@ -0,0 +1,222 @@
+"""Issue #251 (tg12): Tor bundle extraction must refuse symlink and
+hardlink members.
+
+The previous extractor checked ``member.name`` against path traversal
+but never inspected ``member.linkname``. Python 3.11's ``tarfile``
+honors symlinks during ``extractall()``, so a malicious archive could
+ship a member named ``innocent.txt`` whose linkname points at an
+arbitrary filesystem location. After extraction, reads of innocent.txt
+dereference to that location; writes corrupt it.
+
+The fix categorically refuses any link member during extraction.
+Tor Expert Bundles never legitimately contain symlinks or hardlinks,
+so this is non-disruptive for real updates and a hard stop for hostile
+archives.
+
+These tests build synthetic tar archives covering each refused case
+and assert ``_extract_tor_bundle_safely`` rejects them.
+"""
+import io
+import os
+import stat
+import tarfile
+from pathlib import Path
+
+import pytest
+
+from services.tor_hidden_service import _extract_tor_bundle_safely
+
+
+def _build_archive(tmp_path: Path, members: list) -> Path:
+    """Write a .tar.gz with the given (name, builder) pairs.
+
+    Each builder is called with the open tarfile and is responsible for
+    adding its member however it likes (regular file, symlink, etc.).
+    """
+    archive = tmp_path / "test_bundle.tar.gz"
+    with tarfile.open(str(archive), "w:gz") as tar:
+        for name, builder in members:
+            builder(tar, name)
+    return archive
+
+
+def _add_regular_file(tar: tarfile.TarFile, name: str, payload: bytes = b"hello") -> None:
+    info = tarfile.TarInfo(name)
+    info.size = len(payload)
+    info.mode = 0o644
+    info.type = tarfile.REGTYPE
+    tar.addfile(info, io.BytesIO(payload))
+
+
+def _add_symlink(tar: tarfile.TarFile, name: str, linkname: str) -> None:
+    info = tarfile.TarInfo(name)
+    info.size = 0
+    info.type = tarfile.SYMTYPE
+    info.linkname = linkname
+    info.mode = 0o777
+    tar.addfile(info)
+
+
+def _add_hardlink(tar: tarfile.TarFile, name: str, linkname: str) -> None:
+    info = tarfile.TarInfo(name)
+    info.size = 0
+    info.type = tarfile.LNKTYPE
+    info.linkname = linkname
+    info.mode = 0o644
+    tar.addfile(info)
+
+
+def _add_fifo(tar: tarfile.TarFile, name: str) -> None:
+    info = tarfile.TarInfo(name)
+    info.type = tarfile.FIFOTYPE
+    info.mode = 0o644
+    tar.addfile(info)
+
+
+def test_clean_archive_extracts_successfully(tmp_path):
+    """A normal archive with only regular files extracts fine."""
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    def add_normal(tar, name):
+        _add_regular_file(tar, name, b"clean content")
+
+    archive = _build_archive(
+        tmp_path,
+        [
+            ("tor/tor.exe", add_normal),
+            ("tor/data/geoip", add_normal),
+        ],
+    )
+    assert _extract_tor_bundle_safely(archive, install_dir) is True
+    assert (install_dir / "tor" / "tor.exe").is_file()
+    assert (install_dir / "tor" / "data" / "geoip").is_file()
+
+
+def test_symlink_member_is_rejected(tmp_path, caplog):
+    """Issue #251 core regression: symlink members are refused."""
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    archive = _build_archive(
+        tmp_path,
+        [
+            ("tor/innocent.txt", lambda t, n: _add_symlink(t, n, "/etc/passwd")),
+        ],
+    )
+
+    import logging
+
+    with caplog.at_level(logging.ERROR):
+        result = _extract_tor_bundle_safely(archive, install_dir)
+
+    assert result is False
+    # No file should have been created
+    assert not (install_dir / "tor" / "innocent.txt").exists()
+    # Log should explain why
+    assert any(
+        "symlinks/hardlinks are not allowed" in rec.getMessage()
+        for rec in caplog.records
+    )
+
+
+def test_hardlink_member_is_rejected(tmp_path):
+    """Hardlinks are refused for the same reason as symlinks."""
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    archive = _build_archive(
+        tmp_path,
+        [
+            ("tor/regular.txt", lambda t, n: _add_regular_file(t, n)),
+            ("tor/sneaky.txt", lambda t, n: _add_hardlink(t, n, "regular.txt")),
+        ],
+    )
+    assert _extract_tor_bundle_safely(archive, install_dir) is False
+    # The whole extraction is refused even though only one member is bad.
+    assert not (install_dir / "tor" / "regular.txt").exists()
+
+
+def test_symlink_with_relative_target_still_rejected(tmp_path):
+    """Even a relative symlink target inside the install dir is refused.
+
+    We don't allow symlinks at all — there is no legitimate Tor bundle
+    use case for them, and an attacker can chain link redirections in
+    ways the path-resolution check is poor at catching.
+    """
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    archive = _build_archive(
+        tmp_path,
+        [
+            ("tor/alias.txt", lambda t, n: _add_symlink(t, n, "tor/tor.exe")),
+        ],
+    )
+    assert _extract_tor_bundle_safely(archive, install_dir) is False
+
+
+def test_fifo_or_device_member_is_rejected(tmp_path):
+    """Non-regular-non-directory members (FIFOs, devices) are refused."""
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    archive = _build_archive(
+        tmp_path,
+        [
+            ("tor/weird.fifo", _add_fifo),
+        ],
+    )
+    assert _extract_tor_bundle_safely(archive, install_dir) is False
+
+
+def test_path_traversal_member_is_rejected(tmp_path):
+    """Pre-existing path-traversal guard still works under the new shape."""
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    def add_traversal(tar, name):
+        _add_regular_file(tar, name)
+
+    # ../../escape.txt resolves outside install_dir on most platforms.
+    archive = _build_archive(
+        tmp_path,
+        [
+            ("../../escape.txt", add_traversal),
+        ],
+    )
+    assert _extract_tor_bundle_safely(archive, install_dir) is False
+
+
+def test_malformed_tar_is_rejected(tmp_path):
+    """A corrupt/non-tar file is rejected without crashing."""
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    bogus = tmp_path / "not-a-tar.tar.gz"
+    bogus.write_bytes(b"this is not a tar archive at all")
+
+    assert _extract_tor_bundle_safely(bogus, install_dir) is False
+
+
+def test_extraction_failure_does_not_leave_partial_state_referenced_to_caller(tmp_path):
+    """When extraction fails partway, the caller relies on a False return
+    to know it must clean up. We test the contract here — actual cleanup
+    of files that may have been written by tar.extractall() before the
+    failure point isn't part of THIS helper's responsibility (the caller
+    deletes the install dir if needed)."""
+    install_dir = tmp_path / "install"
+    install_dir.mkdir()
+
+    # Hostile archive: one good file, then a symlink. Whether the good
+    # file was written or not, the return value must be False so the
+    # caller refuses the bundle.
+    archive = _build_archive(
+        tmp_path,
+        [
+            ("tor/clean.txt", lambda t, n: _add_regular_file(t, n)),
+            ("tor/evil-link.txt", lambda t, n: _add_symlink(t, n, "/etc/passwd")),
+        ],
+    )
+
+    assert _extract_tor_bundle_safely(archive, install_dir) is False

backend/tests/test_update_integrity_chain.py

@@ -0,0 +1,338 @@
+"""Issue #231 — self-update SHA-256 verification.
+
+Before this fix, ``_validate_zip_hash`` returned silently whenever the
+``MESH_UPDATE_SHA256`` env var was unset (the default — nothing in the
+install docs ever told operators to set it). That made the auto-updater
+a supply-chain RCE on any compromise of the GitHub release pipeline.
+
+The fix introduces a four-source verification chain:
+
+  1. ``MESH_UPDATE_SHA256`` env var (operator override, preserved)
+  2. ``SHA256SUMS.txt`` asset published alongside the release (primary)
+  3. Baked-in ``backend/data/release_digests.json`` (fallback)
+  4. HTTPS-only fallback with a loud warning (preserves auto-update during
+     transient outages so the user isn't stuck)
+
+A mismatch from any source that DID respond is fatal. Only the "no
+source reachable at all" case falls back to HTTPS-only.
+"""
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from services import updater
+from services.updater import (
+    _compute_sha256,
+    _fetch_sha256sums,
+    _load_baked_in_release_digests,
+    _validate_zip_hash,
+)
+
+
+@pytest.fixture
+def fake_archive(tmp_path):
+    """A tiny synthetic zip-shaped file so we can compute a known digest."""
+    archive = tmp_path / "update.zip"
+    payload = b"this is not really a release archive"
+    archive.write_bytes(payload)
+    expected = hashlib.sha256(payload).hexdigest().lower()
+    return str(archive), expected
+
+
+def test_baked_in_release_digests_file_loads():
+    """The shipped release_digests.json must parse and contain v0.9.79."""
+    digests = _load_baked_in_release_digests()
+    assert "v0.9.79" in digests
+    entry = digests["v0.9.79"]
+    assert "ShadowBroker_v0.9.79.zip" in entry
+    digest = entry["ShadowBroker_v0.9.79.zip"]
+    assert len(digest) == 64
+    assert all(c in "0123456789abcdef" for c in digest)
+
+
+def test_baked_in_skips_comment_keys():
+    """The _comment top-level key is ignored, not surfaced as a release."""
+    digests = _load_baked_in_release_digests()
+    assert "_comment" not in digests
+
+
+def test_compute_sha256_matches_known_value(fake_archive):
+    archive, expected = fake_archive
+    assert _compute_sha256(archive) == expected
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# Source 1: MESH_UPDATE_SHA256 env override
+# ──────────────────────────────────────────────────────────────────────────
+
+
+def test_env_override_matching_passes(fake_archive, monkeypatch):
+    """Path 1: operator pinned the exact digest via env. Match = success."""
+    archive, expected = fake_archive
+    monkeypatch.setenv("MESH_UPDATE_SHA256", expected)
+
+    note = _validate_zip_hash(archive)
+    assert "MESH_UPDATE_SHA256" in note
+
+
+def test_env_override_mismatch_fails_loudly(fake_archive, monkeypatch):
+    """Path 1: operator pinned a different digest. Mismatch = fatal."""
+    archive, _expected = fake_archive
+    monkeypatch.setenv("MESH_UPDATE_SHA256", "0" * 64)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        _validate_zip_hash(archive)
+    assert "mismatch" in str(exc_info.value).lower()
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# Source 2: SHA256SUMS.txt asset
+# ──────────────────────────────────────────────────────────────────────────
+
+
+def test_sha256sums_matching_passes(fake_archive, monkeypatch):
+    """Path 2: SHA256SUMS.txt has the correct digest for our asset."""
+    archive, expected = fake_archive
+    monkeypatch.delenv("MESH_UPDATE_SHA256", raising=False)
+
+    def fake_sums(url):
+        return {"ShadowBroker_v9.9.9.zip": expected}
+
+    monkeypatch.setattr(updater, "_fetch_sha256sums", fake_sums)
+    note = _validate_zip_hash(
+        archive,
+        asset_name="ShadowBroker_v9.9.9.zip",
+        sha256sums_url="https://example.test/SHA256SUMS.txt",
+        release_tag="v9.9.9",
+    )
+    assert "SHA256SUMS.txt" in note
+
+
+def test_sha256sums_mismatch_fails_loudly(fake_archive, monkeypatch):
+    """Path 2: SHA256SUMS.txt has a different digest. Refuse."""
+    archive, _expected = fake_archive
+    monkeypatch.delenv("MESH_UPDATE_SHA256", raising=False)
+
+    def fake_sums(url):
+        return {"ShadowBroker_v9.9.9.zip": "0" * 64}
+
+    monkeypatch.setattr(updater, "_fetch_sha256sums", fake_sums)
+    with pytest.raises(RuntimeError) as exc_info:
+        _validate_zip_hash(
+            archive,
+            asset_name="ShadowBroker_v9.9.9.zip",
+            sha256sums_url="https://example.test/SHA256SUMS.txt",
+            release_tag="v9.9.9",
+        )
+    assert "mismatch" in str(exc_info.value).lower()
+    assert "SHA256SUMS" in str(exc_info.value)
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# Source 3: baked-in digest list
+# ──────────────────────────────────────────────────────────────────────────
+
+
+def test_baked_in_matching_passes(fake_archive, monkeypatch):
+    """Path 3: SHA256SUMS unreachable, but the baked-in list has us."""
+    archive, expected = fake_archive
+    monkeypatch.delenv("MESH_UPDATE_SHA256", raising=False)
+    monkeypatch.setattr(updater, "_fetch_sha256sums", lambda url: {})
+    monkeypatch.setattr(
+        updater,
+        "_load_baked_in_release_digests",
+        lambda: {"v9.9.9": {"ShadowBroker_v9.9.9.zip": expected}},
+    )
+
+    note = _validate_zip_hash(
+        archive,
+        asset_name="ShadowBroker_v9.9.9.zip",
+        sha256sums_url="https://example.test/SHA256SUMS.txt",
+        release_tag="v9.9.9",
+    )
+    assert "baked-in" in note
+
+
+def test_baked_in_mismatch_fails_loudly(fake_archive, monkeypatch):
+    """Path 3: baked-in says something different. Refuse."""
+    archive, _expected = fake_archive
+    monkeypatch.delenv("MESH_UPDATE_SHA256", raising=False)
+    monkeypatch.setattr(updater, "_fetch_sha256sums", lambda url: {})
+    monkeypatch.setattr(
+        updater,
+        "_load_baked_in_release_digests",
+        lambda: {"v9.9.9": {"ShadowBroker_v9.9.9.zip": "0" * 64}},
+    )
+
+    with pytest.raises(RuntimeError) as exc_info:
+        _validate_zip_hash(
+            archive,
+            asset_name="ShadowBroker_v9.9.9.zip",
+            sha256sums_url="",
+            release_tag="v9.9.9",
+        )
+    assert "mismatch" in str(exc_info.value).lower()
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# Source 4: HTTPS-only fallback
+# ──────────────────────────────────────────────────────────────────────────
+
+
+def test_https_only_fallback_when_no_source_available(fake_archive, monkeypatch, caplog):
+    """Path 4: nothing matches — fall back to HTTPS-only with loud warning.
+
+    This preserves the auto-update flow during transient outages: an
+    operator on a flaky network during update doesn't get a hostile
+    error, they get a degraded-but-functional update with a clear log
+    message.
+    """
+    import logging
+
+    archive, _expected = fake_archive
+    monkeypatch.delenv("MESH_UPDATE_SHA256", raising=False)
+    monkeypatch.setattr(updater, "_fetch_sha256sums", lambda url: {})
+    monkeypatch.setattr(updater, "_load_baked_in_release_digests", lambda: {})
+
+    with caplog.at_level(logging.WARNING):
+        note = _validate_zip_hash(
+            archive,
+            asset_name="ShadowBroker_v99.99.zip",
+            sha256sums_url="",
+            release_tag="v99.99",
+        )
+
+    assert "https-only" in note.lower()
+    assert any(
+        "fell back to HTTPS-only" in rec.getMessage() for rec in caplog.records
+    )
+
+
+def test_https_only_fallback_when_release_tag_unknown(fake_archive, monkeypatch):
+    """Path 4 also kicks in when we have a baked-in list but it doesn't
+    contain THIS release tag — e.g. a brand-new release that the local
+    install hasn't seen a digest for yet."""
+    archive, _expected = fake_archive
+    monkeypatch.delenv("MESH_UPDATE_SHA256", raising=False)
+    monkeypatch.setattr(updater, "_fetch_sha256sums", lambda url: {})
+    monkeypatch.setattr(
+        updater,
+        "_load_baked_in_release_digests",
+        lambda: {"v0.0.1": {"old.zip": "0" * 64}},  # different tag, doesn't match
+    )
+
+    note = _validate_zip_hash(
+        archive,
+        asset_name="ShadowBroker_v99.99.zip",
+        sha256sums_url="",
+        release_tag="v99.99",
+    )
+    assert "https-only" in note.lower()
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# Precedence (env > SHA256SUMS > baked-in > https-only)
+# ──────────────────────────────────────────────────────────────────────────
+
+
+def test_env_override_beats_all_other_sources(fake_archive, monkeypatch):
+    """When MESH_UPDATE_SHA256 is set, it's the only source consulted.
+
+    The other sources may return false positives or negatives — they
+    shouldn't be queried at all when the operator pinned an exact value.
+    """
+    archive, expected = fake_archive
+    monkeypatch.setenv("MESH_UPDATE_SHA256", expected)
+
+    def boom_sums(url):
+        raise AssertionError("SHA256SUMS source was queried despite env override")
+
+    def boom_baked():
+        raise AssertionError("Baked-in list was queried despite env override")
+
+    monkeypatch.setattr(updater, "_fetch_sha256sums", boom_sums)
+    monkeypatch.setattr(updater, "_load_baked_in_release_digests", boom_baked)
+
+    note = _validate_zip_hash(
+        archive,
+        asset_name="any.zip",
+        sha256sums_url="https://example.test/SHA256SUMS.txt",
+        release_tag="any",
+    )
+    assert "MESH_UPDATE_SHA256" in note
+
+
+# ──────────────────────────────────────────────────────────────────────────
+# _fetch_sha256sums parser
+# ──────────────────────────────────────────────────────────────────────────
+
+
+def test_fetch_sha256sums_parses_standard_format(monkeypatch):
+    """Standard ``sha256sum`` output: ``<digest>  <filename>``."""
+    class _Resp:
+        text = (
+            "f6877c1d66614525315ea82636ce9f7b41178332c4dbf90d27431a1ea1d9cd47  ShadowBroker_v0.9.79.zip\n"
+            "e0713c3cdda184cfbea750bfac0d62a35678fec00847e6476f2cac8e7e42046e  ShadowBroker_0.9.79_x64_en-US.msi\n"
+        )
+
+        def raise_for_status(self):
+            pass
+
+    def fake_get(url, timeout=15):
+        return _Resp()
+
+    monkeypatch.setattr(updater.requests, "get", fake_get)
+    monkeypatch.setattr(updater, "_validate_update_url", lambda url, **kw: url)
+    sums = _fetch_sha256sums("https://example.test/SHA256SUMS.txt")
+    assert sums["ShadowBroker_v0.9.79.zip"].startswith("f6877c1d")
+    assert sums["ShadowBroker_0.9.79_x64_en-US.msi"].startswith("e0713c3c")
+
+
+def test_fetch_sha256sums_handles_binary_marker(monkeypatch):
+    """sha256sum -b output: ``<digest> *<filename>``."""
+    class _Resp:
+        text = "f6877c1d66614525315ea82636ce9f7b41178332c4dbf90d27431a1ea1d9cd47 *ShadowBroker_v0.9.79.zip\n"
+
+        def raise_for_status(self):
+            pass
+
+    monkeypatch.setattr(updater.requests, "get", lambda url, timeout=15: _Resp())
+    monkeypatch.setattr(updater, "_validate_update_url", lambda url, **kw: url)
+    sums = _fetch_sha256sums("https://example.test/SHA256SUMS.txt")
+    assert "ShadowBroker_v0.9.79.zip" in sums
+
+
+def test_fetch_sha256sums_skips_malformed_lines(monkeypatch):
+    """Lines that don't parse cleanly are ignored, not aborted on."""
+    class _Resp:
+        text = (
+            "# comment line\n"
+            "\n"
+            "not-a-digest  bogus.txt\n"
+            "f6877c1d66614525315ea82636ce9f7b41178332c4dbf90d27431a1ea1d9cd47  good.zip\n"
+        )
+
+        def raise_for_status(self):
+            pass
+
+    monkeypatch.setattr(updater.requests, "get", lambda url, timeout=15: _Resp())
+    monkeypatch.setattr(updater, "_validate_update_url", lambda url, **kw: url)
+    sums = _fetch_sha256sums("https://example.test/SHA256SUMS.txt")
+    assert "good.zip" in sums
+    assert "bogus.txt" not in sums
+
+
+def test_fetch_sha256sums_handles_network_failure(monkeypatch):
+    """If the SHA256SUMS asset can't be fetched, return empty (caller
+    falls through to baked-in / https-only)."""
+    import requests as _req
+
+    def fake_get(url, timeout=15):
+        raise _req.exceptions.ConnectionError("upstream down")
+
+    monkeypatch.setattr(updater.requests, "get", fake_get)
+    monkeypatch.setattr(updater, "_validate_update_url", lambda url, **kw: url)
+    sums = _fetch_sha256sums("https://example.test/SHA256SUMS.txt")
+    assert sums == {}

docker-compose.yml

@@ -28,6 +28,15 @@ services:
       - MESH_RELAY_PEERS=${MESH_RELAY_PEERS:-}
       # Shared transport auth for operator peer push. Must be set to a unique secret per deployment.
       - MESH_PEER_PUSH_SECRET=${MESH_PEER_PUSH_SECRET:-}
+      # Issue #256: optional per-peer HMAC secrets. Comma-separated
+      # `url=secret` pairs (no spaces). When a peer URL appears here, only
+      # the listed per-peer secret is accepted for it — the global
+      # MESH_PEER_PUSH_SECRET above is ignored for that specific URL. This
+      # closes the cross-peer impersonation surface for multi-peer fleets.
+      # Single-peer installs leave this empty (default) for unchanged
+      # behavior. Both sides of a peering must agree on the per-peer
+      # secret for a given URL.
+      - MESH_PEER_SECRETS=${MESH_PEER_SECRETS:-}
       # Meshtastic MQTT is opt-in to avoid passive load on the public broker.
       # Set MESH_MQTT_ENABLED=true in .env only when this node should join live MQTT.
       - MESH_MQTT_ENABLED=${MESH_MQTT_ENABLED:-false}
@@ -43,6 +52,11 @@ services:
       # The bundled Docker UI talks to the backend across Docker's private bridge.
       # Treat that bridge as local operator access while ports remain bound to 127.0.0.1 by default.
       - SHADOWBROKER_TRUST_DOCKER_BRIDGE_LOCAL_OPERATOR=${SHADOWBROKER_TRUST_DOCKER_BRIDGE_LOCAL_OPERATOR:-1}
+      # Issue #250: bridge trust is now bound to specific container hostnames
+      # (default: 'frontend' compose service + 'shadowbroker-frontend' container
+      # name). If you rename the frontend service or run with a different
+      # container_name, list the hostnames here (comma-separated, no spaces).
+      - SHADOWBROKER_TRUSTED_FRONTEND_HOSTS=${SHADOWBROKER_TRUSTED_FRONTEND_HOSTS:-frontend,shadowbroker-frontend}
     volumes:
       - backend_data:/app/data
     restart: unless-stopped

frontend/src/__tests__/utils/wikimediaClient.test.ts

@@ -0,0 +1,164 @@
+/**
+ * Issues #218 / #219 / #220 (tg12 external audit):
+ *
+ * Every browser-direct call to Wikipedia or Wikidata must send the
+ * `Api-User-Agent` header that Wikimedia's UA policy asks for. These
+ * tests pin that requirement on the shared `lib/wikimediaClient`
+ * helper that WikiImage, NewsFeed, and useRegionDossier all route
+ * through, so a future refactor that drops the header gets a loud
+ * test failure rather than a silent ToS regression.
+ */
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+  WIKIMEDIA_API_USER_AGENT,
+  fetchWikipediaSummary,
+  fetchWikidataSparql,
+  _resetWikimediaClientCacheForTests,
+} from '@/lib/wikimediaClient';
+
+const originalFetch = globalThis.fetch;
+
+describe('lib/wikimediaClient', () => {
+  beforeEach(() => {
+    _resetWikimediaClientCacheForTests();
+  });
+
+  afterEach(() => {
+    globalThis.fetch = originalFetch;
+    vi.restoreAllMocks();
+  });
+
+  it('exposes a stable Api-User-Agent identifier with a contact path', () => {
+    expect(WIKIMEDIA_API_USER_AGENT).toContain('Shadowbroker');
+    expect(WIKIMEDIA_API_USER_AGENT.toLowerCase()).toContain('github.com');
+    expect(WIKIMEDIA_API_USER_AGENT.toLowerCase()).toContain('issues');
+  });
+
+  it('sends Api-User-Agent on Wikipedia summary fetch', async () => {
+    const calls: Array<{ url: string; init?: RequestInit }> = [];
+    globalThis.fetch = vi.fn(async (url: any, init?: RequestInit) => {
+      calls.push({ url: String(url), init });
+      return new Response(
+        JSON.stringify({
+          type: 'standard',
+          title: 'Boeing 747',
+          description: 'aircraft',
+          extract: 'long extract',
+          thumbnail: { source: 'https://example.org/thumb.jpg' },
+        }),
+        { status: 200 },
+      );
+    }) as any;
+
+    const summary = await fetchWikipediaSummary('Boeing 747');
+    expect(summary?.thumbnail).toBe('https://example.org/thumb.jpg');
+    expect(calls).toHaveLength(1);
+    const headers = (calls[0].init?.headers || {}) as Record<string, string>;
+    expect(headers['Api-User-Agent']).toBe(WIKIMEDIA_API_USER_AGENT);
+  });
+
+  it('sends Api-User-Agent on Wikidata SPARQL fetch', async () => {
+    const calls: Array<{ url: string; init?: RequestInit }> = [];
+    globalThis.fetch = vi.fn(async (url: any, init?: RequestInit) => {
+      calls.push({ url: String(url), init });
+      return new Response(
+        JSON.stringify({
+          results: {
+            bindings: [
+              {
+                leaderLabel: { value: 'Test Leader' },
+                govTypeLabel: { value: 'Test Government' },
+              },
+            ],
+          },
+        }),
+        { status: 200 },
+      );
+    }) as any;
+
+    const bindings = await fetchWikidataSparql('SELECT * WHERE { ?s ?p ?o }');
+    expect(bindings).toHaveLength(1);
+    const headers = (calls[0].init?.headers || {}) as Record<string, string>;
+    expect(headers['Api-User-Agent']).toBe(WIKIMEDIA_API_USER_AGENT);
+    expect(headers['Accept']).toBe('application/sparql-results+json');
+  });
+
+  it('shares cache across consecutive callers for the same Wikipedia title', async () => {
+    let fetchCount = 0;
+    globalThis.fetch = vi.fn(async () => {
+      fetchCount++;
+      return new Response(
+        JSON.stringify({
+          type: 'standard',
+          title: 'Eiffel Tower',
+          description: 'iron lattice tower',
+          extract: '...',
+          thumbnail: { source: 'https://example.org/eiffel.jpg' },
+        }),
+        { status: 200 },
+      );
+    }) as any;
+
+    const a = await fetchWikipediaSummary('Eiffel Tower');
+    const b = await fetchWikipediaSummary('Eiffel Tower');
+    expect(fetchCount).toBe(1);
+    expect(a?.thumbnail).toBe(b?.thumbnail);
+  });
+
+  it('deduplicates concurrent in-flight requests for the same title', async () => {
+    let fetchCount = 0;
+    globalThis.fetch = vi.fn(async () => {
+      fetchCount++;
+      await new Promise((r) => setTimeout(r, 5));
+      return new Response(
+        JSON.stringify({
+          type: 'standard',
+          title: 'Mount Fuji',
+          description: 'stratovolcano',
+          extract: '...',
+          thumbnail: { source: 'https://example.org/fuji.jpg' },
+        }),
+        { status: 200 },
+      );
+    }) as any;
+
+    const [a, b, c] = await Promise.all([
+      fetchWikipediaSummary('Mount Fuji'),
+      fetchWikipediaSummary('Mount Fuji'),
+      fetchWikipediaSummary('Mount Fuji'),
+    ]);
+    expect(fetchCount).toBe(1);
+    expect(a?.thumbnail).toBe('https://example.org/fuji.jpg');
+    expect(b).toEqual(a);
+    expect(c).toEqual(a);
+  });
+
+  it('returns null on disambiguation pages without throwing', async () => {
+    globalThis.fetch = vi.fn(async () =>
+      new Response(JSON.stringify({ type: 'disambiguation' }), { status: 200 }),
+    ) as any;
+    const summary = await fetchWikipediaSummary('Mercury');
+    expect(summary).toBeNull();
+  });
+
+  it('returns null on HTTP error without throwing', async () => {
+    globalThis.fetch = vi.fn(async () => new Response('not found', { status: 404 })) as any;
+    const summary = await fetchWikipediaSummary('Nonexistent Article 12345');
+    expect(summary).toBeNull();
+  });
+
+  it('returns null on network error without throwing', async () => {
+    globalThis.fetch = vi.fn(async () => {
+      throw new Error('network down');
+    }) as any;
+    const summary = await fetchWikipediaSummary('Anything');
+    expect(summary).toBeNull();
+  });
+
+  it('returns null on empty input', async () => {
+    globalThis.fetch = vi.fn(async () => new Response('{}', { status: 200 })) as any;
+    expect(await fetchWikipediaSummary('')).toBeNull();
+    expect(await fetchWikipediaSummary('   ')).toBeNull();
+    expect(globalThis.fetch).not.toHaveBeenCalled();
+  });
+});

frontend/src/components/NewsFeed.tsx

@@ -5,6 +5,7 @@ import { motion, AnimatePresence } from 'framer-motion';
 import { AlertTriangle, Clock, Minus, Plus, ExternalLink, Brain, Loader2 } from 'lucide-react';
 import React, { useEffect, useRef, useCallback } from 'react';
 import WikiImage from '@/components/WikiImage';
+import { fetchWikipediaSummary } from '@/lib/wikimediaClient';
 import type { SelectedEntity, RegionDossier, FimiData } from "@/types/dashboard";
 import { useDataKeys } from '@/hooks/useDataStore';
 import { API_BASE } from '@/lib/api';
@@ -203,34 +204,37 @@ function resolveAircraftWikiTitle(model: string | undefined): string | null {
     return AIRCRAFT_WIKI[model] || resolveAcTypeWiki(model);
 }
 
-// Module-level cache for Wikipedia thumbnails (persists across re-renders)
-const _wikiThumbCache: Record<string, { url: string | null; loading: boolean }> = {};
-
+// Issue #220 (tg12): the previous implementation kept its own
+// module-local Wikipedia thumbnail cache and issued anonymous fetches
+// without `Api-User-Agent`. We now delegate to lib/wikimediaClient,
+// which sends the policy-compliant header and shares one cache with
+// WikiImage and useRegionDossier.
 function useAircraftImage(model: string | undefined): { imgUrl: string | null; wikiUrl: string | null; loading: boolean } {
-    const [, forceUpdate] = useState(0);
+    const [imgUrl, setImgUrl] = useState<string | null>(null);
+    const [loading, setLoading] = useState(false);
     const wikiTitle = resolveAircraftWikiTitle(model) || undefined;
     const wikiUrl = wikiTitle ? `https://en.wikipedia.org/wiki/${wikiTitle.replace(/ /g, '_')}` : null;
 
     useEffect(() => {
-        if (!wikiTitle) return;
-        const key = wikiTitle;
-        if (_wikiThumbCache[key]) return; // Already fetched or in-flight
-        _wikiThumbCache[key] = { url: null, loading: true };
-        fetch(`https://en.wikipedia.org/api/rest_v1/page/summary/${encodeURIComponent(wikiTitle)}`)
-            .then(r => r.json())
-            .then(d => {
-                _wikiThumbCache[key] = { url: d.thumbnail?.source || null, loading: false };
-                forceUpdate(n => n + 1);
-            })
-            .catch(() => {
-                _wikiThumbCache[key] = { url: null, loading: false };
-                forceUpdate(n => n + 1);
-            });
+        let cancelled = false;
+        if (!wikiTitle) {
+            setImgUrl(null);
+            setLoading(false);
+            return;
+        }
+        setLoading(true);
+        fetchWikipediaSummary(wikiTitle).then((summary) => {
+            if (cancelled) return;
+            setImgUrl(summary?.thumbnail || null);
+            setLoading(false);
+        });
+        return () => {
+            cancelled = true;
+        };
     }, [wikiTitle]);
 
     if (!wikiTitle) return { imgUrl: null, wikiUrl: null, loading: false };
-    const cached = _wikiThumbCache[wikiTitle];
-    return { imgUrl: cached?.url || null, wikiUrl, loading: cached?.loading || false };
+    return { imgUrl, wikiUrl, loading };
 }
 
 

frontend/src/components/WikiImage.tsx

@@ -1,13 +1,17 @@
 'use client';
 import React, { useState, useEffect } from 'react';
 import ExternalImage from '@/components/ExternalImage';
-
-// Module-level cache: Wikipedia article title → thumbnail URL
-const _cache: Record<string, { url: string | null; done: boolean }> = {};
+import { fetchWikipediaSummary } from '@/lib/wikimediaClient';
 
 /**
  * WikiImage — displays a Wikipedia thumbnail for a given article URL.
- * Uses the Wikipedia REST API with a module-level cache (only fetches once per article).
+ *
+ * Issue #220 (tg12): this component previously had its own
+ * module-local Wikipedia fetch + cache. It now delegates to
+ * `lib/wikimediaClient`, which sends the policy-compliant
+ * `Api-User-Agent` header and shares one cache across every UI
+ * component that asks Wikipedia for an article summary (WikiImage,
+ * NewsFeed, useRegionDossier).
  *
  * Props:
  *   wikiUrl:  Full Wikipedia URL, e.g. "https://en.wikipedia.org/wiki/Boeing_787_Dreamliner"
@@ -26,32 +30,30 @@ export default function WikiImage({
   maxH?: string;
   accent?: string;
 }) {
-  const [, forceUpdate] = useState(0);
+  const [imgUrl, setImgUrl] = useState<string | null>(null);
+  const [loading, setLoading] = useState(true);
 
   // Extract article title from URL
   const title = wikiUrl.replace(/^https?:\/\/[^/]+\/wiki\//, '');
 
   useEffect(() => {
-    if (!title || _cache[title]?.done) return;
-    if (_cache[title]) return; // In-flight
-    _cache[title] = { url: null, done: false };
-
-    fetch(`https://en.wikipedia.org/api/rest_v1/page/summary/${encodeURIComponent(title)}`)
-      .then((r) => r.json())
-      .then((d) => {
-        _cache[title] = { url: d.thumbnail?.source || d.originalimage?.source || null, done: true };
-        forceUpdate((n) => n + 1);
-      })
-      .catch(() => {
-        _cache[title] = { url: null, done: true };
-        forceUpdate((n) => n + 1);
-      });
+    let cancelled = false;
+    if (!title) {
+      setImgUrl(null);
+      setLoading(false);
+      return;
+    }
+    setLoading(true);
+    fetchWikipediaSummary(title).then((summary) => {
+      if (cancelled) return;
+      setImgUrl(summary?.thumbnail || null);
+      setLoading(false);
+    });
+    return () => {
+      cancelled = true;
+    };
   }, [title]);
 
-  const cached = _cache[title];
-  const imgUrl = cached?.url;
-  const loading = cached && !cached.done;
-
   return (
     <div className="pb-2">
       {loading && (

frontend/src/hooks/useRegionDossier.ts

@@ -1,5 +1,6 @@
 import { useCallback, useState, useEffect } from 'react';
 import type { RegionDossier, SelectedEntity } from '@/types/dashboard';
+import { fetchWikipediaSummary, fetchWikidataSparql } from '@/lib/wikimediaClient';
 
 // ─── CACHE ─────────────────────────────────────────────────────────────────
 // Simple in-memory cache keyed by rounded lat/lng (0.1° ≈ 11km grid), 24h TTL.
@@ -114,7 +115,11 @@ async function fetchCountryData(countryCode: string) {
   return Array.isArray(data) ? data[0] || {} : data || {};
 }
 
-/** Fetch head of state + government type from Wikidata SPARQL (direct browser call). */
+/** Fetch head of state + government type from Wikidata SPARQL.
+ *
+ * Issue #218 (tg12): routes through lib/wikimediaClient so the
+ * Api-User-Agent header is set per Wikimedia's UA policy.
+ */
 async function fetchLeader(countryName: string) {
   if (!countryName) return { leader: 'Unknown', government_type: 'Unknown' };
   const safeName = countryName.replace(/"/g, '\\"').replace(/'/g, "\\'");
@@ -127,13 +132,11 @@ async function fetchLeader(countryName: string) {
       SERVICE wikibase:label { bd:serviceParam wikibase:language "en". }
     } LIMIT 1
   `;
-  const url = `https://query.wikidata.org/sparql?query=${encodeURIComponent(sparql)}&format=json`;
-  const res = await fetch(url, {
-    headers: { Accept: 'application/sparql-results+json' },
-  });
-  if (!res.ok) throw new Error(`Wikidata HTTP ${res.status}`);
-  const results = (await res.json()).results?.bindings || [];
-  if (results.length > 0) {
+  const results = await fetchWikidataSparql<{
+    leaderLabel?: { value: string };
+    govTypeLabel?: { value: string };
+  }>(sparql);
+  if (results && results.length > 0) {
     return {
       leader: results[0].leaderLabel?.value || 'Unknown',
       government_type: results[0].govTypeLabel?.value || 'Unknown',
@@ -142,27 +145,25 @@ async function fetchLeader(countryName: string) {
   return { leader: 'Unknown', government_type: 'Unknown' };
 }
 
-/** Fetch Wikipedia summary for a place (direct browser call). */
+/** Fetch Wikipedia summary for a place.
+ *
+ * Issue #219 (tg12): routes through lib/wikimediaClient so the
+ * Api-User-Agent header is set per Wikimedia's UA policy, AND the
+ * shared cache means consecutive useRegionDossier + WikiImage +
+ * NewsFeed lookups for the same article all hit the same slot.
+ */
 async function fetchLocalWikiSummary(placeName: string, countryName = '') {
   if (!placeName) return {};
   const candidates = [placeName];
   if (countryName) candidates.push(`${placeName}, ${countryName}`);
-
   for (const name of candidates) {
-    try {
-      const slug = encodeURIComponent(name.replace(/ /g, '_'));
-      const url = `https://en.wikipedia.org/api/rest_v1/page/summary/${slug}`;
-      const res = await fetch(url);
-      if (!res.ok) continue;
-      const data = await res.json();
-      if (data.type === 'disambiguation') continue;
+    const summary = await fetchWikipediaSummary(name);
+    if (summary) {
       return {
-        description: data.description || '',
-        extract: data.extract || '',
-        thumbnail: data.thumbnail?.source || '',
+        description: summary.description,
+        extract: summary.extract,
+        thumbnail: summary.thumbnail,
       };
-    } catch {
-      continue;
     }
   }
   return {};

frontend/src/lib/wikimediaClient.ts

@@ -0,0 +1,157 @@
+/**
+ * wikimediaClient — single fetch surface for Wikipedia / Wikidata.
+ *
+ * Issues #218, #219, #220 (tg12 external audit):
+ *
+ * Wikimedia's User-Agent policy asks API clients to identify themselves
+ * via `Api-User-Agent` when calling from browser JavaScript (because the
+ * browser does not let JS set `User-Agent` directly). Before this
+ * module existed, three independent components issued anonymous browser
+ * fetches against Wikipedia / Wikidata:
+ *
+ *   - useRegionDossier  (Wikidata SPARQL + Wikipedia REST summary)
+ *   - WikiImage          (Wikipedia REST summary)
+ *   - NewsFeed           (Wikipedia REST summary)
+ *
+ * Each component shipped its own copy-pasted fetch + module-local cache.
+ * Provider-policy compliance was missing in all three places.
+ *
+ * This module centralizes:
+ *
+ *   1. The `Api-User-Agent` header on every request.
+ *   2. A single LRU cache for Wikipedia summary lookups (keyed by article
+ *      title).  Multiple components asking for the same article share
+ *      one in-flight request and one cache slot.
+ *   3. One predictable kill switch — if Wikimedia ever asks us to back
+ *      off, we change `WIKIMEDIA_API_USER_AGENT` here and the whole
+ *      frontend updates.
+ *
+ * This does NOT change end-user UX:
+ *
+ *   - WikiImage still shows the same thumbnails.
+ *   - NewsFeed still shows aircraft thumbnails.
+ *   - useRegionDossier still returns the same place summary + leader.
+ *
+ * What changes:
+ *
+ *   - Wikimedia can identify our traffic from any other anonymous
+ *     browser visitor pool.
+ *   - Provider-policy fixes happen here once, not in three places.
+ */
+
+// Stable identifier per Wikimedia UA policy. Includes a contact path so
+// Wikimedia's operators can reach the project if they need to rate-limit
+// or coordinate. Bump the version when the contact path changes.
+export const WIKIMEDIA_API_USER_AGENT =
+  'Shadowbroker/1.0 (+https://github.com/BigBodyCobain/Shadowbroker; ' +
+  'report issues at /issues)';
+
+// Module-level cache shared by WikiImage, NewsFeed, and useRegionDossier.
+// Keyed by Wikipedia article title (NOT slug — we keep the human-readable
+// form so debugging the cache is easier). Values track in-flight state
+// so concurrent callers for the same title share one network request.
+export interface WikipediaSummary {
+  title: string;
+  description: string;
+  extract: string;
+  thumbnail: string;
+  type: string; // 'standard' | 'disambiguation' | etc.
+}
+
+interface CacheEntry {
+  summary: WikipediaSummary | null;
+  inflight: Promise<WikipediaSummary | null> | null;
+  loaded: boolean;
+}
+
+const _summaryCache: Map<string, CacheEntry> = new Map();
+const SUMMARY_CACHE_MAX = 512;
+
+function evictIfOverCap() {
+  if (_summaryCache.size <= SUMMARY_CACHE_MAX) return;
+  const oldest = _summaryCache.keys().next().value;
+  if (oldest) _summaryCache.delete(oldest);
+}
+
+/** Fetch a Wikipedia article summary (titles, NOT URLs).
+ *
+ * Empty / invalid input resolves to `null`. Network errors and disambig
+ * pages also resolve to `null` so callers can render a fallback without
+ * a try/catch. Per the audit's "fail forward, not loud" rule.
+ */
+export async function fetchWikipediaSummary(
+  title: string,
+): Promise<WikipediaSummary | null> {
+  const trimmed = (title || '').trim();
+  if (!trimmed) return null;
+
+  const cached = _summaryCache.get(trimmed);
+  if (cached?.loaded) return cached.summary;
+  if (cached?.inflight) return cached.inflight;
+
+  const slug = encodeURIComponent(trimmed.replace(/ /g, '_'));
+  const url = `https://en.wikipedia.org/api/rest_v1/page/summary/${slug}`;
+
+  const promise = fetch(url, {
+    headers: { 'Api-User-Agent': WIKIMEDIA_API_USER_AGENT },
+  })
+    .then(async (r) => {
+      if (!r.ok) return null;
+      const d = await r.json();
+      if (d?.type === 'disambiguation') return null;
+      const summary: WikipediaSummary = {
+        title: trimmed,
+        description: d?.description || '',
+        extract: d?.extract || '',
+        thumbnail: d?.thumbnail?.source || d?.originalimage?.source || '',
+        type: d?.type || 'standard',
+      };
+      return summary;
+    })
+    .catch(() => null)
+    .then((summary) => {
+      _summaryCache.set(trimmed, { summary, inflight: null, loaded: true });
+      evictIfOverCap();
+      return summary;
+    });
+
+  _summaryCache.set(trimmed, { summary: null, inflight: promise, loaded: false });
+  evictIfOverCap();
+  return promise;
+}
+
+/** Fetch a Wikidata SPARQL query result.
+ *
+ * Returns the parsed JSON `results.bindings` array on success; `null`
+ * (not throwing) on any failure so callers can render fallbacks
+ * silently. Kept as a thin wrapper so the audit-required UA header is
+ * applied in exactly one place.
+ */
+export async function fetchWikidataSparql<T = Record<string, { value: string }>>(
+  sparql: string,
+): Promise<T[] | null> {
+  const trimmed = (sparql || '').trim();
+  if (!trimmed) return null;
+  const url = `https://query.wikidata.org/sparql?query=${encodeURIComponent(
+    trimmed,
+  )}&format=json`;
+  try {
+    const res = await fetch(url, {
+      headers: {
+        'Api-User-Agent': WIKIMEDIA_API_USER_AGENT,
+        Accept: 'application/sparql-results+json',
+      },
+    });
+    if (!res.ok) return null;
+    const json = await res.json();
+    const bindings = json?.results?.bindings;
+    return Array.isArray(bindings) ? (bindings as T[]) : null;
+  } catch {
+    return null;
+  }
+}
+
+/** Internal: clear the shared cache. Exposed for tests only. */
+export function _resetWikimediaClientCacheForTests() {
+  _summaryCache.clear();
+}

uv.lock

@@ -82,6 +82,7 @@ dependencies = [
     { name = "cachetools" },
     { name = "cloudscraper" },
     { name = "cryptography" },
+    { name = "defusedxml" },
     { name = "fastapi" },
     { name = "feedparser" },
     { name = "httpx" },
@@ -120,6 +121,7 @@ requires-dist = [
     { name = "cachetools", specifier = "==5.5.2" },
     { name = "cloudscraper", specifier = "==1.2.71" },
     { name = "cryptography", specifier = ">=41.0.0" },
+    { name = "defusedxml", specifier = ">=0.7.1" },
     { name = "fastapi", specifier = "==0.115.12" },
     { name = "feedparser", specifier = "==6.0.10" },
     { name = "httpx", specifier = "==0.28.1" },
@@ -600,6 +602,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/a4/87/d03a718e7bfdbbebaa4b6a66ba5bb069bc00a84e5ad176d8198cc785cd42/dbus_fast-4.0.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:f6af190d8306f1bd506740c39701f5c211aa31ac660a3fcb401ebb97d33166c7", size = 1627620, upload-time = "2026-02-01T21:05:46.878Z" },
 ]
 
+[[package]]
+name = "defusedxml"
+version = "0.7.1"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/0f/d5/c66da9b79e5bdb124974bfe172b4daf3c984ebd9c2a06e2b8a4dc7331c72/defusedxml-0.7.1.tar.gz", hash = "sha256:1bb3032db185915b62d7c6209c5a8792be6a32ab2fedacc84e01b52c51aa3e69", size = 75520, upload-time = "2021-03-08T10:59:26.269Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/07/6c/aa3f2f849e01cb6a001cd8554a88d4c77c5c1a31c95bdf1cf9301e6d9ef4/defusedxml-0.7.1-py2.py3-none-any.whl", hash = "sha256:a352e7e428770286cc899e2542b6cdaedb2b4953ff269a210103ec58f6198a61", size = 25604, upload-time = "2021-03-08T10:59:24.45Z" },
+]
+
 [[package]]
 name = "deprecated"
 version = "1.3.1"