TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { gunzipSync } from "node:zlib";2import { Agent, fetch as undiciFetch, type Dispatcher } from "undici";3import { assertUrlAllowed, safeLookup, UrlPolicyError, type FetchMeta, type Observation } from "@websensor/core";45/**6 * Generic HTTP fetcher: conditional GET (ETag / Last-Modified), manual redirect handling7 * with SSRF validation on every hop, size/time limits, and a dispatcher whose DNS lookup8 * only returns policy-approved addresses (DNS rebinding protection).9 */1011export interface FetchOptions {12 method?: "GET" | "HEAD";13 etag?: string | null;14 lastModified?: string | null;15 headers?: Record<string, string>;16 timeoutMs?: number;17 maxBytes?: number;18 maxRedirects?: number;19 accept?: string;20 userAgent?: string;21 /** keep every response header (minus cookies' values) — used by the `headers` connector */22 keepAllHeaders?: boolean;23}2425export class FetchError extends Error {26 constructor(27 public readonly code: string,28 message: string,29 ) {30 super(message);31 this.name = "FetchError";32 }33}3435const DEFAULT_UA = process.env.WS_USER_AGENT ?? "WebSensorBot/0.1 (+https://www.websensor.io/bot; contact@websensor.io)";36const DEFAULT_TIMEOUT = 25_000;37const DEFAULT_MAX_BYTES = 12 * 1024 * 1024;3839let agent: Dispatcher | null = null;40let agentH1: Dispatcher | null = null;41export function getDispatcher(h1only = false): Dispatcher {42 if (h1only) {43 if (!agentH1) agentH1 = new Agent({ connect: { lookup: safeLookup as never, timeout: 10_000 }, connections: 32, pipelining: 1, keepAliveTimeout: 15_000, headersTimeout: 25_000, bodyTimeout: 30_000, allowH2: false });44 return agentH1;45 }46 if (!agent) {47 agent = new Agent({48 connect: { lookup: safeLookup as never, timeout: 10_000 },49 connections: 64,50 pipelining: 1,51 keepAliveTimeout: 15_000,52 headersTimeout: 20_000,53 bodyTimeout: 30_000,54 allowH2: true,55 });56 }57 return agent;58}5960/** Hosts where HTTP/2 misbehaved (NGHTTP2 stream errors, header timeouts) — pinned to HTTP/1.1. */61const h1Hosts = new Set<string>();6263export async function closeDispatcher(): Promise<void> {64 if (agent) await agent.close();65 if (agentH1) await agentH1.close();66 agent = null;67 agentH1 = null;68}6970/** A conventional browser identity, used only as a second attempt when the bot UA is refused (403). */71export const BROWSER_UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.5 Safari/605.1.15";7273function headerMap(h: Headers, all = false): Record<string, string> {74 const out: Record<string, string> = {};75 h.forEach((v, k) => {76 if (all || /^(content-type|content-length|etag|last-modified|cache-control|server|x-ratelimit-[a-z-]+|retry-after|date|age|via|cf-ray|x-cache|link)$/i.test(k)) out[k.toLowerCase()] = v;77 });78 if (all) {79 // undici joins multiple Set-Cookie into one; getSetCookie() keeps them apart.80 const sc = (h as unknown as { getSetCookie?: () => string[] }).getSetCookie?.();81 if (sc?.length) out["set-cookie"] = sc.join(", ");82 }83 return out;84}8586export async function httpFetch(sensorId: string, urlStr: string, opts: FetchOptions = {}): Promise<Observation> {87 const started = Date.now();88 const method = opts.method ?? "GET";89 const maxRedirects = opts.maxRedirects ?? 5;90 let current = urlStr;91 let redirects = 0;92 const headers: Record<string, string> = {93 "user-agent": opts.userAgent ?? DEFAULT_UA,94 accept: opts.accept ?? "application/rss+xml, application/atom+xml, application/json, application/xml, text/html;q=0.9, text/plain;q=0.8, */*;q=0.5",95 "accept-language": "en-US,en;q=0.8,fr;q=0.5",96 "accept-encoding": "gzip, deflate, br",97 ...opts.headers,98 };99 if (opts.etag) headers["if-none-match"] = opts.etag;100 if (opts.lastModified) headers["if-modified-since"] = opts.lastModified;101102 const fail = (code: string, message: string): Observation => ({103 sensorId,104 url: urlStr,105 fetchedAt: new Date(),106 notModified: false,107 error: { code, message },108 meta: { status: 0, url: urlStr, finalUrl: current, contentType: null, contentLength: 0, etag: null, lastModified: null, durationMs: Date.now() - started, redirects, method, headers: {} },109 });110111 for (;;) {112 try {113 await assertUrlAllowed(current);114 } catch (e) {115 return fail("ssrf_blocked", e instanceof UrlPolicyError ? e.message : String(e));116 }117 const ac = new AbortController();118 const timer = setTimeout(() => ac.abort(), opts.timeoutMs ?? DEFAULT_TIMEOUT);119 let res: Response;120 try {121 const host = new URL(current).hostname;122 res = (await undiciFetch(current, { method, headers, redirect: "manual", signal: ac.signal, dispatcher: getDispatcher(h1Hosts.has(host)) } as never)) as unknown as Response;123 } catch (e) {124 clearTimeout(timer);125 const msg = e instanceof Error ? `${e.name}: ${e.message}${(e as { cause?: Error }).cause ? " — " + String((e as { cause?: Error }).cause?.message ?? (e as { cause?: unknown }).cause) : ""}` : String(e);126 if (/NGHTTP2|HTTP\/2/i.test(msg)) {127 const host = new URL(current).hostname;128 if (!h1Hosts.has(host)) {129 h1Hosts.add(host);130 continue; // retry this hop over HTTP/1.1131 }132 }133 const code = ac.signal.aborted ? "timeout" : /ENOTFOUND|EAI_AGAIN|getaddrinfo/i.test(msg) ? "dns" : /EBLOCKED|blocked/i.test(msg) ? "ssrf_blocked" : /CERT|TLS|SSL|certificate/i.test(msg) ? "tls" : /ECONNREFUSED|ECONNRESET|EHOSTUNREACH|ETIMEDOUT|socket/i.test(msg) ? "connection" : "fetch_failed";134 return fail(code, msg);135 }136137 if ([301, 302, 303, 307, 308].includes(res.status)) {138 clearTimeout(timer);139 const loc = res.headers.get("location");140 if (!loc) return fail("redirect_without_location", `HTTP ${res.status} without Location`);141 if (++redirects > maxRedirects) return fail("too_many_redirects", `More than ${maxRedirects} redirects`);142 try {143 current = new URL(loc, current).toString();144 } catch {145 return fail("bad_redirect", `Invalid Location header: ${loc}`);146 }147 // conditional headers only apply to the original resource148 delete headers["if-none-match"];149 delete headers["if-modified-since"];150 continue;151 }152153 const meta: FetchMeta = {154 status: res.status,155 url: urlStr,156 finalUrl: current,157 contentType: res.headers.get("content-type"),158 contentLength: 0,159 etag: res.headers.get("etag"),160 lastModified: res.headers.get("last-modified"),161 durationMs: 0,162 redirects,163 method,164 headers: headerMap(res.headers, opts.keepAllHeaders),165 };166167 if (res.status === 304) {168 clearTimeout(timer);169 meta.durationMs = Date.now() - started;170 return { sensorId, url: urlStr, fetchedAt: new Date(), meta, notModified: true };171 }172173 if (method === "HEAD") {174 clearTimeout(timer);175 meta.durationMs = Date.now() - started;176 meta.contentLength = Number(res.headers.get("content-length") ?? 0);177 return { sensorId, url: urlStr, fetchedAt: new Date(), meta, notModified: false };178 }179180 const maxBytes = opts.maxBytes ?? DEFAULT_MAX_BYTES;181 const declared = Number(res.headers.get("content-length") ?? 0);182 if (declared > maxBytes) {183 clearTimeout(timer);184 return fail("too_large", `Content-Length ${declared} exceeds ${maxBytes}`);185 }186 const chunks: Uint8Array[] = [];187 let total = 0;188 try {189 const reader = res.body?.getReader();190 if (reader) {191 for (;;) {192 const { done, value } = await reader.read();193 if (done) break;194 total += value.byteLength;195 if (total > maxBytes) {196 await reader.cancel();197 clearTimeout(timer);198 return fail("too_large", `Body exceeded ${maxBytes} bytes`);199 }200 chunks.push(value);201 }202 }203 } catch (e) {204 clearTimeout(timer);205 return fail(ac.signal.aborted ? "timeout" : "body_read_failed", e instanceof Error ? e.message : String(e));206 }207 clearTimeout(timer);208 let body = Buffer.concat(chunks);209 // Some servers send gzip'd sitemaps as application/octet-stream without content-encoding.210 if (body.length > 2 && body[0] === 0x1f && body[1] === 0x8b) {211 try {212 body = gunzipSync(body);213 } catch {214 // keep as-is215 }216 }217 // UTF-16 with BOM (e.g. AWS Health `public/currentevents`) → transcode to UTF-8 so every parser downstream works.218 if (body.length > 2 && ((body[0] === 0xff && body[1] === 0xfe) || (body[0] === 0xfe && body[1] === 0xff))) {219 const le = body[0] === 0xff;220 const payload = body.subarray(2);221 const utf16 = le ? payload : Buffer.from(payload).swap16();222 body = Buffer.from(utf16.toString("utf16le"), "utf8");223 } else if (body.length > 3 && body[0] === 0xef && body[1] === 0xbb && body[2] === 0xbf) {224 body = body.subarray(3); // UTF-8 BOM225 }226 meta.contentLength = body.length;227 meta.durationMs = Date.now() - started;228 return { sensorId, url: urlStr, fetchedAt: new Date(), meta, body, notModified: false };229 }230}231232/**233 * Outbound JSON POST (webhook alerts). Same SSRF policy and dispatcher as every fetch; no redirects234 * are followed (a webhook that redirects is misconfigured), small response cap, short timeout.235 */236export async function postJson(urlStr: string, body: string, headers: Record<string, string> = {}, timeoutMs = 10_000): Promise<{ status: number; error?: string }> {237 try {238 await assertUrlAllowed(urlStr);239 } catch (e) {240 return { status: 0, error: e instanceof UrlPolicyError ? e.message : String(e) };241 }242 const ac = new AbortController();243 const timer = setTimeout(() => ac.abort(), timeoutMs);244 try {245 const res = (await undiciFetch(urlStr, { method: "POST", body, headers: { "content-type": "application/json", "user-agent": DEFAULT_UA, ...headers }, redirect: "manual", signal: ac.signal, dispatcher: getDispatcher(true) } as never)) as unknown as Response;246 // drain (bounded) so the socket can be reused247 const reader = res.body?.getReader();248 let total = 0;249 if (reader) {250 for (;;) {251 const { done, value } = await reader.read();252 if (done) break;253 total += value.byteLength;254 if (total > 64 * 1024) {255 await reader.cancel();256 break;257 }258 }259 }260 return { status: res.status };261 } catch (e) {262 return { status: 0, error: ac.signal.aborted ? "timeout" : e instanceof Error ? e.message : String(e) };263 } finally {264 clearTimeout(timer);265 }266}267268/** Retry-aware wrapper for transient failures (5xx, connection, timeout). */269export async function httpFetchWithRetry(sensorId: string, url: string, opts: FetchOptions = {}, retries = 1): Promise<Observation> {270 let last: Observation | null = null;271 for (let attempt = 0; attempt <= retries; attempt++) {272 const obs = await httpFetch(sensorId, url, opts);273 last = obs;274 if (!obs.error && obs.meta.status === 403 && !opts.userAgent) {275 // Some WAFs refuse unknown bot identities on public feeds; try once as a regular browser.276 const alt = await httpFetch(sensorId, url, { ...opts, userAgent: BROWSER_UA, headers: { ...opts.headers, "sec-fetch-mode": "navigate", "sec-fetch-dest": "document", "upgrade-insecure-requests": "1" } });277 if (alt.error || alt.meta.status === 403) return obs;278 return alt;279 }280 if (obs.error && ["timeout", "connection", "fetch_failed"].includes(obs.error.code) && !opts.userAgent && attempt === 0) {281 // Some edges (Akamai) silently reset unknown bot identities at the TLS/HTTP layer.282 const alt = await httpFetch(sensorId, url, { ...opts, userAgent: BROWSER_UA });283 if (!alt.error) return alt;284 }285 const transient = obs.error ? ["timeout", "connection", "body_read_failed", "fetch_failed"].includes(obs.error.code) : obs.meta.status >= 500 && obs.meta.status !== 501;286 if (!transient) return obs;287 if (attempt < retries) await new Promise((r) => setTimeout(r, 800 * (attempt + 1)));288 }289 return last!;290}291