Close tg12 outbound audit (#348-#366): operator UA, opt-ins, docs

- User-Agent is per-install handle only (no Shadowbroker product token)
- LiveUAMap: Windows UI consent when enabling Global Incidents; env override
- Meshtastic callsign upstream header off by default (opt-in true)
- Expanded docs/OUTBOUND_DATA.md and README link for CCTV, basemap, Broadcastify

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/.env.example

@@ -31,11 +31,9 @@ AIS_API_KEY=              # https://aisstream.io/ — free tier WebSocket key
 # Requires MESH_DEBUG_MODE=true; do not enable this for ordinary use.
 # ALLOW_INSECURE_ADMIN=false
 
-# Per-install operator handle. Round 7a: every outbound third-party API
-# call (Wikipedia, Wikidata, Nominatim, GDELT, OpenMHz, Broadcastify,
-# weather.gov, NUFORC, etc.) includes this handle in the User-Agent so
-# upstreams can rate-limit / contact the specific install instead of
-# treating every Shadowbroker user as one entity.
+# Per-install operator handle. Round 7a: outbound third-party API calls send
+# this handle as the User-Agent (e.g. operator-7f3a92), not a shared app name,
+# so upstreams rate-limit one install instead of blocking every user.
 #
 # Default empty -> a stable pseudonymous handle (e.g. "operator-7f3a92") is
 # auto-generated on first run and persisted to backend/data/operator_handle.json.
@@ -43,10 +41,8 @@ AIS_API_KEY=              # https://aisstream.io/ — free tier WebSocket key
 # set it here. Special characters are sanitized to dashes.
 # OPERATOR_HANDLE=
 
-# Default outbound User-Agent for all third-party HTTP fetchers. Operators
-# who run a public relay and want a completely custom UA can set this; it
-# bypasses the per-operator helper entirely. Most installs should leave it
-# unset and use OPERATOR_HANDLE instead.
+# Full User-Agent override (replaces the operator handle entirely). Rare;
+# most installs should use OPERATOR_HANDLE only.
 # SHADOWBROKER_USER_AGENT=
 
 # Nominatim-specific User-Agent override (OSM usage policy). Leave unset to
@@ -122,12 +118,16 @@ AIS_API_KEY=              # https://aisstream.io/ — free tier WebSocket key
 # can identify per-install traffic instead of aggregated "ShadowBroker" hits.
 # Leave blank to send a generic UA. If you set MESHTASTIC_OPERATOR_CALLSIGN,
 # it is included in outbound headers to meshtastic.org by default so they
-# can rate-limit per-operator. Set MESHTASTIC_SEND_CALLSIGN_HEADER=false to
-# suppress the callsign while still using it locally (e.g. for APRS).
+# can rate-limit per-operator. Callsign is NOT sent upstream unless you opt in.
 # MESHTASTIC_OPERATOR_CALLSIGN=
-# MESHTASTIC_SEND_CALLSIGN_HEADER=true
+# MESHTASTIC_SEND_CALLSIGN_HEADER=false
 # MESH_MQTT_PSK=                       # hex-encoded, empty = default LongFast key
 
+# LiveUAMap Playwright scraper (#348). Linux/macOS: on by default when Global
+# Incidents layer is active. Windows: off until the operator enables Global
+# Incidents in the UI (consent dialog) or sets SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER=true.
+# SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER=false forces off on all platforms.
+
 # ── Mesh / Reticulum (RNS) ─────────────────────────────────────
 # Full-node / participant-node posture for public Infonet sync.
 # MESH_NODE_MODE=participant      # participant | relay | perimeter

backend/routers/data.py

@@ -30,6 +30,10 @@ class LayerUpdate(BaseModel):
     layers: dict[str, bool]
 
 
+class LiveUamapOptInUpdate(BaseModel):
+    opted_in: bool
+
+
 _LAST_VIEWPORT_UPDATE: tuple | None = None
 _LAST_VIEWPORT_UPDATE_TS = 0.0
 _VIEWPORT_UPDATE_LOCK = threading.Lock()
@@ -386,6 +390,38 @@ async def update_viewport(vp: ViewportUpdate, request: Request):  # noqa: ARG001
     return {"status": "ok"}
 
 
+@router.get("/api/liveuamap/scraper-status", dependencies=[Depends(require_local_operator)])
+async def api_liveuamap_scraper_status():
+    """Whether LiveUAMap Playwright may run (Windows needs UI opt-in unless env forces)."""
+    from services.liveuamap_settings import liveuamap_scraper_status
+
+    return liveuamap_scraper_status()
+
+
+@router.post("/api/liveuamap/scraper-opt-in", dependencies=[Depends(require_local_operator)])
+@limiter.limit("10/minute")
+async def api_liveuamap_scraper_opt_in(body: LiveUamapOptInUpdate, request: Request):
+    """Persist operator consent for LiveUAMap scraper (#348)."""
+    from services.liveuamap_settings import liveuamap_scraper_status, set_liveuamap_ui_opt_in
+
+    set_liveuamap_ui_opt_in(body.opted_in)
+    if body.opted_in:
+        from services.fetchers._store import is_any_active
+
+        if is_any_active("global_incidents"):
+            threading.Thread(target=_run_liveuamap_refresh, daemon=True).start()
+    return liveuamap_scraper_status()
+
+
+def _run_liveuamap_refresh() -> None:
+    try:
+        from services.fetchers.geo import update_liveuamap
+
+        update_liveuamap()
+    except Exception as e:
+        logger.warning("LiveUAMap refresh after opt-in failed: %s", e)
+
+
 @router.post("/api/layers", dependencies=[Depends(require_local_operator)])
 @limiter.limit("30/minute")
 async def update_layers(update: LayerUpdate, request: Request):

backend/scripts/convert_power_plants.py

@@ -29,7 +29,7 @@ def main() -> None:
         from services.network_utils import outbound_user_agent
         ua = outbound_user_agent("release-script-power-plants")
     except Exception:
-        ua = "Shadowbroker/0.9 (release-script-power-plants; +https://github.com/BigBodyCobain/Shadowbroker/issues)"
+        ua = "operator-release-script (purpose: power-plants)"
     req = urllib.request.Request(CSV_URL, headers={"User-Agent": ua})
     with urllib.request.urlopen(req, timeout=60) as resp:
         raw = resp.read().decode("utf-8")

backend/services/fetchers/aircraft_database.py

@@ -38,8 +38,6 @@ _S3_NS = "{http://s3.amazonaws.com/doc/2006-03-01/}"
 _REFRESH_INTERVAL_S = 5 * 24 * 3600
 _LIST_TIMEOUT_S = 30
 _DOWNLOAD_TIMEOUT_S = 600
-from services.network_utils import DEFAULT_USER_AGENT as _USER_AGENT
-
 _lock = threading.RLock()
 _aircraft_by_hex: dict[str, dict[str, str]] = {}
 _last_refresh = 0.0

backend/services/fetchers/geo.py

@@ -20,17 +20,9 @@ def _env_flag(name: str) -> str:
 
 
 def liveuamap_scraper_enabled() -> bool:
-    """Return whether the Playwright-based LiveUAMap scraper should run.
+    from services.liveuamap_settings import liveuamap_scraper_enabled as _enabled
 
-    It is useful enrichment, but it starts a browser/Node driver and must not be
-    allowed to destabilize Windows local startup.
-    """
-    setting = _env_flag("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER")
-    if setting in {"1", "true", "yes", "on"}:
-        return True
-    if setting in {"0", "false", "no", "off"}:
-        return False
-    return os.name != "nt"
+    return _enabled()
 
 
 # ---------------------------------------------------------------------------
@@ -210,10 +202,17 @@ def update_liveuamap():
     if not is_any_active("global_incidents"):
         return
     if not liveuamap_scraper_enabled():
-        logger.info(
-            "Liveuamap scraper disabled for this runtime; set "
-            "SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER=1 to opt in."
-        )
+        from services.liveuamap_settings import liveuamap_requires_ui_opt_in
+
+        if liveuamap_requires_ui_opt_in():
+            logger.info(
+                "Liveuamap scraper disabled: enable Global Incidents in the UI to "
+                "consent, or set SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER=1."
+            )
+        else:
+            logger.info(
+                "Liveuamap scraper disabled; set SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER=1 to opt in."
+            )
         return
     logger.info("Running scheduled Liveuamap scraper...")
     try:

backend/services/fetchers/meshtastic_map.py

@@ -188,8 +188,8 @@ def fetch_meshtastic_nodes():
         callsign = ""
 
     send_callsign_header = str(
-        _os.environ.get("MESHTASTIC_SEND_CALLSIGN_HEADER", "true")
-    ).strip().lower() not in {"0", "false", "no", "off", ""}
+        _os.environ.get("MESHTASTIC_SEND_CALLSIGN_HEADER", "false")
+    ).strip().lower() in {"1", "true", "yes", "on"}
 
     # Round 7a: outbound_user_agent already includes the per-install handle.
     # The optional Meshtastic callsign is appended as additional context so

