fix(cctv): admit and bound HLS leases by peer

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');

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 () => {