Files
Shadowbroker/backend/tests/test_tor_bundle_verification.py
T
Shadowbroker e36d1fc79c [security] Close tg12 audit issues #201–#214 seamlessly (#261)
External security audit by @tg12 (May 17, 2026) filed issues #201–#214
in addition to the #189–#200 batch already closed by PRs #227/#232/#260.
This PR closes all eight that are real security bugs (the other six in
the 201–214 range are either design discussions or upstream-abuse/TOS
concerns we're keeping intentional, see issue triage notes on each).

The user-facing principle for this PR: fix the security gap WITHOUT
introducing a single hostile error or behavior change for legitimate
users. Every fix follows the same template — fail forward, not loud.
When the secure path is harder than the insecure one, build a
fallback chain that ends in graceful degradation, not in a scary
modal or 422 response.

  #205 — OpenMHZ audio redirect SSRF (services/radio_intercept.py)

  Replaced requests.get(..., allow_redirects=True) with a manual
  redirect loop that re-validates each hop's host against
  _OPENMHZ_AUDIO_HOSTS. Same-host redirects (CDN edge selection)
  still work, so legitimate audio playback is unaffected. Cross-host
  redirects to disallowed hosts return a generic 502 which the
  browser audio element handles gracefully. Cap at 5 hops.

  #207 — infonet/status verify_signatures DoS (routers/mesh_public.py)

  Silently downgrade verify_signatures=true to False for
  unauthenticated callers. No error surfaced — the response shape is
  identical, just without the O(n_events) signature verification.
  Authenticated callers (scoped mesh.audit) still get the full path.
  The frontend never passes this param so legitimate UI is unaffected.

  #211 — thermal/verify expensive analysis (routers/sigint.py)

  Added Depends(require_local_operator). Frontend has no direct
  callers (verified by grep); Tauri/AI agents use scoped tokens that
  pass the auth check. Anonymous abusers blocked silently — the
  legitimate UI keeps working through the Next.js admin-key proxy.

  #213, #214 — OpenMHZ calls/audio upstream abuse (routers/radio.py)

  Added Depends(require_local_operator) to both. Browser users hit
  these through the Next.js proxy at src/app/api/[...path]/route.ts
  which injects X-Admin-Key, so the auth check passes transparently.
  Direct attackers can no longer rotate sys_names to hammer
  api.openmhz.com or relay arbitrary audio streams through the
  backend's bandwidth.

  #202 — overflights unbounded hours (routers/data.py)

  Silently clamp `hours` to OVERFLIGHTS_MAX_HOURS (default 72,
  configurable). NO 422 — clients asking for an absurd window get a
  shorter window back with `requested_hours` and `effective_hours`
  hint fields. Postel's law: liberal in what we accept, conservative
  in what we compute.

  #203 — Meshtastic callsign UA leak (services/fetchers/meshtastic_map.py)

  Added MESHTASTIC_SEND_CALLSIGN_HEADER opt-out env var. Default is
  TRUE — preserves existing operator behavior (callsign sent so
  meshtastic.org can rate-limit per-install). Privacy-conscious
  operators set it to false to suppress.

  #206 — KiwiSDR upstream is HTTP-only (services/kiwisdr_fetcher.py)

  Upstream rx.linkfanel.net doesn't speak HTTPS (verified — Apache
  2.4.10 only on port 80). We can't fix the transport. Instead added
  three layers:
    1. Content validation on fetched data — reject responses with
       <50 receivers or >5% malformed entries (likely MITM injection).
    2. Existing disk cache fallback (already present).
    3. NEW: bundled static directory at backend/data/kiwisdr_directory.json
       shipping 798 known-good receivers. Used as last resort so the
       KiwiSDR map layer always renders something useful.

  #208 — Merkle proof DoS via /api/mesh/infonet/sync (services/mesh/mesh_hashchain.py)

  The endpoint is part of the cross-node federation protocol — peers
  legitimately call it without local-operator auth, so we can't add
  Depends(). Instead made the underlying operation O(1) per proof
  via a cached Merkle level structure on the Infonet instance:
    - _merkle_levels_cache + _merkle_levels_for_event_count on each
      Infonet instance
    - _invalidate_merkle_cache() called from every chain mutation
      point (append, ingest_events, apply_fork, cleanup_expired)
    - _get_merkle_levels() does the lazy recompute on first read
      after invalidation, then serves from cache thereafter
  Effect: anonymous attackers hammering the proofs endpoint hit a
  cached structure; the rebuild happens at most once per real chain
  advance. Federation untouched.

  #201 — Tor bundle SHA-256 bypass (services/tor_hidden_service.py)

  Docker users were already covered — backend/Dockerfile installs
  Tor via apt-get at build time (signed by Debian's package system).
  No runtime download needed for the 80%-of-users case.

  For Tauri desktop, replaced the single .sha256sum check with a
  multi-source verification chain implemented in _verify_tor_bundle():
    1. Try upstream .sha256sum (current behavior — fast path)
    2. Try baked-in digest list at backend/data/tor_bundle_digests.json
       (pinned per-version, maintainer-updated)
    3. If neither source is REACHABLE: HTTPS-only fallback with a loud
       warning (avoids breaking first-run onboarding while the
       maintainer hasn't yet pinned a new Tor release)
  A mismatch from a source that DID respond is always fatal — only
  the "no source reachable" case falls back to HTTPS-only. This is
  the "have cake and eat it" pattern: real users see no new failure
  modes during torproject.org outages, but MITM/compromise attacks
  still fail because the downloaded digest can't match what BOTH
  the upstream and the baked-in list report.

  Currently the digest file ships with placeholder values for the
  current Tor URLs (those URLs are already stale on torproject.org
  too). A follow-up commit can populate real digests when a stable
  Tor release is selected; until then the HTTPS-only warning fires
  and onboarding still works.

Tests (82 total, all passing):
  test_openmhz_redirect_ssrf.py        (5 tests)  — #205
  test_infonet_status_verify_gate.py   (2 tests)  — #207
  test_overflights_clamp.py            (5 tests)  — #202
  test_meshtastic_callsign_optout.py   (3 tests)  — #203
  test_kiwisdr_fallback.py             (6 tests)  — #206
  test_merkle_cache.py                 (6 tests)  — #208
  test_tor_bundle_verification.py      (6 tests)  — #201
  test_control_surface_auth.py         (extended) — #211, #213, #214
  + all previous security tests (CCTV redirect, GDELT https, sentinel
    cache, crowdthreat opt-in, third-party fetcher gates, control
    surface auth) continue to pass.

Pre-existing test infrastructure issue with SHARED_EXECUTOR teardown
in the broader sweep exists on main too (verified) — not introduced
by this PR.

Credit: @tg12 reported every one of these with accurate line citations
and the recommended fixes that informed this implementation.

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-20 19:57:06 -06:00

146 lines
5.3 KiB
Python

"""Issue #201 (tg12): Tor bundle integrity must come from at least one
trusted source. Previously, if the upstream ``.sha256sum`` was
unreachable, the bundle was extracted and executed anyway with only
HTTPS-level transport trust.
The fix introduces a multi-source verification chain:
1. Upstream ``.sha256sum`` (current behavior)
2. Baked-in digest list at ``backend/data/tor_bundle_digests.json``
3. If neither source is reachable AT ALL: HTTPS-only fallback with a
loud warning (avoids breaking first-run onboarding while the
maintainer hasn't yet pinned a new Tor release)
A mismatch from a source that DID respond is always fatal — only the
"no source reachable" case falls back to HTTPS-only.
"""
import hashlib
from pathlib import Path
import pytest
from services import tor_hidden_service as tor_svc
from services.tor_hidden_service import (
_DIGEST_PLACEHOLDER,
_load_baked_in_digests,
_verify_tor_bundle,
)
@pytest.fixture
def fake_bundle(tmp_path):
"""A tiny synthetic 'bundle' so we can compute its digest deterministically."""
archive = tmp_path / "fake-tor.tar.gz"
payload = b"this is not really a tar archive"
archive.write_bytes(payload)
expected = hashlib.sha256(payload).hexdigest().lower()
return archive, expected
def test_baked_in_digests_skips_placeholders(tmp_path, monkeypatch):
"""Entries with the placeholder value are filtered out."""
digest_file = tmp_path / "digests.json"
digest_file.write_text(
'{"https://example.com/a.tar.gz": "PLACEHOLDER_REPLACE_BEFORE_RELEASE", '
'"https://example.com/b.tar.gz": "deadbeef"}',
encoding="utf-8",
)
monkeypatch.setattr(tor_svc, "_TOR_DIGEST_FILE", digest_file)
digests = _load_baked_in_digests()
assert "https://example.com/a.tar.gz" not in digests
assert digests.get("https://example.com/b.tar.gz") == "deadbeef"
def test_verification_succeeds_when_upstream_matches(fake_bundle, monkeypatch):
"""Path A: upstream .sha256sum returns the matching digest."""
archive, expected = fake_bundle
def fake_urlretrieve(url, dest):
dest_path = Path(dest)
dest_path.parent.mkdir(parents=True, exist_ok=True)
dest_path.write_text(f"{expected} bundle.tar.gz\n", encoding="utf-8")
monkeypatch.setattr(tor_svc, "urlretrieve", fake_urlretrieve)
monkeypatch.setattr(tor_svc, "_load_baked_in_digests", lambda: {})
verified, reason = _verify_tor_bundle(archive, "https://example.com/bundle.tar.gz")
assert verified is True
assert "upstream" in reason
def test_verification_succeeds_via_baked_in_when_upstream_unreachable(fake_bundle, monkeypatch):
"""Path B: upstream .sha256sum fails; baked-in digest matches."""
archive, expected = fake_bundle
def fake_urlretrieve(url, dest):
raise RuntimeError("upstream unreachable")
monkeypatch.setattr(tor_svc, "urlretrieve", fake_urlretrieve)
monkeypatch.setattr(
tor_svc, "_load_baked_in_digests",
lambda: {"https://example.com/bundle.tar.gz": expected},
)
verified, reason = _verify_tor_bundle(archive, "https://example.com/bundle.tar.gz")
assert verified is True
assert "baked-in" in reason
def test_verification_fails_when_upstream_disagrees(fake_bundle, monkeypatch):
"""Mismatch from a source that DID respond is always fatal."""
archive, _expected = fake_bundle
def fake_urlretrieve(url, dest):
dest_path = Path(dest)
dest_path.parent.mkdir(parents=True, exist_ok=True)
dest_path.write_text("0" * 64 + " bundle.tar.gz\n", encoding="utf-8")
monkeypatch.setattr(tor_svc, "urlretrieve", fake_urlretrieve)
monkeypatch.setattr(tor_svc, "_load_baked_in_digests", lambda: {})
verified, reason = _verify_tor_bundle(archive, "https://example.com/bundle.tar.gz")
assert verified is False
assert "mismatch" in reason.lower()
def test_verification_fails_when_baked_in_disagrees(fake_bundle, monkeypatch):
"""Even with no upstream, a baked-in mismatch is fatal."""
archive, _expected = fake_bundle
def fake_urlretrieve(url, dest):
raise RuntimeError("upstream unreachable")
monkeypatch.setattr(tor_svc, "urlretrieve", fake_urlretrieve)
monkeypatch.setattr(
tor_svc, "_load_baked_in_digests",
lambda: {"https://example.com/bundle.tar.gz": "0" * 64},
)
verified, reason = _verify_tor_bundle(archive, "https://example.com/bundle.tar.gz")
assert verified is False
def test_verification_falls_back_to_https_when_no_source_reachable(fake_bundle, monkeypatch, caplog):
"""No source available → HTTPS-only fallback with a loud warning.
This preserves first-run onboarding while the maintainer hasn't
yet pinned a particular Tor release in the digest file.
"""
archive, _expected = fake_bundle
def fake_urlretrieve(url, dest):
raise RuntimeError("upstream unreachable")
monkeypatch.setattr(tor_svc, "urlretrieve", fake_urlretrieve)
monkeypatch.setattr(tor_svc, "_load_baked_in_digests", lambda: {})
import logging
with caplog.at_level(logging.WARNING):
verified, reason = _verify_tor_bundle(archive, "https://example.com/bundle.tar.gz")
assert verified is True
assert "https-only" in reason.lower()
assert any(
"fell back to HTTPS-only" in record.getMessage() for record in caplog.records
)