backend/services/fetchers/route_database.py

@@ -30,8 +30,6 @@ _AIRPORTS_URL = "https://vrs-standing-data.adsb.lol/airports.csv.gz"
 _REFRESH_INTERVAL_S = 5 * 24 * 3600
 _HTTP_TIMEOUT_S = 60
 
-from services.network_utils import DEFAULT_USER_AGENT as _USER_AGENT
-
 _lock = threading.RLock()
 _routes_by_callsign: dict[str, dict[str, Any]] = {}
 _airports_by_icao: dict[str, dict[str, Any]] = {}

backend/services/liveuamap_scraper.py

@@ -27,8 +27,15 @@ def fetch_liveuamap():
         browser = p.chromium.launch(
             headless=True, args=["--disable-blink-features=AutomationControlled"]
         )
+        from services.network_utils import outbound_user_agent
+
+        # Per-install handle (no shared Shadowbroker product token). Stealth remains
+        # for Turnstile; see docs/OUTBOUND_DATA.md #348.
+        playwright_ua = (
+            f"Mozilla/5.0 (compatible; {outbound_user_agent('liveuamap')})"
+        )
         context = browser.new_context(
-            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+            user_agent=playwright_ua,
             viewport={"width": 1920, "height": 1080},
             color_scheme="dark",
         )

