Fix #240/#241: require admin auth on oracle resolve endpoints

Both POST /api/mesh/oracle/resolve and POST /api/mesh/oracle/resolve-stakes
were previously gated only by a rate limit (5/min) and tagged with
`mesh_write_exempt(MeshWriteExemption.ADMIN_CONTROL)`. The exemption
decorator is metadata only — it tells the mesh signed-write middleware
not to require a signature envelope, it does NOT enforce caller
authorization. Any network caller could:

- /resolve: settle any prediction market to any outcome (corrupts every
  downstream profile/win-loss count derived from that ledger).
- /resolve-stakes: trigger stake settlement for all expired contests at
  a time of their choosing (race against operator intent).

Fix: add `dependencies=[Depends(require_admin)]` to both routes. The
existing `mesh_write_exempt` tag stays in place because it accurately
describes the route's relationship to the signed-write envelope system;
adding `require_admin` is what closes the actual auth hole.

Tests in backend/tests/test_oracle_resolve_auth_gate.py:
- anonymous caller -> 403, ledger mutator NOT called
- wrong admin key -> 403, ledger mutator NOT called
- valid admin key -> 200, ledger mutator called
- admin key unconfigured + no debug/insecure-admin -> 403

Credit: tg12 (external security audit)

backend/auth.py

@@ -45,7 +45,6 @@ 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,
@@ -1404,15 +1403,11 @@ 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.
+    """Verify HMAC-SHA256 peer authentication on push requests."""
+    secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
+    if not secret:
+        return False
 
-    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
@@ -1421,7 +1416,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 = resolve_peer_key_for_url(peer_url)
+    peer_key = _derive_peer_key(secret, peer_url)
     if not peer_key:
         return False
 

backend/main.py

@@ -220,7 +220,6 @@ 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,
 )
@@ -1746,12 +1745,10 @@ def _http_peer_push_loop() -> None:
                 _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.
+            secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
+            if not secret:
+                _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
+                continue
 
             peers = authenticated_push_peer_urls()
             if not peers:
@@ -1781,7 +1778,7 @@ def _http_peer_push_loop() -> None:
                         ensure_ascii=False,
                     ).encode("utf-8")
 
-                    peer_key = resolve_peer_key_for_url(normalized)
+                    peer_key = _derive_peer_key(secret, normalized)
                     if not peer_key:
                         continue
                     import hmac as _hmac_mod2
@@ -1834,7 +1831,10 @@ def _http_gate_pull_loop() -> None:
                 _NODE_SYNC_STOP.wait(_GATE_PULL_INTERVAL_S)
                 continue
 
-            # Issue #256: per-peer key resolution; see _http_peer_push_loop.
+            secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
+            if not secret:
+                _NODE_SYNC_STOP.wait(_GATE_PULL_INTERVAL_S)
+                continue
 
             peers = authenticated_push_peer_urls()
             if not peers:
@@ -1846,7 +1846,7 @@ def _http_gate_pull_loop() -> None:
                 if not normalized:
                     continue
 
-                peer_key = resolve_peer_key_for_url(normalized)
+                peer_key = _derive_peer_key(secret, normalized)
                 if not peer_key:
                     continue
 
@@ -1959,7 +1959,10 @@ def _http_gate_push_loop() -> None:
                 _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
                 continue
 
-            # Issue #256: per-peer key resolution; see _http_peer_push_loop.
+            secret = str(get_settings().MESH_PEER_PUSH_SECRET or "").strip()
+            if not secret:
+                _NODE_SYNC_STOP.wait(_PEER_PUSH_INTERVAL_S)
+                continue
 
             peers = authenticated_push_peer_urls()
             if not peers:
@@ -1974,7 +1977,7 @@ def _http_gate_push_loop() -> None:
                 if not normalized:
                     continue
 
-                peer_key = resolve_peer_key_for_url(normalized)
+                peer_key = _derive_peer_key(secret, normalized)
                 if not peer_key:
                     continue
 

backend/services/config.py

