diff --git a/work/dashboard-frontend/src/components/GenericSourcesEditor.tsx b/work/dashboard-frontend/src/components/GenericSourcesEditor.tsx index a59a0a4..027db98 100644 --- a/work/dashboard-frontend/src/components/GenericSourcesEditor.tsx +++ b/work/dashboard-frontend/src/components/GenericSourcesEditor.tsx @@ -31,6 +31,7 @@ function blankSource(n: number): GenericSource { field_mappings: [], summary_template: '', emoji: '', + headers: {}, } } @@ -176,12 +177,43 @@ export default function GenericSourcesEditor() { ) } + // Headers are stored as a Record on the source; edit them as + // an ordered list of key/value rows (mirrors the field_mappings editor), + // rebuilding the record on each change. + const headerEntries = (s: GenericSource): [string, string][] => + Object.entries(s.headers ?? {}) + + const setHeaderEntries = (idx: number, entries: [string, string][]) => { + const obj: Record = {} + for (const [k, v] of entries) obj[k] = v + update(idx, { headers: obj }) + } + + const addHeader = (idx: number) => { + if (!sources) return + setHeaderEntries(idx, [...headerEntries(sources[idx]), ['', '']]) + } + + const updateHeader = (idx: number, hIdx: number, key: string, value: string) => { + if (!sources) return + const entries = headerEntries(sources[idx]) + entries[hIdx] = [key, value] + setHeaderEntries(idx, entries) + } + + const removeHeader = (idx: number, hIdx: number) => { + if (!sources) return + const entries = headerEntries(sources[idx]) + entries.splice(hIdx, 1) + setHeaderEntries(idx, entries) + } + const runPreview = async (idx: number) => { if (!sources) return const src = sources[idx] setPreviewing((p) => ({ ...p, [idx]: true })) try { - const res = await previewGenericSource(src.url, src.items_path) + const res = await previewGenericSource(src.url, src.items_path, src.headers) setPreviews((p) => ({ ...p, [idx]: res })) } catch (e) { setPreviews((p) => ({ @@ -391,6 +423,55 @@ export default function GenericSourcesEditor() { /> + {/* Headers (optional) */} +
+
+
+ Headers (optional) +
+ +
+

+ Custom request headers, e.g. User-Agent or{' '} + Authorization. Leave empty for the default browser UA. +

