Checkpoint guarded source request broker

scraper/adapters/index.js

@@ -3,6 +3,7 @@
 const { runArchive } = require('../jobs/weebcentral-archive');
 const { runMangaDexArchive } = require('../jobs/mangadex-archive');
 const { SlowHttp } = require('../lib/http');
+const { createWeebHttp } = require('../plugins/broker');
 const { discoverWeebCatalog } = require('./weebcentral');
 const { mangaDexInfo, discoverMangaDexCatalog } = require('./mangadex');
 const { ArchiveControlError } = require('../lib/control');
@@ -29,10 +30,11 @@ const adapters = [{
   label: 'WeebCentral',
   match: weebCentralInfo,
   async discover(item, options) {
-    const http = new SlowHttp({
+    const httpOptions = {
       minMs: options.settings.minMs, maxMs: options.settings.maxMs, maxRetries: 2,
       onEvent: options.onEvent,
-    });
+    };
+    const http = process.env.SOURCE_PLUGIN_URL ? createWeebHttp(httpOptions) : new SlowHttp(httpOptions);
     return discoverWeebCatalog(item.url, controlledHttp(http, options.shouldStop));
   },
   run(item, options) {

scraper/jobs/weebcentral-archive.js

@@ -3,6 +3,7 @@
 const fs = require('fs').promises;
 const path = require('path');
 const { SlowHttp } = require('../lib/http');
+const { createWeebHttp } = require('../plugins/broker');
 const { sleep } = require('../lib/delay');
 const { ArchiveControlError } = require('../lib/control');
 const { cleanPart } = require('../lib/names');
@@ -68,10 +69,11 @@ async function runArchive(options) {
     state.updated = entry.time;
     await writeJsonAtomic(stateFile, state);
   };
-  const http = new SlowHttp({
+  const httpOptions = {
     minMs: options.minMs, maxMs: options.maxMs, maxRetries: 2,
     onEvent: item => event(item.level, item.message, { url: item.url, ...(item.status ? { status: item.status } : {}) }),
-  });
+  };
+  const http = process.env.SOURCE_PLUGIN_URL ? createWeebHttp(httpOptions) : new SlowHttp(httpOptions);
 
   checkControl();
   console.log(`[archive] discovering catalog ${options.seriesUrl}`);

scraper/lib/http.js

@@ -11,39 +11,55 @@ class SlowHttp {
     this.maxRetries = options.maxRetries ?? 2;
     this.onEvent = options.onEvent;
     this.timeoutMs = options.timeoutMs || 60000;
+    this.transport = options.transport || fetch;
+    this.validateUrl = options.validateUrl || null;
   }
 
   async get(url, options = {}) {
+    let redirects = 0;
+    let currentUrl = url;
     for (let attempt = 0; ; attempt++) {
+      if (this.validateUrl) this.validateUrl(currentUrl, options);
       await this.gate.wait();
       let response;
       try {
-        response = await fetch(url, {
+        response = await this.transport(currentUrl, {
-          redirect: 'follow',
+          redirect: this.validateUrl ? 'manual' : 'follow',
           signal: AbortSignal.timeout(this.timeoutMs),
           headers: { 'user-agent': this.userAgent, accept: options.accept || '*/*', ...(options.headers || {}) },
         });
         this.gate.mark();
       } catch (error) {
         this.gate.mark();
-        if (attempt >= this.maxRetries) throw new Error(`Network request failed after ${attempt + 1} attempts: ${url}`, { cause: error });
+        if (error.nonRetryable) throw error;
+        if (attempt >= this.maxRetries) throw new Error(`Network request failed after ${attempt + 1} attempts: ${currentUrl}`, { cause: error });
         const backoff = 30000 * (2 ** attempt);
         const message = `Network failure; retrying in ${backoff / 1000}s (${attempt + 1}/${this.maxRetries})`;
         console.warn(`[http] ${message}`);
-        if (this.onEvent) await this.onEvent({ level: 'warning', message, url });
+        if (this.onEvent) await this.onEvent({ level: 'warning', message, url: currentUrl });
         await sleep(backoff);
         continue;
       }
 
+      if (this.validateUrl && response.status >= 300 && response.status < 400) {
+        const location = response.headers.get('location');
+        if (!location || ++redirects > 5) throw new Error('Source redirect limit or missing location');
+        currentUrl = new URL(location, currentUrl).href;
+        this.validateUrl(currentUrl, options);
+        await response.body?.cancel?.();
+        attempt = -1;
+        continue;
+      }
+
       if ([401, 403, 429].includes(response.status)) {
-        if (this.onEvent) await this.onEvent({ level: 'error', message: `Host returned ${response.status}; stopping`, url, status: response.status });
+        if (this.onEvent) await this.onEvent({ level: 'error', message: `Host returned ${response.status}; stopping`, url: currentUrl, status: response.status });
-        throw Object.assign(new StopScrapeError(`Host returned ${response.status}; stop and require human review: ${url}`), { status: response.status });
+        throw Object.assign(new StopScrapeError(`Host returned ${response.status}; stop and require human review: ${currentUrl}`), { status: response.status });
       }
       if (response.status >= 500 && attempt < this.maxRetries) {
         await sleep(30000 * (2 ** attempt));
         continue;
       }
-      if (!response.ok) throw Object.assign(new Error(`HTTP ${response.status}: ${url}`), { status: response.status });
+      if (!response.ok) throw Object.assign(new Error(`HTTP ${response.status}: ${currentUrl}`), { status: response.status });
       return response;
     }
   }

scraper/plugins/broker.js

@@ -0,0 +1,85 @@
+'use strict';
+
+const dns = require('node:dns').promises;
+const https = require('node:https');
+const { BlockList, isIP } = require('node:net');
+const { SlowHttp } = require('../lib/http');
+
+const blocked = new BlockList();
+for (const [address, bits] of [
+  ['0.0.0.0', 8], ['10.0.0.0', 8], ['100.64.0.0', 10], ['127.0.0.0', 8],
+  ['169.254.0.0', 16], ['172.16.0.0', 12], ['192.0.0.0', 24],
+  ['192.0.2.0', 24], ['192.168.0.0', 16], ['198.18.0.0', 15],
+  ['198.51.100.0', 24], ['203.0.113.0', 24], ['224.0.0.0', 4],
+  ['240.0.0.0', 4], ['::', 128], ['::1', 128], ['fc00::', 7],
+  ['fe80::', 10], ['ff00::', 8], ['2001:db8::', 32],
+]) blocked.addSubnet(address, bits, isIP(address) === 6 ? 'ipv6' : 'ipv4');
+
+function publicAddress(address) {
+  const family = isIP(address);
+  return !!family && !address.toLowerCase().startsWith('::ffff:')
+    && !blocked.check(address, family === 4 ? 'ipv4' : 'ipv6');
+}
+
+function validateWeebRequest(value, options = {}) {
+  const url = new URL(value);
+  const host = url.hostname.toLowerCase();
+  if (url.protocol !== 'https:' || url.username || url.password || url.port
+    || isIP(host) || host === 'localhost' || host.endsWith('.localhost')
+    || host.endsWith('.local') || !host.includes('.')) {
+    throw new Error('Source request URL is not public HTTPS');
+  }
+  if (options.accept === 'text/html' && host !== 'weebcentral.com') {
+    throw new Error('WeebCentral catalog request left its source host');
+  }
+  return url;
+}
+
+async function pinnedFetch(value, options = {}) {
+  const url = new URL(value);
+  if (url.protocol !== 'https:') throw new Error('Pinned transport requires HTTPS');
+  const answers = await dns.lookup(url.hostname, { all: true });
+  if (!answers.length || answers.some(answer => !publicAddress(answer.address))) {
+    throw Object.assign(new Error('Source hostname resolved to a non-public address'), { nonRetryable: true });
+  }
+  const selected = answers[0];
+  const maxBytes = options.headers?.accept === 'text/html' ? 2 * 1024 * 1024 : 50 * 1024 * 1024;
+  return new Promise((resolve, reject) => {
+    const request = https.request(url, {
+      method: 'GET', headers: options.headers, signal: options.signal,
+      agent: false,
+      lookup: (_hostname, _lookupOptions, callback) => callback(null, selected.address, selected.family),
+    }, response => {
+      let bytes = 0;
+      const chunks = [];
+      response.on('data', chunk => {
+        bytes += chunk.length;
+        if (bytes > maxBytes) {
+          response.destroy(Object.assign(new Error('Source response exceeds size limit'), { nonRetryable: true }));
+          return;
+        }
+        chunks.push(chunk);
+      });
+      response.on('error', reject);
+      response.on('end', () => {
+        const body = Buffer.concat(chunks);
+        const status = response.statusCode || 0;
+        resolve({
+          status, ok: status >= 200 && status < 300,
+          headers: { get: name => response.headers[String(name).toLowerCase()] || null },
+          text: async () => body.toString('utf8'),
+          json: async () => JSON.parse(body.toString('utf8')),
+          arrayBuffer: async () => body,
+        });
+      });
+    });
+    request.on('error', reject);
+    request.end();
+  });
+}
+
+function createWeebHttp(options = {}) {
+  return new SlowHttp({ ...options, transport: pinnedFetch, validateUrl: validateWeebRequest });
+}
+
+module.exports = { publicAddress, validateWeebRequest, pinnedFetch, createWeebHttp };

scraper/test/source-broker.test.js

@@ -0,0 +1,43 @@
+'use strict';
+
+const test = require('node:test');
+const assert = require('node:assert/strict');
+const { SlowHttp, StopScrapeError } = require('../lib/http');
+const { publicAddress, validateWeebRequest } = require('../plugins/broker');
+
+test('plugin broker rejects local, private, non-TLS, and cross-source URLs', () => {
+  for (const address of ['127.0.0.1', '10.0.0.5', '192.168.1.1', '169.254.169.254', '::1', '::ffff:127.0.0.1']) {
+    assert.equal(publicAddress(address), false, address);
+  }
+  assert.equal(publicAddress('1.1.1.1'), true);
+  assert.equal(publicAddress('2606:4700:4700::1111'), true);
+  assert.throws(() => validateWeebRequest('http://weebcentral.com/series/X', { accept: 'text/html' }), /public HTTPS/);
+  assert.throws(() => validateWeebRequest('https://localhost/secret'), /public HTTPS/);
+  assert.throws(() => validateWeebRequest('https://other.example/series/X', { accept: 'text/html' }), /source host/);
+  assert.doesNotThrow(() => validateWeebRequest('https://img.example/page.jpg'));
+});
+
+test('redirect hops are validated and paced before another request', async () => {
+  const calls = [];
+  const http = new SlowHttp({
+    minMs: 0, maxMs: 0, maxRetries: 0, validateUrl: validateWeebRequest,
+    transport: async url => {
+      calls.push(url);
+      return calls.length === 1
+        ? { status: 302, headers: { get: () => 'https://localhost/private' } }
+        : { status: 200, ok: true };
+    },
+  });
+  await assert.rejects(http.get('https://weebcentral.com/series/X', { accept: 'text/html' }), /public HTTPS/);
+  assert.deepEqual(calls, ['https://weebcentral.com/series/X']);
+});
+
+test('a guarded 403 stops immediately without redirect or retry', async () => {
+  let calls = 0;
+  const http = new SlowHttp({
+    minMs: 0, maxMs: 0, maxRetries: 2, validateUrl: validateWeebRequest,
+    transport: async () => { calls++; return { status: 403 }; },
+  });
+  await assert.rejects(http.get('https://weebcentral.com/series/X', { accept: 'text/html' }), StopScrapeError);
+  assert.equal(calls, 1);
+});

vision/first-party-source-plugins.md

@@ -6,8 +6,9 @@ Compose worker. No external plugin was installed, no host was probed, and no
 live queue, delay, or container was changed. The broker, generic runner, new
 hoster pilot, admin controls, and full deployment validation below remain work
 to do; the diagram describes the target rather than the current implementation.
-Parser-result URL validation and bounded worker messages were added next; full
+Parser-result URL validation and bounded worker messages were added next. The
-redirect/DNS enforcement and source-specific CDN allowlists are still pending.
+opt-in WeebCentral path now validates redirect hops and pins public DNS answers;
+source-specific image CDN allowlists and generic broker integration remain pending.
 
 ## Decision