backend/services/liveuamap_settings.py

@@ -0,0 +1,73 @@
+"""LiveUAMap Playwright scraper opt-in (#348) — UI consent on Windows."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import threading
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_OPT_IN_FILE = Path(__file__).resolve().parent.parent / "data" / "liveuamap_scraper_opt_in.json"
+_OPT_IN_LOCK = threading.Lock()
+
+
+def _env_flag(name: str) -> str:
+    return str(os.getenv(name, "")).strip().lower()
+
+
+def liveuamap_requires_ui_opt_in() -> bool:
+    """Windows local installs need explicit consent before Playwright contacts LiveUAMap."""
+    return os.name == "nt"
+
+
+def get_liveuamap_ui_opt_in() -> bool:
+    if not _OPT_IN_FILE.exists():
+        return False
+    try:
+        payload = json.loads(_OPT_IN_FILE.read_text(encoding="utf-8"))
+        return bool(payload.get("opted_in"))
+    except (OSError, json.JSONDecodeError, TypeError) as e:
+        logger.warning("LiveUAMap opt-in file unreadable: %s", e)
+        return False
+
+
+def set_liveuamap_ui_opt_in(opted_in: bool) -> None:
+    _OPT_IN_FILE.parent.mkdir(parents=True, exist_ok=True)
+    with _OPT_IN_LOCK:
+        _OPT_IN_FILE.write_text(
+            json.dumps({"opted_in": bool(opted_in)}, indent=2),
+            encoding="utf-8",
+        )
+
+
+def liveuamap_scraper_enabled() -> bool:
+    """Whether the Playwright LiveUAMap scraper may run on this backend."""
+    setting = _env_flag("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER")
+    if setting in {"1", "true", "yes", "on"}:
+        return True
+    if setting in {"0", "false", "no", "off"}:
+        return False
+    if not liveuamap_requires_ui_opt_in():
+        return True
+    return get_liveuamap_ui_opt_in()
+
+
+def liveuamap_scraper_status() -> dict[str, Any]:
+    setting = _env_flag("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER")
+    env_override = None
+    if setting in {"1", "true", "yes", "on"}:
+        env_override = "on"
+    elif setting in {"0", "false", "no", "off"}:
+        env_override = "off"
+    ui_opted_in = get_liveuamap_ui_opt_in()
+    requires = liveuamap_requires_ui_opt_in()
+    return {
+        "platform_requires_opt_in": requires,
+        "ui_opted_in": ui_opted_in,
+        "scraper_enabled": liveuamap_scraper_enabled(),
+        "env_override": env_override,
+    }

backend/services/mesh/mesh_wormhole_prekey.py

@@ -234,12 +234,12 @@ def _fetch_dm_prekey_bundle_from_public_lookup(lookup_token: str) -> dict[str, A
         # Generic UA: any peer-facing crypto request should not carry a
         # fork-specific identifier — that turns prekey lookups into a
         # software-fingerprinting beacon.
-        from services.network_utils import DEFAULT_USER_AGENT
+        from services.network_utils import default_user_agent
         request = urllib.request.Request(
             f"{normalized_peer_url}/api/mesh/dm/prekey-bundle?{encoded}",
             headers={
                 "Accept": "application/json",
-                "User-Agent": DEFAULT_USER_AGENT,
+                "User-Agent": default_user_agent(),
             },
             method="GET",
         )

backend/services/network_utils.py

@@ -34,9 +34,9 @@ _session.mount("http://", HTTPAdapter(max_retries=_retry, pool_maxsize=10))
 # upstream's only recourse was to block "Shadowbroker" as a whole — which
 # would take out every other install too.
 #
-# Fix: give each install a stable pseudonymous handle and include it in
-# the User-Agent. Now an upstream can rate-limit or block the offending
-# operator without affecting anyone else.
+# Fix: give each install a stable pseudonymous handle used as the entire
+# User-Agent product token (no shared "Shadowbroker" label). Upstreams see
+# ``operator-7f3a92`` (or ``OPERATOR_HANDLE``), not one monolithic app name.
 #
 # The handle:
 #
@@ -51,7 +51,6 @@ _session.mount("http://", HTTPAdapter(max_retries=_retry, pool_maxsize=10))
 # - Is NEVER mixed into mesh / Wormhole / Infonet identity. This layer is
 #   strictly for public third-party API attribution.
 
-_SHADOWBROKER_VERSION = "0.9"
 _OPERATOR_HANDLE_FILE = (
     Path(__file__).parent.parent / "data" / "operator_handle.json"
 )
@@ -175,41 +174,21 @@ def _normalize_handle(raw: str) -> str:
     return safe[:48] if safe else "anonymous"
 
 
-_CONTACT_URL = "https://github.com/BigBodyCobain/Shadowbroker/issues"
-
-
 def outbound_user_agent(purpose: str = "") -> str:
     """Build a User-Agent for an outbound third-party HTTP request.
 