+ {Object.entries(src.headers ?? {}).length > 0 && ( +
+ {Object.entries(src.headers ?? {}).map(([hKey, hVal], hIdx) => ( +
+ updateHeader(i, hIdx, e.target.value, hVal)} + placeholder="header name (e.g. Authorization)" + className="flex-1 bg-[#0d0d0d] border border-border px-2 py-1.5 text-xs font-mono text-[#e0e0e0] placeholder:text-[#555]" + /> + : + updateHeader(i, hIdx, hKey, e.target.value)} + placeholder="value (e.g. Bearer …)" + className="flex-1 bg-[#0d0d0d] border border-border px-2 py-1.5 text-xs font-mono text-[#e0e0e0] placeholder:text-[#555]" + /> + +
+ ))} +
+ )} +
+ {/* Location */} Location — GeoJSON geometry OR lat + lon paths
diff --git a/work/dashboard-frontend/src/lib/api.ts b/work/dashboard-frontend/src/lib/api.ts index f9fb38b..74e6aa6 100644 --- a/work/dashboard-frontend/src/lib/api.ts +++ b/work/dashboard-frontend/src/lib/api.ts @@ -326,6 +326,9 @@ export interface GenericSource { field_mappings: FieldMapping[] summary_template?: string emoji?: string + // Optional custom request headers (e.g. User-Agent override or Authorization). + // Empty/absent = the default browser UA. Sent on both poll and preview. + headers?: Record } export interface GenericSourcePreview { @@ -359,12 +362,13 @@ export async function saveGenericSources( export async function previewGenericSource( url: string, - items_path?: string + items_path?: string, + headers?: Record ): Promise { const response = await fetch('/api/generic-sources/preview', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ url, items_path }), + body: JSON.stringify({ url, items_path, headers }), }) // The endpoint never raises; it returns {ok:false,error} on failure too. return response.json() diff --git a/work/meshai/dashboard/api/generic_sources_routes.py b/work/meshai/dashboard/api/generic_sources_routes.py index af069f7..090230e 100644 --- a/work/meshai/dashboard/api/generic_sources_routes.py +++ b/work/meshai/dashboard/api/generic_sources_routes.py @@ -12,6 +12,7 @@ The single route never raises: on any error it returns import json import logging +import time from urllib.error import HTTPError, URLError from urllib.request import Request as UrlRequest, urlopen @@ -19,8 +20,9 @@ from fastapi import APIRouter, Request from starlette.concurrency import run_in_threadpool # Reuse the adapter's dotted-path walker so Preview resolves items_path exactly -# the way the running GenericHttpAdapter will. -from meshai.env.generic_http import _dig +# the way the running GenericHttpAdapter will. Share the browser UA too so +# Preview hits WAF'd feeds the same way the real poll does. +from meshai.env.generic_http import _dig, _BROWSER_UA logger = logging.getLogger(__name__) @@ -31,7 +33,9 @@ router = APIRouter(tags=["generic-sources"]) _SAMPLE_MAX = 8192 _ITEM_MAX = 2048 _FETCH_TIMEOUT = 30 -_USER_AGENT = "MeshAI/1.0" +# Browser-like default UA (WAFs 403 the short "MeshAI/1.0"). A source's custom +# headers layer on top and can override it. +_USER_AGENT = _BROWSER_UA def _truncate(text: str, limit: int) -> str: @@ -40,22 +44,40 @@ def _truncate(text: str, limit: int) -> str: return text[:limit] + f"\n… (truncated, {len(text)} chars total)" -def _fetch_preview(url: str, items_path: str) -> dict: - """Blocking fetch + parse. NEVER raises — always returns a result dict.""" +def _fetch_preview(url: str, items_path: str, headers: dict = None) -> dict: + """Blocking fetch + parse. NEVER raises — always returns a result dict. + + Uses the browser UA by default; any per-source ``headers`` (UA override / + auth) layer on top so Preview hits the feed exactly as the real poll will. + Retries ONCE on a 403/429 (WAF intermittent block), matching the adapter. + """ if not url or not isinstance(url, str): return {"ok": False, "error": "A url is required."} - try: - req = UrlRequest(url, headers={"User-Agent": _USER_AGENT}) - with urlopen(req, timeout=_FETCH_TIMEOUT) as resp: - status = getattr(resp, "status", None) or resp.getcode() - raw = resp.read() - except HTTPError as e: - return {"ok": False, "status": e.code, "error": f"HTTP {e.code}: {e.reason}"} - except URLError as e: - return {"ok": False, "error": f"Could not reach URL: {e.reason}"} - except Exception as e: # timeout, DNS, connection reset, … - return {"ok": False, "error": f"Fetch failed: {e}"} + req_headers = {"User-Agent": _USER_AGENT, + "Accept": "application/json, text/plain, */*"} + if isinstance(headers, dict): + req_headers.update(headers) + + raw = status = None + for attempt in (1, 2): + try: + req = UrlRequest(url, headers=req_headers) + with urlopen(req, timeout=_FETCH_TIMEOUT) as resp: + status = getattr(resp, "status", None) or resp.getcode() + raw = resp.read() + break + except HTTPError as e: + if attempt == 1 and e.code in (403, 429): + time.sleep(1) + continue + return {"ok": False, "status": e.code, "error": f"HTTP {e.code}: {e.reason}"} + except URLError as e: + return {"ok": False, "error": f"Could not reach URL: {e.reason}"} + except Exception as e: # timeout, DNS, connection reset, … + return {"ok": False, "error": f"Fetch failed: {e}"} + if raw is None: + return {"ok": False, "error": "Fetch failed: no response."} try: text = raw.decode("utf-8", errors="replace") @@ -103,7 +125,7 @@ def _fetch_preview(url: str, items_path: str) -> dict: async def preview_generic_source(request: Request): """Server-side fetch of a candidate feed URL for the no-code editor. - Body: ``{"url": str, "items_path"?: str}``. + Body: ``{"url": str, "items_path"?: str, "headers"?: dict}``. Returns ``{ok, status?, error?, sample?, item_count?, first_item?}`` — never raises; failures come back as ``{ok: false, error: ...}``. """ @@ -114,5 +136,8 @@ async def preview_generic_source(request: Request): url = (body or {}).get("url") items_path = (body or {}).get("items_path") or "" + headers = (body or {}).get("headers") + if not isinstance(headers, dict): + headers = None - return await run_in_threadpool(_fetch_preview, url, items_path) + return await run_in_threadpool(_fetch_preview, url, items_path, headers) diff --git a/work/meshai/env/generic_http.py b/work/meshai/env/generic_http.py index a9cc9a0..dd9a21f 100644 --- a/work/meshai/env/generic_http.py +++ b/work/meshai/env/generic_http.py @@ -37,6 +37,13 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +# Browser-like default User-Agent. The old short "MeshAI/1.0" UA intermittently +# trips WAFs (Idaho Power's Azure Front Door 403s it ~2/30 requests); a +# browser UA gets 200 every time. A source can still override this (or add +# auth) via its optional per-source `headers` dict. +_BROWSER_UA = ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36") + # Only substitute {word} tokens in summary_template so stray braces in data # values can never blow up the formatter. _TEMPLATE_TOKEN = re.compile(r"\{(\w+)\}") @@ -88,6 +95,7 @@ class GenericHttpAdapter: category, poll_seconds, severity, field_mappings: [{source_path, dest_key}, ...], summary_template, emoji, + headers: {header_name: value, ...}, # optional — UA/auth override } """ @@ -153,7 +161,7 @@ class GenericHttpAdapter: def _poll_source(self, source: dict, now: float) -> bool: """Fetch + map one source. Returns True if its id-set changed.""" name = source.get("name") or source.get("url") - raw = self._fetch(source["url"]) + raw = self._fetch(source["url"], extra_headers=source.get("headers")) if raw is None: return False @@ -187,32 +195,47 @@ class GenericHttpAdapter: name, len(mapped)) return changed - def _fetch(self, url: str): + def _fetch(self, url: str, extra_headers: dict = None): """GET ``url`` and return parsed JSON, or None on any error. - stdlib urllib (mirrors env/usgs_quake.py); 30s timeout; UA header. + stdlib urllib (mirrors env/usgs_quake.py); 30s timeout. Defaults to a + browser-like User-Agent (WAFs 403 the short "MeshAI/1.0" UA); a source's + optional ``headers`` dict is layered on top so it can override the UA or + add auth (e.g. ``Authorization``). Retries ONCE on a 403/429 (a WAF + that intermittently blocks succeeds on the immediate retry). """ - headers = {"User-Agent": "MeshAI/1.0", "Accept": "application/json"} - try: - req = Request(url, headers=headers) - with urlopen(req, timeout=30) as resp: - return json.loads(resp.read().decode("utf-8")) - except HTTPError as e: - logger.warning("generic_http HTTP error %s for %s", e.code, url) - self._last_error = f"HTTP {e.code}" - self._consecutive_errors += 1 - return None - except URLError as e: - logger.warning("generic_http connection error for %s: %s", - url, e.reason) - self._last_error = str(e.reason) - self._consecutive_errors += 1 - return None - except Exception as e: - logger.warning("generic_http fetch error for %s: %s", url, e) - self._last_error = str(e) - self._consecutive_errors += 1 - return None + headers = {"User-Agent": _BROWSER_UA, + "Accept": "application/json, text/plain, */*"} + headers.update(extra_headers or {}) + for attempt in (1, 2): + try: + req = Request(url, headers=headers) + with urlopen(req, timeout=30) as resp: + return json.loads(resp.read().decode("utf-8")) + except HTTPError as e: + # Retry once on a WAF-style block (403/429) — the live test + # showed an immediate retry succeeds. + if attempt == 1 and e.code in (403, 429): + logger.info("generic_http HTTP %s for %s — retrying once", + e.code, url) + time.sleep(1) + continue + logger.warning("generic_http HTTP error %s for %s", e.code, url) + self._last_error = f"HTTP {e.code}" + self._consecutive_errors += 1 + return None + except URLError as e: + logger.warning("generic_http connection error for %s: %s", + url, e.reason) + self._last_error = str(e.reason) + self._consecutive_errors += 1 + return None + except Exception as e: + logger.warning("generic_http fetch error for %s: %s", url, e) + self._last_error = str(e) + self._consecutive_errors += 1 + return None + return None # ------------------------------------------------------------------ # Item -> internal event dict diff --git a/work/tests/test_generic_http.py b/work/tests/test_generic_http.py index 44c2c46..2b40fab 100644 --- a/work/tests/test_generic_http.py +++ b/work/tests/test_generic_http.py @@ -11,7 +11,9 @@ from __future__ import annotations import json -from meshai.env.generic_http import GenericHttpAdapter, _dig +from urllib.error import HTTPError + +from meshai.env.generic_http import GenericHttpAdapter, _BROWSER_UA, _dig from meshai.env.store import EnvironmentalStore from meshai.config import EnvironmentalConfig from meshai.notifications.pipeline.bus import EventBus @@ -199,7 +201,7 @@ def _make_store_with_generic(): def test_cold_start_silent_first_poll_seeds_persists_no_emit(): store, adapter, captured = _make_store_with_generic() # Stub the network fetch with one active outage. - adapter._fetch = lambda url: _payload([IDAHO_POWER_ITEM]) + adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM]) store.refresh() # poll 1 == pre-existing backlog @@ -217,13 +219,13 @@ def test_cold_start_silent_first_poll_seeds_persists_no_emit(): def test_later_poll_broadcasts_newly_received_item(): store, adapter, captured = _make_store_with_generic() - adapter._fetch = lambda url: _payload([IDAHO_POWER_ITEM]) + adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM]) store.refresh() # poll 1 — seed silently assert captured == [] # A genuinely NEW outage appears on a later poll -> it must broadcast. new_item = dict(IDAHO_POWER_ITEM, omsOutageId="456", omsCustomerCount=99) - adapter._fetch = lambda url: _payload([IDAHO_POWER_ITEM, new_item]) + adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM, new_item]) adapter._last_poll.clear() # force cadence to elapse store.refresh() # poll 2 @@ -236,10 +238,135 @@ def test_later_poll_broadcasts_newly_received_item(): assert n == 2 +# =========================================================================== +# _fetch: browser UA default, per-source header override, 403 retry +# =========================================================================== + +def test_fetch_uses_browser_ua_by_default(monkeypatch): + """Default fetch sends the browser UA (not the WAF-tripping MeshAI/1.0).""" + captured = {} + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return b'{"ok": true}' + + def _fake_urlopen(req, timeout=None): + captured["headers"] = dict(req.headers) + return _Resp() + + monkeypatch.setattr("meshai.env.generic_http.urlopen", _fake_urlopen) + adapter = GenericHttpAdapter([IDAHO_POWER_SOURCE]) + result = adapter._fetch("https://example.com/feed") + + assert result == {"ok": True} + # urllib title-cases header names in Request.headers. + assert captured["headers"].get("User-agent") == _BROWSER_UA + + +def test_fetch_per_source_headers_override_ua_and_add_auth(monkeypatch): + """A source's headers dict overrides the UA and adds arbitrary auth.""" + captured = {} + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return b'{"ok": true}' + + def _fake_urlopen(req, timeout=None): + captured["headers"] = dict(req.headers) + return _Resp() + + monkeypatch.setattr("meshai.env.generic_http.urlopen", _fake_urlopen) + adapter = GenericHttpAdapter([IDAHO_POWER_SOURCE]) + result = adapter._fetch( + "https://example.com/feed", + extra_headers={"User-Agent": "X", "Authorization": "Bearer y"}, + ) + + assert result == {"ok": True} + assert captured["headers"].get("User-agent") == "X" # overridden + assert captured["headers"].get("Authorization") == "Bearer y" # added + + +def test_fetch_retries_once_on_403_then_succeeds(monkeypatch): + """First 403 (WAF block) is retried once and the retry's JSON is returned.""" + calls = {"n": 0} + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return b'{"ok": true}' + + def _fake_urlopen(req, timeout=None): + calls["n"] += 1 + if calls["n"] == 1: + raise HTTPError("https://example.com/feed", 403, "Forbidden", {}, None) + return _Resp() + + monkeypatch.setattr("meshai.env.generic_http.urlopen", _fake_urlopen) + monkeypatch.setattr("meshai.env.generic_http.time.sleep", lambda *_: None) + adapter = GenericHttpAdapter([IDAHO_POWER_SOURCE]) + result = adapter._fetch("https://example.com/feed") + + assert result == {"ok": True} + assert calls["n"] == 2, "must retry exactly once on 403" + + +def test_fetch_persistent_403_returns_none_after_one_retry(monkeypatch): + """A feed that 403s on both attempts gives up (None) after the single retry.""" + calls = {"n": 0} + + def _fake_urlopen(req, timeout=None): + calls["n"] += 1 + raise HTTPError("https://example.com/feed", 403, "Forbidden", {}, None) + + monkeypatch.setattr("meshai.env.generic_http.urlopen", _fake_urlopen) + monkeypatch.setattr("meshai.env.generic_http.time.sleep", lambda *_: None) + adapter = GenericHttpAdapter([IDAHO_POWER_SOURCE]) + result = adapter._fetch("https://example.com/feed") + + assert result is None + assert calls["n"] == 2, "one initial attempt + one retry, then give up" + assert adapter._last_error == "HTTP 403" + + +def test_poll_source_threads_source_headers_into_fetch(monkeypatch): + """_poll_source passes the source's headers through to _fetch.""" + seen = {} + + def _fake_fetch(url, extra_headers=None): + seen["url"] = url + seen["extra_headers"] = extra_headers + return _payload([IDAHO_POWER_ITEM]) + + source = dict(IDAHO_POWER_SOURCE, headers={"Authorization": "Bearer z"}) + adapter = GenericHttpAdapter([source]) + adapter._fetch = _fake_fetch + adapter._poll_source(source, now=1000.0) + + assert seen["extra_headers"] == {"Authorization": "Bearer z"} + + def test_build_generic_detail_reader(): from meshai.notifications.env_reporter import EnvReporter store, adapter, captured = _make_store_with_generic() - adapter._fetch = lambda url: _payload([IDAHO_POWER_ITEM]) + adapter._fetch = lambda url, extra_headers=None: _payload([IDAHO_POWER_ITEM]) store.refresh() text = EnvReporter().build_generic_detail()