From 167aa947ce338d90af77eb22af04d08be25eea47 Mon Sep 17 00:00:00 2001 From: Said Date: Tue, 20 Oct 2026 00:11:31 +0100 Subject: [PATCH] fix(cctv): admit and bound HLS leases by peer --- docs/CURRENT-STATE.md | 7 ++++ scripts/package-boundaries.json | 2 + server/providers/cctv.js | 23 +++++++++-- server/providers/cctv/stream.js | 59 ++++++++++++++++++++++------- src/data/cctvHlsStream.test.mjs | 37 ++++++++++++++++++ src/data/cctvMediaFeedType.test.mjs | 43 +++++++++++++++++++++ 6 files changed, 154 insertions(+), 17 deletions(-) diff --git a/docs/CURRENT-STATE.md b/docs/CURRENT-STATE.md index 55b7a63ab..9d674d105 100644 --- a/docs/CURRENT-STATE.md +++ b/docs/CURRENT-STATE.md @@ -1,5 +1,12 @@ # God's Eye View Current State +CCTV live-media requests are same-site gated before catalog or media work, including +HLS renewal, segment reads and DELETE cleanup. HLS leases belong to their socket +peer (IPv4-mapped IPv6 is normalized), with eight leases per peer across the two +global sessions. Peers behind the same NAT share that budget; this is resource +admission, not authentication. Native same-origin video and range requests remain +supported. + Shared distance checks use `src/geoDistance.js` with `{ lat, lon }` points and the IUGG mean spherical Earth radius, 6371.0088 km. `greatCircleKm` is canonical; `greatCircleMeters` is its exact metre conversion. Missing, diff --git a/scripts/package-boundaries.json b/scripts/package-boundaries.json index ca5dc0c7f..fc9e4b5e9 100644 --- a/scripts/package-boundaries.json +++ b/scripts/package-boundaries.json @@ -187,6 +187,8 @@ "exports": ["./server/providers/cctv"], "modules": [ "server/providers/cctv.js", + "server/providers/common/same-site.js", + "src/localRequestGate.mjs", "server/providers/cctv/cap.js", "server/providers/cctv/groundHeights.js", "server/providers/cctv/catalog.js", diff --git a/server/providers/cctv.js b/server/providers/cctv.js index e549753e8..fc8f4f692 100644 --- a/server/providers/cctv.js +++ b/server/providers/cctv.js @@ -17,6 +17,7 @@ import { CCTV_MAX_SOURCES_CEILING, } from './cctv/constants.js'; import { sanitizeCctvRangeHeader } from './cctv/range.js'; +import { admitSameSite } from './common/same-site.js'; import { createHlsPuller } from './cctv/stream.js'; import { googleServerApiKey } from './places/google-key.js'; export { CCTV_FRAME_FETCH_TIMEOUT_MS, fetchCctvImageFromUpstream }; @@ -135,11 +136,18 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { }); server.middlewares.use('/api/cctv', async (req, res) => { try { + const url = new URL(req.url || '/', 'http://localhost'); + // Refuse before catalog I/O, session creation, renewal or release. + if (url.pathname.startsWith('/media/') && admitSameSite(req, res)) + return; + const clientId = String(req.socket?.remoteAddress || 'unknown').replace( + /^::ffff:/, + '', + ); const sources = await getCctvSources(); const sourceById = new Map( sources.map((source) => [source.id, source]), ); - const url = new URL(req.url || '/', 'http://localhost'); if (url.pathname === '/sources') { const body = { @@ -238,7 +246,7 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { return; } if (req.method === 'DELETE') { - if (leaseId) puller.release(cameraId, leaseId); + if (leaseId) puller.release(cameraId, leaseId, clientId); res.writeHead(204); res.end(); return; @@ -268,6 +276,7 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { url.searchParams.get('session'), Number(match[3]), leaseId, + clientId, ); res.writeHead(body ? 200 : 404, { 'Content-Type': 'video/mp2t', @@ -279,10 +288,15 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { const downstream = watchDownstreamClose(res); let entry; const cancelPending = () => { - if (entry) puller.release(cameraId, leaseId); + if (entry) puller.release(cameraId, leaseId, clientId); }; try { - entry = await puller.ensure(cameraId, mediaUrl, leaseId); + entry = await puller.ensure( + cameraId, + mediaUrl, + leaseId, + clientId, + ); if (downstream.closed) { cancelPending(); return; @@ -296,6 +310,7 @@ export function cctvProxy({ sourceRoot = process.cwd() } = {}) { entry, cameraId, leaseId, + clientId, ); if (downstream.closed) return; if (!playlist) throw new Error('Stream unavailable'); diff --git a/server/providers/cctv/stream.js b/server/providers/cctv/stream.js index 14d1743bb..46608f294 100644 --- a/server/providers/cctv/stream.js +++ b/server/providers/cctv/stream.js @@ -4,6 +4,7 @@ import { randomUUID } from 'node:crypto'; export const HLS_LIMITS = Object.freeze({ sessions: 2, leasesPerSession: 8, + leasesPerClient: 8, playlistBytes: 256 * 1024, segmentBytes: 4 * 1024 * 1024, sessionBytes: 24 * 1024 * 1024, @@ -138,25 +139,37 @@ export function createHlsPuller({ entry.upstream.clear(); entry.bytes = 0; }; - const release = (cameraId, leaseId) => { + const release = (cameraId, leaseId, clientId = 'legacy') => { const entry = active.get(cameraId); const lease = entry?.leases.get(leaseId); - if (!lease) return; + if (!lease || lease.clientId !== clientId) return; clearTimeout(lease.timer); entry.leases.delete(leaseId); if (!entry.leases.size) stop(cameraId, entry.token); }; - const touch = (entry, leaseId, create = false) => { + const touch = (entry, leaseId, create = false, clientId = 'legacy') => { let lease = entry.leases.get(leaseId); + if (lease && lease.clientId !== clientId) + throw new Error('HLS lease belongs to another client'); if (!lease) { + const clientLeases = [...active.values()].reduce( + (count, session) => + count + + [...session.leases.values()].filter( + (item) => item.clientId === clientId, + ).length, + 0, + ); + if (clientLeases >= limits.leasesPerClient) + throw new Error('HLS client lease capacity reached'); if (!create || entry.leases.size >= limits.leasesPerSession) throw new Error('HLS lease capacity reached'); - lease = {}; + lease = { clientId }; entry.leases.set(leaseId, lease); } clearTimeout(lease.timer); lease.timer = setTimeout( - () => release(entry.cameraId, leaseId), + () => release(entry.cameraId, leaseId, clientId), limits.idleMs, ); lease.timer.unref?.(); @@ -248,7 +261,12 @@ export function createHlsPuller({ } } }; - const ensure = async (cameraId, url, leaseId = 'legacy') => { + const ensure = async ( + cameraId, + url, + leaseId = 'legacy', + clientId = 'legacy', + ) => { if (closed) throw new Error('HLS service closed'); sameOriginHlsUrl(url, url); let entry = active.get(cameraId); @@ -257,7 +275,7 @@ export function createHlsPuller({ entry = null; } if (entry) { - touch(entry, leaseId, true); + touch(entry, leaseId, true, clientId); return entry; } // Reserve synchronously, before any await: concurrent creation is singleflight. @@ -279,8 +297,8 @@ export function createHlsPuller({ pendingDiscontinuity: false, discontinuitiesRemoved: 0, }; + touch(entry, leaseId, true, clientId); active.set(cameraId, entry); - touch(entry, leaseId, true); entry.polling = poll(entry); return entry; }; @@ -292,9 +310,14 @@ export function createHlsPuller({ } return false; }; - const buildPlaylist = async (entry, cameraId, leaseId = 'legacy') => { + const buildPlaylist = async ( + entry, + cameraId, + leaseId = 'legacy', + clientId = 'legacy', + ) => { if (entry.stopping || entry.segments.size < 2) return null; - touch(entry, leaseId); + touch(entry, leaseId, false, clientId); const segments = [...entry.segments.values()].sort((a, b) => a.seq - b.seq); const lines = [ '#EXTM3U', @@ -318,12 +341,22 @@ export function createHlsPuller({ } return lines.join('\n') + '\n'; }; - const getSegment = (cameraId, token, seq, leaseId = 'legacy') => { + const getSegment = ( + cameraId, + token, + seq, + leaseId = 'legacy', + clientId = 'legacy', + ) => { const entry = active.get(cameraId); - if (!entry || entry.token !== token || !entry.leases.has(leaseId)) + if ( + !entry || + entry.token !== token || + entry.leases.get(leaseId)?.clientId !== clientId + ) return null; const body = entry.segments.get(seq)?.body; - if (body) touch(entry, leaseId); + if (body) touch(entry, leaseId, false, clientId); return body || null; }; const shutdown = async () => { diff --git a/src/data/cctvHlsStream.test.mjs b/src/data/cctvHlsStream.test.mjs index a087eebdf..01f5b15e4 100644 --- a/src/data/cctvHlsStream.test.mjs +++ b/src/data/cctvHlsStream.test.mjs @@ -223,3 +223,40 @@ test('an abandoned consumer expires while a renewed consumer keeps the session', assert.equal(entry.controller.signal.aborted, false); await manager.shutdown(); }); + +test('client lease ownership covers refresh, segments and release, with a bounded client budget', async (t) => { + const manager = createHlsPuller({ + limits: { ...HLS_LIMITS, leasesPerClient: 1, pollMs: 100000 }, + fetchImpl: async (url) => + new Response(url.endsWith('.m3u8') ? playlist : 'abc'), + }); + t.after(() => manager.shutdown()); + const entry = await manager.ensure('a', base, 'lease-a', 'peer-a'); + await manager.waitReady(entry); + await assert.rejects( + manager.ensure('a', base, 'lease-a', 'peer-b'), + /another client/, + ); + assert.equal( + manager.getSegment('a', entry.token, 0, 'lease-a', 'peer-b'), + null, + ); + await assert.rejects( + manager.buildPlaylist(entry, 'a', 'lease-a', 'peer-b'), + /another client/, + ); + manager.release('a', 'lease-a', 'peer-b'); + assert.equal(entry.leases.size, 1); + await assert.rejects( + manager.ensure('b', base, 'lease-b', 'peer-a'), + /client lease capacity/, + ); + assert.equal( + manager.stats().sessions, + 1, + 'failed admission reserves no session', + ); + await manager.ensure('b', base, 'lease-b', 'peer-b'); + manager.release('a', 'lease-a', 'peer-a'); + assert.equal(entry.stopping, true); +}); diff --git a/src/data/cctvMediaFeedType.test.mjs b/src/data/cctvMediaFeedType.test.mjs index 5791e433b..e6ba9e3eb 100644 --- a/src/data/cctvMediaFeedType.test.mjs +++ b/src/data/cctvMediaFeedType.test.mjs @@ -268,3 +268,46 @@ test('/frame does not follow a still camera off its origin', async (t) => { [STILL.url], ); }); + +test('cross-site media creation, segment reads and release fail before any catalog or media fetch', async (t) => { + let handler; + const originalFetch = globalThis.fetch; + let calls = 0; + globalThis.fetch = async () => { + calls++; + throw new Error('must not fetch'); + }; + t.after(() => { + globalThis.fetch = originalFetch; + }); + cctvProxy().configureServer({ + middlewares: { + use: (_path, fn) => { + handler = fn; + }, + }, + }); + for (const headers of [ + { origin: 'https://foreign.example' }, + { origin: 'null' }, + { 'sec-fetch-site': 'cross-site' }, + { 'sec-fetch-site': 'same-site' }, + ]) { + for (const [method, url] of [ + ['GET', '/media/camera?lease=test'], + ['DELETE', '/media/camera?lease=test'], + ['GET', '/media/camera/seg_0.ts?lease=test'], + ]) { + const { res } = recordingResponse(); + res.setHeader = (name, value) => { + res.headers[name] = value; + }; + await handler( + { method, url, headers: { host: 'localhost:4173', ...headers } }, + res, + ); + assert.equal(res.statusCode, 403); + } + } + assert.equal(calls, 0); +});