mirror of
https://github.com/bilawalsidhu/gods-eye-view.git
synced 2026-11-04 18:34:02 +00:00
Merge pull request #383: harden Places and CCTV providers
Harden Places input and CCTV provider downloads
This commit is contained in:
9 files changed
+580
-47
No files matched your search
@@ -13,6 +13,14 @@ of current runtime behavior, see [`docs/CURRENT-STATE.md`](docs/CURRENT-STATE.md
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
### Security
|
||||
|
||||
- Validate configured Google Places coordinates and text queries before rate
|
||||
limiting or upstream requests; preserve the keyless capability response.
|
||||
- Bound CCTV media response headers to 15 seconds and cancel error bodies.
|
||||
Cap buffered snapshot downloads at 16 MiB while streaming.
|
||||
|
||||
|
||||
- Cancel the active location lookup when its controls are disposed.
|
||||
|
||||
|
||||
|
||||
@@ -1,5 +1,20 @@
|
||||
# God's Eye View Current State
|
||||
|
||||
## Places and CCTV request bounds
|
||||
|
||||
With a Google key configured, nearby and text search reject missing, blank,
|
||||
non-numeric and out-of-range coordinates before the opt-in limiter and upstream
|
||||
request. Text search also requires a nonblank query. Keyless requests retain
|
||||
their `configured: false` response.
|
||||
|
||||
CCTV media waits at most 15 seconds for upstream response headers and returns
|
||||
504 on timeout. Its timer stops when headers arrive, so live bodies can continue
|
||||
streaming; body idle deadlines are separate from this header deadline. Error
|
||||
responses are cancelled. Buffered snapshots have a 16 MiB streaming cap; an
|
||||
oversized image remains an upstream miss and uses the normal fallback chain.
|
||||
The existing declared media size ceiling remains 64 MiB.
|
||||
|
||||
|
||||
## GBFS upstream bounds
|
||||
|
||||
GBFS refuses upstream redirects and enforces its 5 MiB response cap while
|
||||
|
||||
@@ -8,6 +8,7 @@ import {
|
||||
buildSyntheticCctvSvg,
|
||||
proxyMediaResponse,
|
||||
fetchCctvImageFromUpstream,
|
||||
fetchCctvMediaUpstream,
|
||||
} from './cctv/media.js';
|
||||
import { CCTV_FRAME_FETCH_TIMEOUT_MS } from './cctv/constants.js';
|
||||
import { googleServerApiKey } from './places/google-key.js';
|
||||
@@ -216,11 +217,16 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
|
||||
};
|
||||
const requestRange = req.headers?.range;
|
||||
if (requestRange) upstreamHeaders.Range = requestRange;
|
||||
const upstream = await fetch(mediaUrl, {
|
||||
const upstream = await fetchCctvMediaUpstream(mediaUrl, {
|
||||
headers: upstreamHeaders,
|
||||
});
|
||||
const contentType = upstream.headers.get('content-type') || '';
|
||||
if (!upstream.ok) {
|
||||
try {
|
||||
await upstream.body?.cancel();
|
||||
} catch {
|
||||
/* already closed */
|
||||
}
|
||||
setHealth(cameraId, {
|
||||
status: 'degraded',
|
||||
sourceKind: 'upstream',
|
||||
@@ -270,17 +276,25 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) {
|
||||
});
|
||||
return;
|
||||
} catch (error) {
|
||||
const timedOut =
|
||||
error?.name === 'AbortError' || error?.name === 'TimeoutError';
|
||||
setHealth(cameraId, {
|
||||
status: 'degraded',
|
||||
sourceKind: 'upstream',
|
||||
label: source?.provider || 'Configured source',
|
||||
message: error?.message || 'Media fetch failed',
|
||||
});
|
||||
res.writeHead(502, {
|
||||
res.writeHead(timedOut ? 504 : 502, {
|
||||
'Content-Type': 'application/json',
|
||||
'Cache-Control': 'no-store',
|
||||
});
|
||||
res.end(JSON.stringify({ error: 'Media proxy failed' }));
|
||||
res.end(
|
||||
JSON.stringify({
|
||||
error: timedOut
|
||||
? 'Upstream media timeout'
|
||||
: 'Media proxy failed',
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -38,3 +38,11 @@ export const CCTV_SOURCE_FETCH_TIMEOUT_MS = 15 * 1000;
|
||||
* client refresh cadence. A bounded miss can fall through to Street View or
|
||||
* the synthetic frame instead of leaving the browser preview pending. */
|
||||
export const CCTV_FRAME_FETCH_TIMEOUT_MS = 8 * 1000;
|
||||
|
||||
/** Maximum buffered snapshot size. */
|
||||
export const CCTV_FRAME_MAX_BODY_BYTES = 16 * 1024 * 1024;
|
||||
|
||||
/** Deadline for upstream response headers; live bodies keep streaming afterward. */
|
||||
export const CCTV_MEDIA_FETCH_TIMEOUT_MS = 15 * 1000;
|
||||
/** Declared size ceiling for fixed media responses. */
|
||||
export const CCTV_MEDIA_MAX_BODY_BYTES = 64 * 1024 * 1024;
|
||||
+102
-10
@@ -1,6 +1,11 @@
|
||||
import { Readable } from 'node:stream';
|
||||
import { hashSeed, escapeXml } from './normalize.js';
|
||||
import { CCTV_FRAME_FETCH_TIMEOUT_MS } from './constants.js';
|
||||
import {
|
||||
CCTV_FRAME_FETCH_TIMEOUT_MS,
|
||||
CCTV_FRAME_MAX_BODY_BYTES,
|
||||
CCTV_MEDIA_FETCH_TIMEOUT_MS,
|
||||
CCTV_MEDIA_MAX_BODY_BYTES,
|
||||
} from './constants.js';
|
||||
/**
|
||||
* Generate a synthetic SVG billboard image for a CCTV camera placeholder.
|
||||
*
|
||||
@@ -120,10 +125,9 @@ export async function proxyMediaResponse(
|
||||
// Cheap defense: reject an upstream that DECLARES an oversized fixed body.
|
||||
// Live MJPEG/HLS streams are unbounded by design and send no content-length,
|
||||
// so they pipe normally (piping streams to the client, never buffering).
|
||||
const MEDIA_DECLARED_CAP_BYTES = 64 * 1024 * 1024;
|
||||
if (
|
||||
Number.isFinite(Number(contentLength)) &&
|
||||
Number(contentLength) > MEDIA_DECLARED_CAP_BYTES
|
||||
Number(contentLength) > CCTV_MEDIA_MAX_BODY_BYTES
|
||||
) {
|
||||
res.writeHead(502, {
|
||||
'Content-Type': 'application/json',
|
||||
@@ -153,6 +157,87 @@ export async function proxyMediaResponse(
|
||||
stream.pipe(res);
|
||||
}
|
||||
|
||||
/** Read a snapshot incrementally, retaining at most maxBytes of owned chunks. */
|
||||
async function readCappedResponseBytes(upstream, maxBytes) {
|
||||
const declared = Number(upstream.headers.get('content-length'));
|
||||
if (Number.isFinite(declared) && declared > maxBytes) {
|
||||
try {
|
||||
await upstream.body?.cancel();
|
||||
} catch {
|
||||
/* no-op */
|
||||
}
|
||||
return null;
|
||||
}
|
||||
if (!upstream.body) return null;
|
||||
const chunks = [];
|
||||
let total = 0;
|
||||
// Every retained chunk is an OWNED copy: a chunk can be a small view over a
|
||||
// much larger backing ArrayBuffer, and keeping the view would retain that
|
||||
// whole allocation while the byte accounting only counted the view.
|
||||
const keep = (chunk) => {
|
||||
total += chunk.byteLength;
|
||||
if (total > maxBytes) return false;
|
||||
chunks.push(Buffer.from(chunk));
|
||||
return true;
|
||||
};
|
||||
if (typeof upstream.body[Symbol.asyncIterator] === 'function') {
|
||||
for await (const chunk of upstream.body) {
|
||||
if (!keep(chunk)) {
|
||||
try {
|
||||
await upstream.body.cancel();
|
||||
} catch {
|
||||
/* no-op */
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
return Buffer.concat(chunks, total);
|
||||
}
|
||||
// No async iterator: stream through a reader so the cap still applies while
|
||||
// reading. A body that cannot be streamed at all is refused rather than
|
||||
// buffered uncapped — the helper's whole contract is the cap.
|
||||
const reader =
|
||||
typeof upstream.body.getReader === 'function'
|
||||
? upstream.body.getReader()
|
||||
: null;
|
||||
if (!reader) return null;
|
||||
try {
|
||||
for (;;) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
if (!keep(value)) {
|
||||
try {
|
||||
await reader.cancel();
|
||||
} catch {
|
||||
/* already closed */
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
return Buffer.concat(chunks, total);
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
}
|
||||
}
|
||||
|
||||
/** Open registered media within a header deadline; leave timely live bodies running. */
|
||||
export async function fetchCctvMediaUpstream(
|
||||
url,
|
||||
{
|
||||
headers = {},
|
||||
fetchImpl = fetch,
|
||||
timeoutMs = CCTV_MEDIA_FETCH_TIMEOUT_MS,
|
||||
} = {},
|
||||
) {
|
||||
const controller = new AbortController();
|
||||
const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
|
||||
try {
|
||||
return await fetchImpl(url, { headers, signal: controller.signal });
|
||||
} finally {
|
||||
clearTimeout(timeoutId);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch one upstream CCTV image within the frame-refresh budget.
|
||||
*
|
||||
@@ -164,11 +249,16 @@ export async function proxyMediaResponse(
|
||||
* @param {object} [options]
|
||||
* @param {typeof fetch} [options.fetchImpl=fetch] - Fetch implementation.
|
||||
* @param {number} [options.timeoutMs=CCTV_FRAME_FETCH_TIMEOUT_MS] - Abort timeout.
|
||||
* @param {number} [options.maxBytes=CCTV_FRAME_MAX_BODY_BYTES] - Snapshot byte cap.
|
||||
* @returns {Promise<{ok:true,body:Buffer,contentType:string}|null>}
|
||||
*/
|
||||
export async function fetchCctvImageFromUpstream(
|
||||
url,
|
||||
{ fetchImpl = fetch, timeoutMs = CCTV_FRAME_FETCH_TIMEOUT_MS } = {},
|
||||
{
|
||||
fetchImpl = fetch,
|
||||
timeoutMs = CCTV_FRAME_FETCH_TIMEOUT_MS,
|
||||
maxBytes = CCTV_FRAME_MAX_BODY_BYTES,
|
||||
} = {},
|
||||
) {
|
||||
if (!url || !/^https?:\/\//i.test(url)) return null;
|
||||
const controller = new AbortController();
|
||||
@@ -183,15 +273,17 @@ export async function fetchCctvImageFromUpstream(
|
||||
signal: controller.signal,
|
||||
});
|
||||
const contentType = upstream.headers.get('content-type') || '';
|
||||
if (!upstream.ok || !contentType.startsWith('image/')) return null;
|
||||
return {
|
||||
ok: true,
|
||||
body: Buffer.from(await upstream.arrayBuffer()),
|
||||
contentType,
|
||||
};
|
||||
if (!upstream.ok || !contentType.startsWith('image/')) {
|
||||
controller.abort();
|
||||
return null;
|
||||
}
|
||||
const body = await readCappedResponseBytes(upstream, maxBytes);
|
||||
if (!body) return null;
|
||||
return { ok: true, body, contentType };
|
||||
} catch {
|
||||
return null;
|
||||
} finally {
|
||||
clearTimeout(timeoutId);
|
||||
controller.abort();
|
||||
}
|
||||
}
|
||||
@@ -21,6 +21,27 @@ function googleRateLimiter() {
|
||||
return _googleRateLimiter;
|
||||
}
|
||||
|
||||
/** Validate raw lat/lon presence and WGS84 bounds before consuming request quota. */
|
||||
export function validatePlacesCoordinates(searchParams) {
|
||||
const rawLat = searchParams.get('lat');
|
||||
const rawLon = searchParams.get('lon');
|
||||
if (rawLat === null || rawLon === null || !rawLat.trim() || !rawLon.trim()) {
|
||||
return { ok: false, error: 'lat and lon are required' };
|
||||
}
|
||||
const latitude = Number(rawLat);
|
||||
const longitude = Number(rawLon);
|
||||
if (!Number.isFinite(latitude) || !Number.isFinite(longitude)) {
|
||||
return { ok: false, error: 'Valid lat and lon are required' };
|
||||
}
|
||||
if (latitude < -90 || latitude > 90 || longitude < -180 || longitude > 180) {
|
||||
return {
|
||||
ok: false,
|
||||
error: 'lat must be within [-90, 90] and lon within [-180, 180]',
|
||||
};
|
||||
}
|
||||
return { ok: true, latitude, longitude };
|
||||
}
|
||||
|
||||
/** Nearby place labels and view-biased text search, with request-time key resolution. */
|
||||
export function googlePlacesContextProxy({
|
||||
resolveApiKey = googleServerApiKey,
|
||||
@@ -47,6 +68,16 @@ export function googlePlacesContextProxy({
|
||||
return;
|
||||
}
|
||||
|
||||
const requestUrl = new URL(req.url || '', 'http://localhost');
|
||||
const coordinates = validatePlacesCoordinates(requestUrl.searchParams);
|
||||
if (!coordinates.ok) {
|
||||
res.statusCode = 400;
|
||||
res.setHeader('Content-Type', 'application/json');
|
||||
res.end(JSON.stringify({ error: coordinates.error, places: [] }));
|
||||
return;
|
||||
}
|
||||
const { latitude, longitude } = coordinates;
|
||||
|
||||
// Opt-in per-IP throttle (GEV_RATELIMIT_GOOGLE_PER_MIN). No-op when unset.
|
||||
// Inlined (not the shared helper) so the 429 body keeps this endpoint's
|
||||
// `places: []` contract that the client expects on every error response.
|
||||
@@ -59,24 +90,10 @@ export function googlePlacesContextProxy({
|
||||
return;
|
||||
}
|
||||
|
||||
const requestUrl = new URL(req.url || '', 'http://localhost');
|
||||
const latitude = Number(requestUrl.searchParams.get('lat'));
|
||||
const longitude = Number(requestUrl.searchParams.get('lon'));
|
||||
const radiusM = Math.max(
|
||||
25,
|
||||
Math.min(5000, Number(requestUrl.searchParams.get('radiusM')) || 250),
|
||||
);
|
||||
if (!Number.isFinite(latitude) || !Number.isFinite(longitude)) {
|
||||
res.statusCode = 400;
|
||||
res.setHeader('Content-Type', 'application/json');
|
||||
res.end(
|
||||
JSON.stringify({
|
||||
error: 'Valid lat and lon are required',
|
||||
places: [],
|
||||
}),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const response = await fetch(
|
||||
@@ -160,6 +177,25 @@ export function googlePlacesContextProxy({
|
||||
return;
|
||||
}
|
||||
|
||||
const requestUrl = new URL(req.url || '', 'http://localhost');
|
||||
const textQuery = String(requestUrl.searchParams.get('q') || '').trim();
|
||||
if (!textQuery) {
|
||||
res.statusCode = 400;
|
||||
res.setHeader('Content-Type', 'application/json');
|
||||
res.end(
|
||||
JSON.stringify({ error: 'q, lat and lon are required', places: [] }),
|
||||
);
|
||||
return;
|
||||
}
|
||||
const coordinates = validatePlacesCoordinates(requestUrl.searchParams);
|
||||
if (!coordinates.ok) {
|
||||
res.statusCode = 400;
|
||||
res.setHeader('Content-Type', 'application/json');
|
||||
res.end(JSON.stringify({ error: coordinates.error, places: [] }));
|
||||
return;
|
||||
}
|
||||
const { latitude, longitude } = coordinates;
|
||||
|
||||
// Opt-in per-IP throttle (GEV_RATELIMIT_GOOGLE_PER_MIN). No-op when unset.
|
||||
// Inlined (like nearby-places) so the 429 body keeps the `places: []`
|
||||
// contract the client expects on every error response.
|
||||
@@ -172,26 +208,10 @@ export function googlePlacesContextProxy({
|
||||
return;
|
||||
}
|
||||
|
||||
const requestUrl = new URL(req.url || '', 'http://localhost');
|
||||
const textQuery = String(requestUrl.searchParams.get('q') || '').trim();
|
||||
const latitude = Number(requestUrl.searchParams.get('lat'));
|
||||
const longitude = Number(requestUrl.searchParams.get('lon'));
|
||||
const radiusM = Math.max(
|
||||
50,
|
||||
Math.min(50000, Number(requestUrl.searchParams.get('radiusM')) || 4000),
|
||||
);
|
||||
if (
|
||||
!textQuery ||
|
||||
!Number.isFinite(latitude) ||
|
||||
!Number.isFinite(longitude)
|
||||
) {
|
||||
res.statusCode = 400;
|
||||
res.setHeader('Content-Type', 'application/json');
|
||||
res.end(
|
||||
JSON.stringify({ error: 'q, lat and lon are required', places: [] }),
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
try {
|
||||
const response = await fetch(
|
||||
|
||||
+225
-2
@@ -1,9 +1,27 @@
|
||||
import { CCTV_FRAME_FETCH_TIMEOUT_MS, CCTV_FRAME_MAX_BODY_BYTES, CCTV_MEDIA_FETCH_TIMEOUT_MS, CCTV_MEDIA_MAX_BODY_BYTES } from '../../server/providers/cctv/constants.js';
|
||||
import { test } from 'node:test';
|
||||
import assert from 'node:assert/strict';
|
||||
import {
|
||||
CCTV_FRAME_FETCH_TIMEOUT_MS,
|
||||
fetchCctvImageFromUpstream,
|
||||
} from '../../vite.config.js';
|
||||
fetchCctvMediaUpstream,
|
||||
} from '../../server/providers/cctv/media.js';
|
||||
|
||||
/** A body that arrives in chunks and never declares a Content-Length. */
|
||||
function chunkedImageResponse(chunkBytes, chunkCount, { onChunk = () => {}, onCancel = () => {} } = {}) {
|
||||
const stream = new ReadableStream({
|
||||
async pull(controller) {
|
||||
if (chunkCount <= 0) {
|
||||
controller.close();
|
||||
return;
|
||||
}
|
||||
chunkCount -= 1;
|
||||
onChunk();
|
||||
controller.enqueue(new Uint8Array(chunkBytes));
|
||||
},
|
||||
cancel: onCancel,
|
||||
});
|
||||
return new Response(stream, { status: 200, headers: { 'Content-Type': 'image/jpeg' } });
|
||||
}
|
||||
|
||||
test('CCTV upstream frame fetch supplies a bounded abort signal', async () => {
|
||||
let observedSignal = null;
|
||||
@@ -36,3 +54,208 @@ test('CCTV upstream frame fetch returns a valid image response', async () => {
|
||||
assert.equal(result?.contentType, 'image/jpeg');
|
||||
assert.deepEqual(result?.body, Buffer.from([1, 2, 3]));
|
||||
});
|
||||
|
||||
test('CCTV upstream frame fetch rejects a declared oversize body without draining it', async () => {
|
||||
const chunkBytes = 64 * 1024;
|
||||
let pulled = 0;
|
||||
let cancelled = false;
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 1000,
|
||||
maxBytes: chunkBytes * 4,
|
||||
fetchImpl: async () => {
|
||||
const response = chunkedImageResponse(chunkBytes, 64, {
|
||||
onChunk: () => { pulled += 1; },
|
||||
onCancel: () => { cancelled = true; },
|
||||
});
|
||||
response.headers.set('Content-Length', String(chunkBytes * 64));
|
||||
return response;
|
||||
},
|
||||
});
|
||||
|
||||
// Assert on the byte count, not the object: a regressed proxy returns a
|
||||
// multi-megabyte Buffer here, and diffing one into the failure report is
|
||||
// slower than the check it is reporting on.
|
||||
assert.equal(result?.body?.length ?? null, null, 'a declared oversize snapshot is a miss, not a buffered body');
|
||||
assert.equal(cancelled, true, 'the declared cap is enforced by cancelling, not reading');
|
||||
// Only the stream's own one-chunk prefetch may have run; the proxy pulls none.
|
||||
assert.ok(pulled <= 1, `declared cap short-circuits the read, pulled ${pulled} chunks`);
|
||||
});
|
||||
|
||||
test('CCTV upstream frame fetch aborts an undeclared body once it crosses the cap', async () => {
|
||||
const chunkBytes = 64 * 1024;
|
||||
let pulled = 0;
|
||||
let cancelled = false;
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 1000,
|
||||
maxBytes: chunkBytes * 4,
|
||||
// Sixty-four chunks are offered with no Content-Length; a proxy that
|
||||
// buffers first would take all of them.
|
||||
fetchImpl: async () => chunkedImageResponse(chunkBytes, 64, {
|
||||
onChunk: () => { pulled += 1; },
|
||||
onCancel: () => { cancelled = true; },
|
||||
}),
|
||||
});
|
||||
|
||||
assert.equal(result?.body?.length ?? null, null, 'a chunked body over the cap is a miss');
|
||||
// Four chunks fit, the fifth crosses the cap, and one more sits in the
|
||||
// stream's prefetch queue — nothing past that is ever pulled.
|
||||
assert.ok(pulled <= 6, `stopped reading at the cap, pulled ${pulled} chunks`);
|
||||
assert.equal(cancelled, true, 'the upstream stream is cancelled, not drained');
|
||||
});
|
||||
|
||||
test('CCTV upstream frame fetch still returns a chunked body under the cap', async () => {
|
||||
const chunkBytes = 1024;
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 1000,
|
||||
maxBytes: chunkBytes * 8,
|
||||
fetchImpl: async () => chunkedImageResponse(chunkBytes, 3),
|
||||
});
|
||||
|
||||
assert.equal(result?.ok, true);
|
||||
assert.equal(result?.body?.length, chunkBytes * 3, 'every chunk is reassembled in order');
|
||||
assert.ok(CCTV_FRAME_MAX_BODY_BYTES > 0 && CCTV_FRAME_MAX_BODY_BYTES <= CCTV_MEDIA_MAX_BODY_BYTES,
|
||||
'the frame cap must be at or under the media route cap — pinned against the real constant');
|
||||
});
|
||||
|
||||
test('CCTV media upstream fetch aborts when response headers never arrive', async () => {
|
||||
let observedSignal = null;
|
||||
const startedAt = Date.now();
|
||||
await assert.rejects(
|
||||
fetchCctvMediaUpstream('https://example.com/stream.m3u8', {
|
||||
timeoutMs: 25,
|
||||
fetchImpl: (url, { signal }) => new Promise((resolve, reject) => {
|
||||
observedSignal = signal;
|
||||
signal.addEventListener('abort', () => reject(Object.assign(new Error('aborted'), { name: 'AbortError' })));
|
||||
}),
|
||||
}),
|
||||
(error) => error.name === 'AbortError',
|
||||
);
|
||||
assert.equal(observedSignal?.aborted, true, 'the header deadline aborts the attempt');
|
||||
assert.ok(Date.now() - startedAt < 2_000, 'the deadline is the injected one, not the production one');
|
||||
assert.ok(CCTV_MEDIA_FETCH_TIMEOUT_MS >= 10_000, 'production keeps a generous header deadline for slow cameras');
|
||||
});
|
||||
|
||||
test('CCTV media upstream fetch never cuts a stream whose headers arrived in time', async () => {
|
||||
let observedSignal = null;
|
||||
const upstream = await fetchCctvMediaUpstream('https://example.com/stream.m3u8', {
|
||||
timeoutMs: 20,
|
||||
fetchImpl: async (url, { signal }) => {
|
||||
observedSignal = signal;
|
||||
return { ok: true, status: 200, headers: new Map([['content-type', 'video/mp2t']]), body: null };
|
||||
},
|
||||
});
|
||||
assert.equal(upstream.ok, true);
|
||||
// Well past the header deadline: the timer was cleared at header arrival,
|
||||
// so the live body is still allowed to flow.
|
||||
await new Promise((resolve) => setTimeout(resolve, 60));
|
||||
assert.equal(observedSignal?.aborted, false, 'a slow body after timely headers is never aborted here');
|
||||
});
|
||||
|
||||
test('CCTV upstream frame fetch keeps a body that lands exactly on the cap', async () => {
|
||||
const chunkBytes = 512;
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 1000,
|
||||
maxBytes: chunkBytes * 4,
|
||||
fetchImpl: async () => chunkedImageResponse(chunkBytes, 4),
|
||||
});
|
||||
assert.equal(result?.ok, true, 'exactly-at-cap is under the ceiling, not over it');
|
||||
assert.equal(result?.body?.length, chunkBytes * 4);
|
||||
});
|
||||
|
||||
test('CCTV upstream frame fetch ignores a garbage Content-Length and still caps the stream', async () => {
|
||||
const chunkBytes = 1024;
|
||||
let cancelled = false;
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 1000,
|
||||
maxBytes: chunkBytes * 2,
|
||||
fetchImpl: async () => {
|
||||
const response = chunkedImageResponse(chunkBytes, 8, { onCancel: () => { cancelled = true; } });
|
||||
response.headers.set('Content-Length', 'not-a-number');
|
||||
return response;
|
||||
},
|
||||
});
|
||||
assert.equal(result?.body?.length ?? null, null, 'an unparseable length falls back to counting bytes');
|
||||
assert.equal(cancelled, true);
|
||||
});
|
||||
|
||||
test('CCTV upstream frame fetch retains owned bytes, not the chunk\'s backing allocation', async () => {
|
||||
const backing = new ArrayBuffer(4 * 1024 * 1024);
|
||||
const view = new Uint8Array(backing, 8, 3);
|
||||
view.set([7, 8, 9]);
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 1000,
|
||||
maxBytes: 1024,
|
||||
fetchImpl: async () => new Response(new ReadableStream({
|
||||
start(controller) {
|
||||
controller.enqueue(view);
|
||||
controller.close();
|
||||
},
|
||||
}), { status: 200, headers: { 'Content-Type': 'image/jpeg' } }),
|
||||
});
|
||||
assert.equal(result?.ok, true);
|
||||
assert.deepEqual(Array.from(result.body), [7, 8, 9]);
|
||||
assert.ok(result.body.buffer.byteLength < backing.byteLength,
|
||||
'the 4 MB backing allocation is not retained behind a 3-byte frame');
|
||||
});
|
||||
|
||||
test('CCTV upstream frame fetch refuses a body it cannot stream instead of buffering it uncapped', async () => {
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 1000,
|
||||
maxBytes: 1024,
|
||||
fetchImpl: async () => ({
|
||||
ok: true,
|
||||
status: 200,
|
||||
headers: new Map([['content-type', 'image/jpeg']]),
|
||||
body: { /* neither async-iterable nor reader-backed */ },
|
||||
arrayBuffer: async () => { throw new Error('arrayBuffer must never be used as an uncapped fallback'); },
|
||||
}),
|
||||
});
|
||||
assert.equal(result, null);
|
||||
});
|
||||
|
||||
test('reader-only snapshots release their lock on success, overflow and read failure', async () => {
|
||||
for (const mode of ['success', 'overflow', 'failure']) {
|
||||
const body = new ReadableStream({
|
||||
pull(controller) {
|
||||
if (mode === 'failure') controller.error(new Error('read failed'));
|
||||
else { controller.enqueue(new Uint8Array(4)); controller.close(); }
|
||||
},
|
||||
});
|
||||
Object.defineProperty(body, Symbol.asyncIterator, { value: undefined });
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
maxBytes: mode === 'overflow' ? 2 : 8,
|
||||
fetchImpl: async () => ({ ok: true, headers: new Headers({ 'content-type': 'image/jpeg' }), body }),
|
||||
});
|
||||
assert.equal(body.locked, false, mode);
|
||||
assert.equal(result?.body.length ?? null, mode === 'success' ? 4 : null);
|
||||
}
|
||||
});
|
||||
|
||||
test('the snapshot deadline remains active after headers while reading a stalled body', async () => {
|
||||
let signal;
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
timeoutMs: 20,
|
||||
fetchImpl: async (_url, init) => {
|
||||
signal = init.signal;
|
||||
return new Response(new ReadableStream({ start(controller) {
|
||||
signal.addEventListener('abort', () => controller.error(signal.reason), { once: true });
|
||||
}}), { headers: { 'content-type': 'image/jpeg' } });
|
||||
},
|
||||
});
|
||||
assert.equal(result, null);
|
||||
assert.equal(signal.aborted, true);
|
||||
});
|
||||
|
||||
test('rejected snapshot responses abort the upstream download', async () => {
|
||||
for (const [status, contentType] of [[503, 'image/jpeg'], [200, 'text/html']]) {
|
||||
let signal;
|
||||
const result = await fetchCctvImageFromUpstream('https://example.com/frame.jpg', {
|
||||
fetchImpl: async (_url, init) => {
|
||||
signal = init.signal;
|
||||
return new Response('rejected', { status, headers: { 'content-type': contentType } });
|
||||
},
|
||||
});
|
||||
assert.equal(result, null);
|
||||
assert.equal(signal.aborted, true);
|
||||
}
|
||||
});
|
||||
@@ -0,0 +1,106 @@
|
||||
import assert from 'node:assert/strict';
|
||||
import test from 'node:test';
|
||||
import { googlePlacesContextProxy, validatePlacesCoordinates } from '../server/providers/places/google.js';
|
||||
|
||||
function installGooglePlacesRoutes() {
|
||||
const routes = new Map();
|
||||
googlePlacesContextProxy({ resolveApiKey: () => process.env.GOOGLE_MAPS_API_KEY }).configureServer({
|
||||
middlewares: {
|
||||
use(path, handler) {
|
||||
routes.set(path, handler);
|
||||
},
|
||||
},
|
||||
});
|
||||
return routes;
|
||||
}
|
||||
|
||||
function invokeRoute(handler, { method = 'GET', url = '/', remoteAddress = '127.0.0.1' } = {}) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const headers = new Map();
|
||||
const req = { method, url, headers: {}, socket: { remoteAddress } };
|
||||
const res = {
|
||||
statusCode: 200,
|
||||
setHeader(name, value) { headers.set(String(name).toLowerCase(), String(value)); },
|
||||
end(body = '') {
|
||||
resolve({ statusCode: this.statusCode, headers: Object.fromEntries(headers), body: body ? JSON.parse(String(body)) : null });
|
||||
},
|
||||
};
|
||||
Promise.resolve(handler(req, res)).catch(reject);
|
||||
});
|
||||
}
|
||||
|
||||
test('validatePlacesCoordinates refuses missing, blank, non-numeric, and out-of-range values', () => {
|
||||
const check = (query) => validatePlacesCoordinates(new URLSearchParams(query));
|
||||
for (const bad of ['', 'lat=30.27', 'lon=-97.74', 'lat=&lon=', 'lat=%20&lon=-97.74', 'lat=abc&lon=-97.74',
|
||||
'lat=Infinity&lon=-97.74', 'lat=NaN&lon=0', 'lat=90.001&lon=0', 'lat=-90.001&lon=0', 'lat=0&lon=180.5', 'lat=0&lon=-180.5']) {
|
||||
assert.equal(check(bad).ok, false, `${bad || '(empty)'} must be refused`);
|
||||
}
|
||||
// Inclusive boundaries are real places (the poles, the antimeridian).
|
||||
assert.deepEqual(check('lat=90&lon=-180'), { ok: true, latitude: 90, longitude: -180 });
|
||||
assert.deepEqual(check('lat=-90&lon=180'), { ok: true, latitude: -90, longitude: 180 });
|
||||
assert.deepEqual(check('lat=30.27&lon=-97.74'), { ok: true, latitude: 30.27, longitude: -97.74 });
|
||||
});
|
||||
|
||||
test('both Places routes answer bad coordinates with a 400 before the limiter and before Google', async () => {
|
||||
const previousKey = process.env.GOOGLE_MAPS_API_KEY;
|
||||
const previousLimit = process.env.GEV_RATELIMIT_GOOGLE_PER_MIN;
|
||||
const originalFetch = globalThis.fetch;
|
||||
process.env.GOOGLE_MAPS_API_KEY = 'test-key';
|
||||
process.env.GEV_RATELIMIT_GOOGLE_PER_MIN = '1';
|
||||
let upstreamCalls = 0;
|
||||
globalThis.fetch = async () => {
|
||||
upstreamCalls += 1;
|
||||
return {
|
||||
ok: true,
|
||||
status: 200,
|
||||
headers: new Map(),
|
||||
json: async () => ({ places: [] }),
|
||||
text: async () => '{"places":[]}',
|
||||
};
|
||||
};
|
||||
try {
|
||||
const routes = installGooglePlacesRoutes();
|
||||
const nearby = routes.get('/api/google/nearby-places');
|
||||
const textSearch = routes.get('/api/google/text-search');
|
||||
const badQueries = ['', 'lat=30.27', 'lat=&lon=', 'lat=abc&lon=-97.74', 'lat=999&lon=-200'];
|
||||
for (const query of badQueries) {
|
||||
const nearbyResponse = await invokeRoute(nearby, { url: `/?${query}` });
|
||||
assert.equal(nearbyResponse.statusCode, 400, `nearby-places must refuse ${query || '(empty)'}`);
|
||||
assert.deepEqual(nearbyResponse.body.places, []);
|
||||
const textResponse = await invokeRoute(textSearch, { url: `/?q=capitol&${query}` });
|
||||
assert.equal(textResponse.statusCode, 400, `text-search must refuse ${query || '(empty)'}`);
|
||||
assert.deepEqual(textResponse.body.places, []);
|
||||
}
|
||||
const missingQuery = await invokeRoute(textSearch, { url: '/?lat=30.27&lon=-97.74' });
|
||||
assert.equal(missingQuery.statusCode, 400, 'text-search still requires q');
|
||||
assert.equal(upstreamCalls, 0, 'invalid input never reaches Google');
|
||||
|
||||
// The limiter allows ONE request per minute per client. None of the
|
||||
// rejected requests may have consumed it, so the first valid request
|
||||
// still goes upstream instead of meeting a 429.
|
||||
const valid = await invokeRoute(nearby, { url: '/?lat=30.27&lon=-97.74' });
|
||||
assert.notEqual(valid.statusCode, 429, 'malformed requests must not consume limiter capacity');
|
||||
assert.equal(upstreamCalls, 1, 'a valid request reaches Google exactly once');
|
||||
} finally {
|
||||
globalThis.fetch = originalFetch;
|
||||
if (previousKey === undefined) delete process.env.GOOGLE_MAPS_API_KEY;
|
||||
else process.env.GOOGLE_MAPS_API_KEY = previousKey;
|
||||
if (previousLimit === undefined) delete process.env.GEV_RATELIMIT_GOOGLE_PER_MIN;
|
||||
else process.env.GEV_RATELIMIT_GOOGLE_PER_MIN = previousLimit;
|
||||
}
|
||||
});
|
||||
|
||||
test('preview validates both Places routes and keeps the keyless response', async () => {
|
||||
for (const key of ['', 'test-key']) {
|
||||
const routes = new Map();
|
||||
googlePlacesContextProxy({ resolveApiKey: () => key }).configurePreviewServer({
|
||||
middlewares: { use: (path, handler) => routes.set(path, handler) },
|
||||
});
|
||||
for (const name of ['nearby-places', 'text-search']) {
|
||||
const result = await invokeRoute(routes.get(`/api/google/${name}`), { url: '/?q=capitol&lat=&lon=', remoteAddress: 'preview-test' });
|
||||
assert.equal(result.statusCode, key ? 400 : 200);
|
||||
assert.deepEqual(result.body.places, []);
|
||||
if (!key) assert.equal(result.body.configured, false);
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -7,9 +7,9 @@ import { cctvProxy } from '../../server/providers/cctv.js';
|
||||
import { radioBrowserProxy } from '../../server/providers/radio.js';
|
||||
import { localProviderPlugins } from '../../server/providers/local.js';
|
||||
|
||||
function install(plugin) {
|
||||
function install(plugin, hook = 'configureServer') {
|
||||
let handler;
|
||||
plugin.configureServer({
|
||||
plugin[hook]({
|
||||
middlewares: {
|
||||
use(_route, fn) {
|
||||
handler = fn;
|
||||
@@ -102,3 +102,50 @@ test('composition creates exactly one CCTV and radio provider without acquisitio
|
||||
assert.equal(plugins.filter((p) => p.name === factory().name).length, 1);
|
||||
}
|
||||
});
|
||||
|
||||
for (const hook of ['configureServer', 'configurePreviewServer']) {
|
||||
test(`CCTV ${hook} maps header timeouts and cancels upstream error bodies`, async (t) => {
|
||||
isolate(t);
|
||||
process.env.CCTV_SOURCES_JSON = JSON.stringify([
|
||||
{
|
||||
id: 'bounded',
|
||||
name: 'Test',
|
||||
lat: 30,
|
||||
lon: -97,
|
||||
feedType: 'video',
|
||||
url: 'https://camera.example.org/live.mp4',
|
||||
},
|
||||
]);
|
||||
const request = install(
|
||||
cctvProxy({ sourceRoot: fixture(t, 'unused') }),
|
||||
hook,
|
||||
);
|
||||
// Resolve the catalog before substituting transport behavior.
|
||||
await request('/sources');
|
||||
t.mock.method(globalThis, 'fetch', async () => {
|
||||
throw new DOMException('timeout', 'AbortError');
|
||||
});
|
||||
const timeout = await request('/media/bounded');
|
||||
assert.equal(timeout.status, 504);
|
||||
assert.deepEqual(JSON.parse(timeout.body), {
|
||||
error: 'Upstream media timeout',
|
||||
});
|
||||
let cancelled = false;
|
||||
t.mock.method(
|
||||
globalThis,
|
||||
'fetch',
|
||||
async () =>
|
||||
new Response(
|
||||
new ReadableStream({
|
||||
cancel() {
|
||||
cancelled = true;
|
||||
},
|
||||
}),
|
||||
{ status: 503 },
|
||||
),
|
||||
);
|
||||
const error = await request('/media/bounded');
|
||||
assert.equal(error.status, 503);
|
||||
assert.equal(cancelled, true);
|
||||
});
|
||||
}
|
||||
Reference in new issue
Block a user