-    Returns something like::
+    Returns the per-install handle only, e.g. ``operator-7f3a92`` or
+    ``operator-7f3a92 (purpose: wikipedia)``. No shared project name — so
+    upstream abuse teams cannot block every install with one ``Shadowbroker``
+    rule.
 
-        Shadowbroker/0.9 (operator: operator-7f3a92; purpose: wikipedia;
-         +https://github.com/BigBodyCobain/Shadowbroker/issues)
-
-    The ``purpose`` is optional but recommended — it tells the upstream
-    what feature of ours is making the call (``wikipedia``, ``openmhz``,
-    ``nominatim``, etc.), which makes their logs and our complaints
-    actionable.
-
-    Every outbound call in the backend that previously sent a custom
-    User-Agent should call this helper instead. Centralizing here means:
-      - one place to change the contact URL,
-      - one place to bump the version on release,
-      - one place a Wikimedia / OpenMHz operator can reach to ask for
-        the project to back off, with a per-install handle so they can
-        target the specific install instead of the project as a whole.
+    Set ``SHADOWBROKER_USER_AGENT`` to override the entire string if needed.
     """
     handle = get_operator_handle()
     if purpose:
         purpose_clean = _normalize_handle(purpose)
-        return (
-            f"Shadowbroker/{_SHADOWBROKER_VERSION} "
-            f"(operator: {handle}; purpose: {purpose_clean}; +{_CONTACT_URL})"
-        )
-    return (
-        f"Shadowbroker/{_SHADOWBROKER_VERSION} "
-        f"(operator: {handle}; +{_CONTACT_URL})"
-    )
+        return f"{handle} (purpose: {purpose_clean})"
+    return handle
 
 
 def _reset_operator_handle_cache_for_tests() -> None:
@@ -220,19 +199,13 @@ def _reset_operator_handle_cache_for_tests() -> None:
         _OPERATOR_HANDLE_CACHE = ""
 
 
-# Default outbound User-Agent. Retained for backwards compatibility with
-# call sites that haven't been migrated to ``outbound_user_agent()`` yet.
-# Operators who want full per-install attribution should set the
-# ``OPERATOR_HANDLE`` setting and migrate call sites incrementally.
-#
-# Operators who run a public-facing relay can also override the whole UA
-# string via the ``SHADOWBROKER_USER_AGENT`` env var. That override
-# completely bypasses the per-operator helper; only use it if you know
-# what you're doing.
-DEFAULT_USER_AGENT = os.environ.get(
-    "SHADOWBROKER_USER_AGENT",
-    f"Shadowbroker/{_SHADOWBROKER_VERSION}",
-)
+def default_user_agent() -> str:
+    """Default User-Agent for ``fetch_with_curl`` and legacy call sites."""
+    custom = (os.environ.get("SHADOWBROKER_USER_AGENT") or "").strip()
+    if custom:
+        return custom
+    return outbound_user_agent()
+
 
 # Find bash for curl fallback — Git bash's curl has the TLS features
 # needed to pass CDN fingerprint checks (brotli, zstd, libpsl)
@@ -288,7 +261,7 @@ def fetch_with_curl(url, method="GET", json_data=None, timeout=15, headers=None,
     both Python requests and the barebones Windows system curl.
     """
     default_headers = {
-        "User-Agent": DEFAULT_USER_AGENT,
+        "User-Agent": default_user_agent(),
     }
     if headers:
         default_headers.update(headers)

backend/tests/test_liveuamap_scraper_opt_in.py

@@ -0,0 +1,45 @@
+"""LiveUAMap scraper UI opt-in on Windows (#348)."""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from services import liveuamap_settings as settings
+
+
+@pytest.fixture
+def opt_in_file(tmp_path, monkeypatch):
+    path = tmp_path / "liveuamap_scraper_opt_in.json"
+    monkeypatch.setattr(settings, "_OPT_IN_FILE", path)
+    return path
+
+
+def test_windows_defaults_off_without_opt_in(monkeypatch, opt_in_file):
+    monkeypatch.setattr(settings.os, "name", "nt")
+    monkeypatch.delenv("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER", raising=False)
+    assert settings.liveuamap_scraper_enabled() is False
+    assert settings.liveuamap_requires_ui_opt_in() is True
+
+
+def test_windows_opt_in_enables_scraper(monkeypatch, opt_in_file):
+    monkeypatch.setattr(settings.os, "name", "nt")
+    monkeypatch.delenv("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER", raising=False)
+    settings.set_liveuamap_ui_opt_in(True)
+    assert settings.liveuamap_scraper_enabled() is True
+    assert json.loads(opt_in_file.read_text())["opted_in"] is True
+
+
+def test_linux_enabled_without_opt_in(monkeypatch, opt_in_file):
+    monkeypatch.setattr(settings.os, "name", "posix")
+    monkeypatch.delenv("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER", raising=False)
+    assert settings.liveuamap_requires_ui_opt_in() is False
+    assert settings.liveuamap_scraper_enabled() is True
+
+
+def test_env_force_off_overrides_ui_opt_in(monkeypatch, opt_in_file):
+    monkeypatch.setattr(settings.os, "name", "nt")
+    monkeypatch.setenv("SHADOWBROKER_ENABLE_LIVEUAMAP_SCRAPER", "false")
+    settings.set_liveuamap_ui_opt_in(True)
+    assert settings.liveuamap_scraper_enabled() is False

backend/tests/test_meshtastic_callsign_optout.py

@@ -1,56 +1,27 @@
-"""Issue #203 (tg12): meshtastic_map.py was unconditionally including
-``MESHTASTIC_OPERATOR_CALLSIGN`` in the outbound User-Agent header,
-which contradicted the README's "no user data transmitted" claim.
-
-The fix preserves the existing default behavior (callsign sent — that's
-what operators who configured the variable expected) but adds an
-opt-out env var ``MESHTASTIC_SEND_CALLSIGN_HEADER=false`` for
-privacy-conscious operators.
-"""
-import importlib
-import sys
+"""Issue #350: Meshtastic callsign in outbound UA is opt-in, not default."""
+import os
 
 import pytest
 
 
-def _reload_meshtastic_module():
-    """Reload meshtastic_map so settings are re-read on demand."""
-    if "services.fetchers.meshtastic_map" in sys.modules:
-        del sys.modules["services.fetchers.meshtastic_map"]
-    return importlib.import_module("services.fetchers.meshtastic_map")
+def _send_callsign_header_from_env() -> bool:
+    raw = str(os.environ.get("MESHTASTIC_SEND_CALLSIGN_HEADER", "false")).strip().lower()
+    return raw in {"1", "true", "yes", "on"}
 
 
-def test_default_behavior_includes_callsign(monkeypatch):
-    """Operators who set the callsign and don't change anything else
-    keep their existing behavior (callsign sent in UA)."""
-    # We test the UA construction logic by exercising the same branches
-    # the fetcher uses. Direct fetch isn't run because it makes a real
-    # network call — we just verify the env-var-driven decision.
-    import os
+def test_default_does_not_send_callsign(monkeypatch):
     monkeypatch.setenv("MESHTASTIC_OPERATOR_CALLSIGN", "N0CALL")
     monkeypatch.delenv("MESHTASTIC_SEND_CALLSIGN_HEADER", raising=False)
-
-    raw = str(os.environ.get("MESHTASTIC_SEND_CALLSIGN_HEADER", "true")).strip().lower()
-    send_callsign_header = raw not in {"0", "false", "no", "off", ""}
-    assert send_callsign_header is True
+    assert _send_callsign_header_from_env() is False
 
 
-def test_opt_out_suppresses_callsign(monkeypatch):
-    """Setting MESHTASTIC_SEND_CALLSIGN_HEADER=false suppresses the header."""
-    import os
+def test_opt_in_sends_callsign(monkeypatch):
     monkeypatch.setenv("MESHTASTIC_OPERATOR_CALLSIGN", "N0CALL")
-    monkeypatch.setenv("MESHTASTIC_SEND_CALLSIGN_HEADER", "false")
-
-    raw = str(os.environ.get("MESHTASTIC_SEND_CALLSIGN_HEADER", "true")).strip().lower()
-    send_callsign_header = raw not in {"0", "false", "no", "off", ""}
-    assert send_callsign_header is False
+    monkeypatch.setenv("MESHTASTIC_SEND_CALLSIGN_HEADER", "true")
+    assert _send_callsign_header_from_env() is True
 
 
-def test_various_falsy_values_all_opt_out(monkeypatch):
-    """Common falsy strings should all suppress the callsign header."""
-    import os
-    for falsy in ("0", "false", "FALSE", "no", "off"):
+def test_various_falsy_values_do_not_opt_in(monkeypatch):
+    for falsy in ("0", "false", "FALSE", "no", "off", ""):
         monkeypatch.setenv("MESHTASTIC_SEND_CALLSIGN_HEADER", falsy)
-        raw = str(os.environ.get("MESHTASTIC_SEND_CALLSIGN_HEADER", "true")).strip().lower()
-        send_callsign_header = raw not in {"0", "false", "no", "off", ""}
-        assert send_callsign_header is False, f"value {falsy!r} did not opt out"
+        assert _send_callsign_header_from_env() is False, f"value {falsy!r} should not opt in"

backend/tests/test_per_operator_outbound_attribution.py

@@ -133,23 +133,19 @@ class TestOperatorHandleGeneration:
 
 
 class TestOutboundUserAgentString:
-    def test_includes_operator_handle(self, isolated_handle):
+    def test_ua_is_operator_handle(self, isolated_handle):
         ua = isolated_handle.outbound_user_agent()
         handle = isolated_handle.get_operator_handle()
-        assert f"operator: {handle}" in ua
+        assert ua == handle
 
     def test_includes_purpose_when_provided(self, isolated_handle):
         ua = isolated_handle.outbound_user_agent("wikipedia")
-        assert "purpose: wikipedia" in ua
+        handle = isolated_handle.get_operator_handle()
+        assert ua == f"{handle} (purpose: wikipedia)"
 
-    def test_includes_contact_path(self, isolated_handle):
-        ua = isolated_handle.outbound_user_agent()
-        assert "github.com" in ua.lower()
-        assert "shadowbroker" in ua.lower()
-
-    def test_version_prefix(self, isolated_handle):
-        ua = isolated_handle.outbound_user_agent()
-        assert ua.startswith("Shadowbroker/")
+    def test_no_shadowbroker_product_token(self, isolated_handle):
+        ua = isolated_handle.outbound_user_agent("nominatim")
+        assert "shadowbroker" not in ua.lower()
 
 
 # ---------------------------------------------------------------------------
@@ -181,8 +177,8 @@ class TestWikimediaCallsAreNowPerOperator:
         assert "Api-User-Agent" in headers
         handle = isolated_handle.get_operator_handle()
         for header_value in (headers["User-Agent"], headers["Api-User-Agent"]):
-            assert f"operator: {handle}" in header_value, (
-                f"Wikimedia UA must include the per-operator handle; got {header_value!r}"
+            assert header_value.startswith(handle), (
+                f"Wikimedia UA must be the per-operator handle; got {header_value!r}"
             )
 
     def test_wikipedia_summary_uses_per_operator_ua(self, isolated_handle, monkeypatch):
@@ -211,7 +207,8 @@ class TestWikimediaCallsAreNowPerOperator:
         assert wikipedia_hits, "Wikipedia summary fetch was not called"
         for _url, headers in wikipedia_hits:
             handle = isolated_handle.get_operator_handle()
-            assert f"operator: {handle}" in headers.get("User-Agent", "")
+            ua = headers.get("User-Agent", "")
+            assert ua.startswith(handle), f"Wikipedia UA must be the operator handle; got {ua!r}"
 
 
 # ---------------------------------------------------------------------------
@@ -233,6 +230,7 @@ class TestNoMonsterUserAgentRemains:
     """
 
     BANNED_LITERALS = (
+        "Shadowbroker/",
         "ShadowBroker-OSINT/1.0",
         "ShadowBroker-OSINT/0.9",
         "ShadowBroker-FeedIngester/1.0",