fix(generic): browser UA default + per-source custom headers + 403 retry (#85)

The MeshAI/1.0 UA intermittently trips WAFs (Idaho Power's Azure Front Door
403s it ~2/30; a browser UA gets 200 every time). Default the adapter +
preview to a browser User-Agent, retry once on 403/429, and add optional
per-source custom headers (UA/auth) editable in the GUI. Makes WAF'd and
keyed feeds pollable.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-07 02:24:27 -06:00 committed by GitHub
commit 1a1aef2e6e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 310 additions and 50 deletions

View file

@ -31,6 +31,7 @@ function blankSource(n: number): GenericSource {
field_mappings: [], field_mappings: [],
summary_template: '', summary_template: '',
emoji: '', emoji: '',
headers: {},
} }
} }
@ -176,12 +177,43 @@ export default function GenericSourcesEditor() {
) )
} }
// Headers are stored as a Record<string,string> 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<string, string> = {}
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) => { const runPreview = async (idx: number) => {
if (!sources) return if (!sources) return
const src = sources[idx] const src = sources[idx]
setPreviewing((p) => ({ ...p, [idx]: true })) setPreviewing((p) => ({ ...p, [idx]: true }))
try { 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 })) setPreviews((p) => ({ ...p, [idx]: res }))
} catch (e) { } catch (e) {
setPreviews((p) => ({ setPreviews((p) => ({
@ -391,6 +423,55 @@ export default function GenericSourcesEditor() {
/> />
</div> </div>
{/* Headers (optional) */}
<div className="border border-border p-3 space-y-2">
<div className="flex items-center justify-between">
<div className="text-[10px] font-sans font-medium uppercase tracking-widest text-[#666]">
Headers (optional)
</div>
<button
onClick={() => addHeader(i)}
className="flex items-center gap-1 px-2 py-1 text-xs bg-accent/20 hover:bg-accent/30 text-accent border border-accent/30"
>
<Plus size={12} /> Add header
</button>
</div>
<p className="text-xs text-[#555]">
Custom request headers, e.g. <code>User-Agent</code> or{' '}
<code>Authorization</code>. Leave empty for the default browser UA.
</p>
{Object.entries(src.headers ?? {}).length > 0 && (
<div className="space-y-2">
{Object.entries(src.headers ?? {}).map(([hKey, hVal], hIdx) => (
<div key={hIdx} className="flex items-center gap-2">
<input
type="text"
value={hKey}
onChange={(e) => 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]"
/>
<span className="text-[#555] text-xs">:</span>
<input
type="text"
value={hVal}
onChange={(e) => 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]"
/>
<button
onClick={() => removeHeader(i, hIdx)}
title="Remove header"
className="flex items-center px-2 py-1.5 text-xs text-[#777] hover:text-red-400 border border-border"
>
<Trash2 size={12} />
</button>
</div>
))}
</div>
)}
</div>
{/* Location */} {/* Location */}
<SectionLabel>Location GeoJSON geometry OR lat + lon paths</SectionLabel> <SectionLabel>Location GeoJSON geometry OR lat + lon paths</SectionLabel>
<div className="grid grid-cols-3 gap-2"> <div className="grid grid-cols-3 gap-2">

View file

@ -326,6 +326,9 @@ export interface GenericSource {
field_mappings: FieldMapping[] field_mappings: FieldMapping[]
summary_template?: string summary_template?: string
emoji?: 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<string, string>
} }
export interface GenericSourcePreview { export interface GenericSourcePreview {
@ -359,12 +362,13 @@ export async function saveGenericSources(
export async function previewGenericSource( export async function previewGenericSource(
url: string, url: string,
items_path?: string items_path?: string,
headers?: Record<string, string>
): Promise<GenericSourcePreview> { ): Promise<GenericSourcePreview> {
const response = await fetch('/api/generic-sources/preview', { const response = await fetch('/api/generic-sources/preview', {
method: 'POST', method: 'POST',
headers: { 'Content-Type': 'application/json' }, 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. // The endpoint never raises; it returns {ok:false,error} on failure too.
return response.json() return response.json()

View file

@ -12,6 +12,7 @@ The single route never raises: on any error it returns
import json import json
import logging import logging
import time
from urllib.error import HTTPError, URLError from urllib.error import HTTPError, URLError
from urllib.request import Request as UrlRequest, urlopen from urllib.request import Request as UrlRequest, urlopen
@ -19,8 +20,9 @@ from fastapi import APIRouter, Request
from starlette.concurrency import run_in_threadpool from starlette.concurrency import run_in_threadpool
# Reuse the adapter's dotted-path walker so Preview resolves items_path exactly # Reuse the adapter's dotted-path walker so Preview resolves items_path exactly
# the way the running GenericHttpAdapter will. # the way the running GenericHttpAdapter will. Share the browser UA too so
from meshai.env.generic_http import _dig # 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__) logger = logging.getLogger(__name__)
@ -31,7 +33,9 @@ router = APIRouter(tags=["generic-sources"])
_SAMPLE_MAX = 8192 _SAMPLE_MAX = 8192
_ITEM_MAX = 2048 _ITEM_MAX = 2048
_FETCH_TIMEOUT = 30 _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: 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)" return text[:limit] + f"\n… (truncated, {len(text)} chars total)"
def _fetch_preview(url: str, items_path: str) -> dict: def _fetch_preview(url: str, items_path: str, headers: dict = None) -> dict:
"""Blocking fetch + parse. NEVER raises — always returns a result 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): if not url or not isinstance(url, str):
return {"ok": False, "error": "A url is required."} return {"ok": False, "error": "A url is required."}
try: req_headers = {"User-Agent": _USER_AGENT,
req = UrlRequest(url, headers={"User-Agent": _USER_AGENT}) "Accept": "application/json, text/plain, */*"}
with urlopen(req, timeout=_FETCH_TIMEOUT) as resp: if isinstance(headers, dict):
status = getattr(resp, "status", None) or resp.getcode() req_headers.update(headers)
raw = resp.read()
except HTTPError as e: raw = status = None
return {"ok": False, "status": e.code, "error": f"HTTP {e.code}: {e.reason}"} for attempt in (1, 2):
except URLError as e: try:
return {"ok": False, "error": f"Could not reach URL: {e.reason}"} req = UrlRequest(url, headers=req_headers)
except Exception as e: # timeout, DNS, connection reset, … with urlopen(req, timeout=_FETCH_TIMEOUT) as resp:
return {"ok": False, "error": f"Fetch failed: {e}"} 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: try:
text = raw.decode("utf-8", errors="replace") 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): async def preview_generic_source(request: Request):
"""Server-side fetch of a candidate feed URL for the no-code editor. """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 Returns ``{ok, status?, error?, sample?, item_count?, first_item?}`` never
raises; failures come back as ``{ok: false, error: ...}``. raises; failures come back as ``{ok: false, error: ...}``.
""" """
@ -114,5 +136,8 @@ async def preview_generic_source(request: Request):
url = (body or {}).get("url") url = (body or {}).get("url")
items_path = (body or {}).get("items_path") or "" 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)

View file

@ -37,6 +37,13 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__) 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 # Only substitute {word} tokens in summary_template so stray braces in data
# values can never blow up the formatter. # values can never blow up the formatter.
_TEMPLATE_TOKEN = re.compile(r"\{(\w+)\}") _TEMPLATE_TOKEN = re.compile(r"\{(\w+)\}")
@ -88,6 +95,7 @@ class GenericHttpAdapter:
category, poll_seconds, severity, category, poll_seconds, severity,
field_mappings: [{source_path, dest_key}, ...], field_mappings: [{source_path, dest_key}, ...],
summary_template, emoji, 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: def _poll_source(self, source: dict, now: float) -> bool:
"""Fetch + map one source. Returns True if its id-set changed.""" """Fetch + map one source. Returns True if its id-set changed."""
name = source.get("name") or source.get("url") 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: if raw is None:
return False return False
@ -187,32 +195,47 @@ class GenericHttpAdapter:
name, len(mapped)) name, len(mapped))
return changed 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. """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"} headers = {"User-Agent": _BROWSER_UA,
try: "Accept": "application/json, text/plain, */*"}
req = Request(url, headers=headers) headers.update(extra_headers or {})
with urlopen(req, timeout=30) as resp: for attempt in (1, 2):
return json.loads(resp.read().decode("utf-8")) try:
except HTTPError as e: req = Request(url, headers=headers)
logger.warning("generic_http HTTP error %s for %s", e.code, url) with urlopen(req, timeout=30) as resp:
self._last_error = f"HTTP {e.code}" return json.loads(resp.read().decode("utf-8"))
self._consecutive_errors += 1 except HTTPError as e:
return None # Retry once on a WAF-style block (403/429) — the live test
except URLError as e: # showed an immediate retry succeeds.
logger.warning("generic_http connection error for %s: %s", if attempt == 1 and e.code in (403, 429):
url, e.reason) logger.info("generic_http HTTP %s for %s — retrying once",
self._last_error = str(e.reason) e.code, url)
self._consecutive_errors += 1 time.sleep(1)
return None continue
except Exception as e: logger.warning("generic_http HTTP error %s for %s", e.code, url)
logger.warning("generic_http fetch error for %s: %s", url, e) self._last_error = f"HTTP {e.code}"
self._last_error = str(e) self._consecutive_errors += 1
self._consecutive_errors += 1 return None
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 # Item -> internal event dict

View file

@ -11,7 +11,9 @@ from __future__ import annotations
import json 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.env.store import EnvironmentalStore
from meshai.config import EnvironmentalConfig from meshai.config import EnvironmentalConfig
from meshai.notifications.pipeline.bus import EventBus 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(): def test_cold_start_silent_first_poll_seeds_persists_no_emit():
store, adapter, captured = _make_store_with_generic() store, adapter, captured = _make_store_with_generic()
# Stub the network fetch with one active outage. # 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 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(): def test_later_poll_broadcasts_newly_received_item():
store, adapter, captured = _make_store_with_generic() 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 store.refresh() # poll 1 — seed silently
assert captured == [] assert captured == []
# A genuinely NEW outage appears on a later poll -> it must broadcast. # A genuinely NEW outage appears on a later poll -> it must broadcast.
new_item = dict(IDAHO_POWER_ITEM, omsOutageId="456", omsCustomerCount=99) 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 adapter._last_poll.clear() # force cadence to elapse
store.refresh() # poll 2 store.refresh() # poll 2
@ -236,10 +238,135 @@ def test_later_poll_broadcasts_newly_received_item():
assert n == 2 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(): def test_build_generic_detail_reader():
from meshai.notifications.env_reporter import EnvReporter from meshai.notifications.env_reporter import EnvReporter
store, adapter, captured = _make_store_with_generic() 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() store.refresh()
text = EnvReporter().build_generic_detail() text = EnvReporter().build_generic_detail()