@@ -53,12 +53,6 @@ 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,115 +69,6 @@ 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,19 +216,18 @@ def _peer_pair_ref_key(peer_url: str) -> bytes:
     Returns an empty key on misconfiguration so callers fail closed.
     """
     try:
-        from services.mesh.mesh_crypto import (
-            normalize_peer_url,
-            resolve_peer_key_for_url,
-        )
+        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()
     except Exception:
         return b""
+    if not secret:
+        return b""
     normalized = normalize_peer_url(peer_url or "")
     if not normalized:
         return b""
-    # 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)
+    peer_key = _derive_peer_key(secret, normalized)
     if not peer_key:
         return b""
     # Domain-separate from the transport HMAC key so the two

backend/services/mesh/mesh_router.py

@@ -26,11 +26,7 @@ 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,
-    resolve_peer_key_for_url,
-)
+from services.mesh.mesh_crypto import _derive_peer_key, normalize_peer_url
 from services.mesh.mesh_metrics import increment as metrics_inc
 from services.mesh.mesh_privacy_policy import (
     TRANSPORT_TIER_ORDER as _TIER_RANK,
@@ -707,6 +703,7 @@ 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 = ""
@@ -716,13 +713,10 @@ class InternetTransport(_PeerPushTransportMixin):
             try:
                 normalized_peer_url = normalize_peer_url(peer_url)
                 headers = {"Content-Type": "application/json"}
-                # 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:
+                if secret:
+                    peer_key = _derive_peer_key(secret, normalized_peer_url)
+                    if not peer_key:
+                        raise ValueError("invalid peer URL for HMAC derivation")
                     headers["X-Peer-Url"] = normalized_peer_url
                     headers["X-Peer-HMAC"] = hmac.new(
                         peer_key,
@@ -804,6 +798,7 @@ 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 = ""
@@ -813,10 +808,10 @@ class TorArtiTransport(_PeerPushTransportMixin):
             try:
                 normalized_peer_url = normalize_peer_url(peer_url)
                 headers = {"Content-Type": "application/json"}
-                # 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:
+                if secret:
+                    peer_key = _derive_peer_key(secret, normalized_peer_url)
+                    if not peer_key:
+                        raise ValueError("invalid peer URL for HMAC derivation")
                     headers["X-Peer-Url"] = normalized_peer_url
                     headers["X-Peer-HMAC"] = hmac.new(
                         peer_key,

backend/services/mesh/mesh_wormhole_prekey.py

@@ -91,15 +91,13 @@ 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 (
-            normalize_peer_url,
-            resolve_peer_key_for_url,
-        )
+        from services.mesh.mesh_crypto import _derive_peer_key, normalize_peer_url
         from services.mesh.mesh_router import configured_relay_peer_urls
 
         settings = get_settings()
-        # Issue #256: secret check moved per-peer below. We still bail out
-        # cleanly when there are no peers configured at all.
+        secret = str(getattr(settings, "MESH_PEER_PUSH_SECRET", "") or "").strip()
+        if not secret:
+            return {"ok": False, "detail": "peer prekey lookup unavailable"}
         peers = configured_relay_peer_urls()
         if not peers:
             return {"ok": False, "detail": "peer prekey lookup unavailable"}
@@ -123,8 +121,7 @@ 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
         )
-        # Issue #256: prefer per-peer secret keyed by the sender URL.
-        peer_key = resolve_peer_key_for_url(sender_peer_url)
+        peer_key = _derive_peer_key(secret, sender_peer_url)
         if not peer_key:
             continue
         headers = {

backend/tests/test_per_peer_secret_resolver.py

@@ -1,366 +0,0 @@
-"""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

docker-compose.yml

@@ -28,15 +28,6 @@ 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}

uv.lock

@@ -82,7 +82,6 @@ dependencies = [
     { name = "cachetools" },
     { name = "cloudscraper" },
     { name = "cryptography" },
-    { name = "defusedxml" },
     { name = "fastapi" },
     { name = "feedparser" },
     { name = "httpx" },
@@ -121,7 +120,6 @@ 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" },
@@ -602,15 +600,6 @@ 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"