From d3a2b55fedfdfded2aa65919144cecb9c5ef47fe Mon Sep 17 00:00:00 2001 From: Shadowbroker <43977454+BigBodyCobain@users.noreply.github.com> Date: Tue, 18 Aug 2026 16:05:26 -0600 Subject: [PATCH] fix(liveuamap): harden provider secrets and challenge handling --- backend/services/liveuamap_scraper.py | 81 +++++++++++++++++------- backend/services/liveuamap_settings.py | 7 +- backend/tests/test_liveuamap_provider.py | 57 +++++++++++++++-- 3 files changed, 116 insertions(+), 29 deletions(-) diff --git a/backend/services/liveuamap_scraper.py b/backend/services/liveuamap_scraper.py index 9426655..8b57823 100644 --- a/backend/services/liveuamap_scraper.py +++ b/backend/services/liveuamap_scraper.py @@ -3,7 +3,7 @@ Global Incidents itself is backed independently by GDELT. This module adds LiveUAMap pins when either an operator-configured supported API is available or the existing Playwright provider is allowed. Provider failures are isolated and -return an empty enrichment set instead of breaking the scheduler. +return an empty enrichment set instead of breaking the fetch scheduler. The browser provider intentionally does not add any new anti-bot behavior. It retains the repository's pre-existing Playwright/stealth profile for backward @@ -82,11 +82,22 @@ def _api_url() -> str: parsed = urlparse(raw) except ValueError: return "" - if parsed.scheme.lower() != "https" or not parsed.netloc: + if ( + parsed.scheme.lower() != "https" + or not parsed.hostname + or parsed.username is not None + or parsed.password is not None + ): return "" return raw +def _public_api_base_url(raw: str) -> str: + """Return a URL safe to use for relative links without exposing query auth.""" + parsed = urlparse(raw) + return parsed._replace(query="", fragment="").geturl() + + def _api_headers() -> dict[str, str]: from services.network_utils import outbound_user_agent @@ -116,22 +127,35 @@ def _fetch_liveuamap_api() -> list[dict[str, Any]]: timeout_s = _bounded_int_env("LIVEUAMAP_API_TIMEOUT_S", 30, minimum=5, maximum=120) hostname = urlparse(url).hostname or "configured endpoint" logger.info("Fetching LiveUAMap supported API from %s", hostname) - response = requests.get(url, headers=_api_headers(), timeout=(5, timeout_s)) + # Refuse redirects while sending operator credentials. A custom auth header + # is not guaranteed to be stripped by requests on a redirect, so the safest + # contract is that LIVEUAMAP_API_URL names the final HTTPS endpoint. + response = requests.get( + url, + headers=_api_headers(), + timeout=(5, timeout_s), + allow_redirects=False, + ) + if 300 <= response.status_code < 400: + raise requests.HTTPError("LiveUAMap API redirected; configure the final HTTPS endpoint") response.raise_for_status() try: payload = response.json() - except (requests.JSONDecodeError, ValueError) as exc: + except ValueError as exc: raise ValueError("LiveUAMap API did not return JSON/GeoJSON") from exc candidates = normalize_liveuamap_payload(payload) markers = _format_markers( candidates, region="LiveUAMap", - base_url=url, + base_url=_public_api_base_url(url), + fallback_link="https://liveuamap.com", provider="api", ) if not markers: - raise ValueError(f"LiveUAMap API returned no recognizable point markers ({payload_shape(payload)})") + raise ValueError( + f"LiveUAMap API returned no recognizable point markers ({payload_shape(payload)})" + ) logger.info("LiveUAMap API returned %s normalized markers", len(markers)) return markers @@ -249,21 +273,23 @@ def _fetch_liveuamap_browser() -> list[dict[str, Any]]: ) page.wait_for_timeout(5_000) html = page.content() - if _looks_like_challenge(html): - logger.warning( - "LiveUAMap %s appears to be serving an access challenge; " - "leaving this region empty", - region["name"], - ) - failed_regions += 1 - continue + # Try the useful payload before classifying the page as + # a challenge. Normal pages may legitimately load a + # Turnstile asset; valid marker data should win. payload = _read_page_payload(page, html) if payload is None: - logger.warning( - "LiveUAMap %s did not expose an ovens payload", - region["name"], - ) + if _looks_like_challenge(html): + logger.warning( + "LiveUAMap %s appears to be serving an access challenge; " + "leaving this region empty", + region["name"], + ) + else: + logger.warning( + "LiveUAMap %s did not expose an ovens payload", + region["name"], + ) failed_regions += 1 continue @@ -276,11 +302,17 @@ def _fetch_liveuamap_browser() -> list[dict[str, Any]]: seen_ids=seen_ids, ) if not region_markers: - logger.warning( - "LiveUAMap %s payload contained no recognizable point markers (%s)", - region["name"], - payload_shape(payload), - ) + if _looks_like_challenge(html): + logger.warning( + "LiveUAMap %s returned no markers and appears challenge-gated", + region["name"], + ) + else: + logger.warning( + "LiveUAMap %s payload contained no recognizable point markers (%s)", + region["name"], + payload_shape(payload), + ) failed_regions += 1 continue @@ -316,6 +348,7 @@ def _format_markers( base_url: str, provider: str, seen_ids: set[str] | None = None, + fallback_link: str | None = None, ) -> list[dict[str, Any]]: output: list[dict[str, Any]] = [] dedupe = seen_ids if seen_ids is not None else set() @@ -364,7 +397,7 @@ def _format_markers( "lng": lng, "timestamp": event_time if event_time is not None else "", "date": date_str, - "link": link or base_url, + "link": link or fallback_link or base_url, "region": _as_text(marker.get("region") or region).strip() or region, "category": category, "image": image, diff --git a/backend/services/liveuamap_settings.py b/backend/services/liveuamap_settings.py index 5e3d36a..75dd4d8 100644 --- a/backend/services/liveuamap_settings.py +++ b/backend/services/liveuamap_settings.py @@ -31,7 +31,12 @@ def _valid_https_url(raw: str) -> bool: parsed = urlparse(raw) except ValueError: return False - return parsed.scheme.lower() == "https" and bool(parsed.netloc) + return ( + parsed.scheme.lower() == "https" + and bool(parsed.hostname) + and parsed.username is None + and parsed.password is None + ) def liveuamap_requires_ui_opt_in() -> bool: diff --git a/backend/tests/test_liveuamap_provider.py b/backend/tests/test_liveuamap_provider.py index 8455e62..0bd031c 100644 --- a/backend/tests/test_liveuamap_provider.py +++ b/backend/tests/test_liveuamap_provider.py @@ -29,8 +29,8 @@ def test_api_geojson_is_normalized_and_auth_header_is_sent(monkeypatch): ) seen = {} - def fake_get(url, *, headers, timeout): - seen.update(url=url, headers=headers, timeout=timeout) + def fake_get(url, *, headers, timeout, allow_redirects): + seen.update(url=url, headers=headers, timeout=timeout, allow_redirects=allow_redirects) return _Response( { "type": "FeatureCollection", @@ -53,12 +53,49 @@ def test_api_geojson_is_normalized_and_auth_header_is_sent(monkeypatch): assert markers[0]["provider"] == "api" assert seen["headers"]["Authorization"] == "Bearer secret-key" assert seen["timeout"] == (5, 12) + assert seen["allow_redirects"] is False + + +def test_api_query_token_is_not_exposed_as_marker_fallback(monkeypatch): + monkeypatch.setenv( + "LIVEUAMAP_API_URL", + "https://api.example.test/events?token=super-secret", + ) + monkeypatch.setattr( + scraper.requests, + "get", + lambda *args, **kwargs: _Response( + [{"id": "evt", "lat": 10, "lng": 20, "title": "No link"}] + ), + ) + markers = scraper._fetch_liveuamap_api() + assert markers[0]["link"] == "https://liveuamap.com" + assert "super-secret" not in repr(markers) + + +def test_api_redirect_is_refused_before_following_credentials(monkeypatch): + monkeypatch.setenv("LIVEUAMAP_API_URL", "https://api.example.test/events") + monkeypatch.setattr( + scraper.requests, + "get", + lambda *args, **kwargs: _Response({}, status_code=302), + ) + try: + scraper._fetch_liveuamap_api() + except requests.HTTPError as exc: + assert "redirected" in str(exc) + else: + raise AssertionError("redirect should have been rejected") def test_api_failure_falls_back_to_browser_when_browser_is_allowed(monkeypatch): monkeypatch.setattr(settings, "liveuamap_api_configured", lambda: True) monkeypatch.setattr(settings, "liveuamap_browser_scraper_enabled", lambda: True) - monkeypatch.setattr(scraper, "_fetch_liveuamap_api", lambda: (_ for _ in ()).throw(requests.Timeout("boom"))) + monkeypatch.setattr( + scraper, + "_fetch_liveuamap_api", + lambda: (_ for _ in ()).throw(requests.Timeout("boom")), + ) monkeypatch.setattr(scraper, "_fetch_liveuamap_browser", lambda: [{"id": "browser"}]) assert scraper.fetch_liveuamap() == [{"id": "browser"}] @@ -66,7 +103,11 @@ def test_api_failure_falls_back_to_browser_when_browser_is_allowed(monkeypatch): def test_api_failure_does_not_force_browser_when_browser_is_disabled(monkeypatch): monkeypatch.setattr(settings, "liveuamap_api_configured", lambda: True) monkeypatch.setattr(settings, "liveuamap_browser_scraper_enabled", lambda: False) - monkeypatch.setattr(scraper, "_fetch_liveuamap_api", lambda: (_ for _ in ()).throw(requests.Timeout("boom"))) + monkeypatch.setattr( + scraper, + "_fetch_liveuamap_api", + lambda: (_ for _ in ()).throw(requests.Timeout("boom")), + ) called = False def browser(): @@ -88,6 +129,14 @@ def test_browser_disable_does_not_disable_configured_api_scheduler_gate(monkeypa assert settings.liveuamap_scraper_enabled() is True +def test_api_url_with_embedded_credentials_is_rejected(monkeypatch, tmp_path): + monkeypatch.setattr(settings, "_OPT_IN_FILE", tmp_path / "choice.json") + monkeypatch.setattr(settings.os, "name", "nt") + monkeypatch.setenv("LIVEUAMAP_API_URL", "https://user:password@api.example.test/events") + assert settings.liveuamap_api_configured() is False + assert scraper._api_url() == "" + + def test_http_api_endpoint_is_not_used(monkeypatch): monkeypatch.setenv("LIVEUAMAP_API_URL", "http://api.example.test/events") monkeypatch.setattr(scraper, "_fetch_liveuamap_browser", lambda: [])