Integrate PR #1233 while preserving the shared 30-second timeout

changelog.d/1233-tarktee-datex-catalog-hardening.md

@@ -0,0 +1,4 @@
+- Harden the Tarktee (Estonia) DATEX2 camera catalog fetches: refuse redirects,
+  cancel both response streams when either request fails or overflows the 4 MB
+  byte cap, retain the shared 30 s dual-XML fetch deadline, and deduplicate camera
+  IDs before per-pack prioritization (yaka0007, #1233).

server/providers/cctv/constants.js

@@ -188,6 +188,8 @@ export const TARKTEE_IMAGES_URL =
   'https://tarktee.transpordiamet.ee/api/v1/datex/roadCameraImages';
 export const TARKTEE_IMAGE_ORIGIN = 'https://tarktee.transpordiamet.ee/images/';
 export const DEFAULT_TARKTEE_MAX_SOURCES = 179;
+/** Hard ceiling per Tarktee DATEX XML document (~120–165 KB live). */
+export const TARKTEE_MAX_CATALOG_BYTES = 4 * 1024 * 1024;
 export const TARKTEE_ANCHORS = [
   { lat: 59.437, lon: 24.753 }, // Tallinn
   { lat: 58.378, lon: 26.729 }, // Tartu

server/providers/cctv/sources.js

@@ -42,6 +42,8 @@ import {
   TARKTEE_IMAGES_URL,
   TARKTEE_IMAGE_ORIGIN,
   DEFAULT_TARKTEE_MAX_SOURCES,
+  TARKTEE_MAX_CATALOG_BYTES,
+  TARKTEE_FETCH_TIMEOUT_MS,
   TARKTEE_ANCHORS,
   DEFAULT_WARENDORF_SOURCE_FILE,
   WARENDORF_IMAGE_ORIGINS,
@@ -70,7 +72,6 @@ import {
   TORONTO_GROUND_ELEVATION_M,
   TORONTO_MAX_CATALOG_BYTES,
   CCTV_SOURCE_FETCH_TIMEOUT_MS,
-  TARKTEE_FETCH_TIMEOUT_MS,
   DELDOT_CCTV_URL,
   DEFAULT_DELDOT_MAX_SOURCES,
   DELDOT_ANCHORS,
@@ -1195,25 +1196,62 @@ export async function loadTarkteeSourcesFromDatex() {
     const [locResp, imgResp] = await Promise.all([
       fetch(TARKTEE_LOCATIONS_URL, {
         headers: { Accept: 'application/xml,text/xml,*/*' },
+        redirect: 'manual',
         signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS),
       }),
       fetch(TARKTEE_IMAGES_URL, {
         headers: { Accept: 'application/xml,text/xml,*/*' },
+        redirect: 'manual',
         signal: AbortSignal.timeout(TARKTEE_FETCH_TIMEOUT_MS),
       }),
     ]);
+    // Both responses are fetched in parallel; if either is rejected or overflows
+    // its byte cap, both bodies must be released so neither transport leaks.
+    const discardBoth = async () => {
+      await Promise.allSettled([
+        locResp?.body?.cancel(),
+        imgResp?.body?.cancel(),
+      ]);
+      return [];
+    };
+    if (
+      (locResp.status >= 300 && locResp.status < 400) ||
+      (imgResp.status >= 300 && imgResp.status < 400)
+    ) {
+      console.warn(
+        '[CCTV] Tarktee DATEX catalog redirected; redirects are not followed',
+      );
+      return discardBoth();
+    }
     if (!locResp.ok) {
       console.warn('[CCTV] Tarktee locations download failed:', locResp.status);
-      return [];
+      return discardBoth();
     }
     if (!imgResp.ok) {
       console.warn('[CCTV] Tarktee images download failed:', imgResp.status);
-      return [];
+      return discardBoth();
     }
-    const [locXml, imgXml] = await Promise.all([
+    let locXml;
-      locResp.text(),
+    let imgXml;
-      imgResp.text(),
+    const readAbort = new AbortController();
+    try {
+      [locXml, imgXml] = await Promise.all([
+        readResponseTextCapped(
+          locResp,
+          TARKTEE_MAX_CATALOG_BYTES,
+          readAbort.signal,
+        ),
+        readResponseTextCapped(
+          imgResp,
+          TARKTEE_MAX_CATALOG_BYTES,
+          readAbort.signal,
+        ),
       ]);
+    } catch (error) {
+      readAbort.abort();
+      await discardBoth();
+      throw error;
+    }
     const locations = parseTarkteeDatexLocations(locXml);
     const images = parseTarkteeDatexImages(imgXml);
     if (!locations.size || !images.size) {
@@ -1225,6 +1263,7 @@ export async function loadTarkteeSourcesFromDatex() {
     }
 
     const cameras = [];
+    const seen = new Set();
     for (const [locationId, loc] of locations.entries()) {
       const imageUrl = images.get(locationId);
       if (!imageUrl) continue;
@@ -1236,6 +1275,8 @@ export async function loadTarkteeSourcesFromDatex() {
       const cameraId = numMatch
         ? `ee-tarktee-${numMatch[1]}`
         : `ee-tarktee-${locationId}`;
+      if (seen.has(cameraId)) continue;
+      seen.add(cameraId);
       cameras.push({
         id: cameraId,
         name: loc.name,

src/data/cctvEstonia.test.mjs

@@ -1,5 +1,10 @@
 import { test } from 'node:test';
 import assert from 'node:assert/strict';
+import {
+  TARKTEE_IMAGES_URL,
+  TARKTEE_LOCATIONS_URL,
+  TARKTEE_MAX_CATALOG_BYTES,
+} from '../../server/providers/cctv/constants.js';
 import {
   loadTallinnSourcesFromCatalog,
   loadTarkteeSourcesFromDatex,
@@ -8,6 +13,12 @@ import {
 } from '../../server/providers/cctv/sources.js';
 import { TARKTEE_FETCH_TIMEOUT_MS } from '../../server/providers/cctv/constants.js';
 
+/** Silence the loaders' progress and failure logging. */
+function quiet(t) {
+  t.mock.method(console, 'log', () => {});
+  t.mock.method(console, 'warn', () => {});
+}
+
 const SAMPLE_LOCATIONS_XML = `<?xml version="1.0" encoding="UTF-8"?>
 <d2LogicalModel xmlns="http://datex2.eu/schema/2/2_0">
   <payloadPublication>
@@ -110,3 +121,130 @@ test('Tarktee DATEX requests wait 30 s, past the slow answers measured live', as
   assert.equal(TARKTEE_FETCH_TIMEOUT_MS, 30_000);
   assert.equal(cameras.length, 1);
 });
+
+test('Tarktee loader deduplicates cameras sharing an image number before the cap', async (t) => {
+  quiet(t);
+  const dupLocationsXml = `<?xml version="1.0" encoding="UTF-8"?>
+<d2LogicalModel xmlns="http://datex2.eu/schema/2/2_0">
+  <payloadPublication>
+    <predefinedLocationContainer id="TRAFFIC_CAMERAS" version="0">
+      <predefinedLocation id="loc-1" version="0">
+        <predefinedLocationName><values><value lang="et">Tallinn Ringtee</value></values></predefinedLocationName>
+        <location><pointByCoordinates><pointCoordinates><latitude>59.4370</latitude><longitude>24.7530</longitude></pointCoordinates></pointByCoordinates></location>
+      </predefinedLocation>
+      <predefinedLocation id="loc-2" version="0">
+        <predefinedLocationName><values><value lang="et">Tallinn Ringtee duplicate</value></values></predefinedLocationName>
+        <location><pointByCoordinates><pointCoordinates><latitude>59.4371</latitude><longitude>24.7531</longitude></pointCoordinates></pointByCoordinates></location>
+      </predefinedLocation>
+    </predefinedLocationContainer>
+  </payloadPublication>
+</d2LogicalModel>`;
+  const dupImagesXml = `<?xml version="1.0" encoding="UTF-8"?>
+<d2LogicalModel xmlns="http://datex2.eu/schema/2/2_0">
+  <payloadPublication>
+    <trafficView id="1"><linearTrafficView id="1"><linearPredefinedLocationReference id="loc-1"/><trafficViewRecord id="1-1"><urlLink><urlLinkAddress>https://tarktee.transpordiamet.ee/images/42/42_1.jpg</urlLinkAddress></urlLink></trafficViewRecord></linearTrafficView></trafficView>
+    <trafficView id="2"><linearTrafficView id="2"><linearPredefinedLocationReference id="loc-2"/><trafficViewRecord id="2-1"><urlLink><urlLinkAddress>https://tarktee.transpordiamet.ee/images/42/42_2.jpg</urlLinkAddress></urlLink></trafficViewRecord></linearTrafficView></trafficView>
+  </payloadPublication>
+</d2LogicalModel>`;
+
+  const seen = [];
+  t.mock.method(globalThis, 'fetch', async (url, init) => {
+    seen.push([String(url), init?.redirect]);
+    const body =
+      String(url) === TARKTEE_LOCATIONS_URL ? dupLocationsXml : dupImagesXml;
+    return new Response(body, {
+      status: 200,
+      headers: { 'Content-Type': 'application/xml' },
+    });
+  });
+
+  const cameras = await loadTarkteeSourcesFromDatex();
+  assert.deepEqual(seen, [
+    [TARKTEE_LOCATIONS_URL, 'manual'],
+    [TARKTEE_IMAGES_URL, 'manual'],
+  ]);
+  assert.equal(cameras.length, 1);
+  assert.equal(cameras[0].id, 'ee-tarktee-42');
+  assert.equal(cameras[0].name, 'Tallinn Ringtee');
+});
+
+test('Tarktee loader refuses a redirect and cancels both response bodies', async (t) => {
+  quiet(t);
+  const cancelled = { locations: false, images: false };
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    const isLocations = String(url) === TARKTEE_LOCATIONS_URL;
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode('<xml/>'));
+      },
+      cancel() {
+        if (isLocations) cancelled.locations = true;
+        else cancelled.images = true;
+      },
+    });
+    return new Response(body, {
+      status: isLocations ? 302 : 200,
+      headers: isLocations ? { location: 'https://evil.example/loc.xml' } : {},
+    });
+  });
+
+  assert.deepEqual(await loadTarkteeSourcesFromDatex(), []);
+  assert.equal(cancelled.locations, true, 'redirected body is cancelled');
+  assert.equal(cancelled.images, true, 'sibling 200 body is also cancelled');
+});
+
+test('Tarktee loader cancels the sibling body when one feed fails or overflows the byte cap', async (t) => {
+  quiet(t);
+  let siblingCancelled = false;
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    const isLocations = String(url) === TARKTEE_LOCATIONS_URL;
+    if (isLocations) {
+      return new Response('', { status: 503 });
+    }
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(SAMPLE_IMAGES_XML));
+      },
+      cancel() {
+        siblingCancelled = true;
+      },
+    });
+    return new Response(body, { status: 200 });
+  });
+
+  assert.deepEqual(await loadTarkteeSourcesFromDatex(), []);
+  assert.equal(
+    siblingCancelled,
+    true,
+    'sibling 200 stream is cancelled on 503',
+  );
+
+  let overflowSiblingCancelled = false;
+  t.mock.method(globalThis, 'fetch', async (url) => {
+    const isLocations = String(url) === TARKTEE_LOCATIONS_URL;
+    if (isLocations) {
+      const padded =
+        SAMPLE_LOCATIONS_XML + ' '.repeat(TARKTEE_MAX_CATALOG_BYTES);
+      return new Response(padded, {
+        status: 200,
+        headers: { 'Content-Length': String(padded.length) },
+      });
+    }
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode(SAMPLE_IMAGES_XML));
+      },
+      cancel() {
+        overflowSiblingCancelled = true;
+      },
+    });
+    return new Response(body, { status: 200 });
+  });
+
+  assert.deepEqual(await loadTarkteeSourcesFromDatex(), []);
+  assert.equal(
+    overflowSiblingCancelled,
+    true,
+    'sibling stream is cancelled when the other feed exceeds the byte cap',
+  );
+});