From f4123b32e41bc02fdd7610f8f8e14d0df47ba17b Mon Sep 17 00:00:00 2001 From: Codex Date: Tue, 13 Oct 2026 21:51:03 +0200 Subject: [PATCH] Checkpoint guarded source request broker --- scraper/adapters/index.js | 6 +- scraper/jobs/weebcentral-archive.js | 6 +- scraper/lib/http.js | 30 +++++++--- scraper/plugins/broker.js | 85 ++++++++++++++++++++++++++++ scraper/test/source-broker.test.js | 43 ++++++++++++++ vision/first-party-source-plugins.md | 5 +- 6 files changed, 162 insertions(+), 13 deletions(-) create mode 100644 scraper/plugins/broker.js create mode 100644 scraper/test/source-broker.test.js diff --git a/scraper/adapters/index.js b/scraper/adapters/index.js index cd22afb..9b88ebb 100644 --- a/scraper/adapters/index.js +++ b/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) { diff --git a/scraper/jobs/weebcentral-archive.js b/scraper/jobs/weebcentral-archive.js index 76b0455..f15d7d9 100644 --- a/scraper/jobs/weebcentral-archive.js +++ b/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}`); diff --git a/scraper/lib/http.js b/scraper/lib/http.js index bb188bf..0279170 100644 --- a/scraper/lib/http.js +++ b/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, { - redirect: 'follow', + response = await this.transport(currentUrl, { + 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 }); - throw Object.assign(new StopScrapeError(`Host returned ${response.status}; stop and require human review: ${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: ${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; } } diff --git a/scraper/plugins/broker.js b/scraper/plugins/broker.js new file mode 100644 index 0000000..df9979e --- /dev/null +++ b/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 }; diff --git a/scraper/test/source-broker.test.js b/scraper/test/source-broker.test.js new file mode 100644 index 0000000..32bdb16 --- /dev/null +++ b/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); +}); diff --git a/vision/first-party-source-plugins.md b/vision/first-party-source-plugins.md index 8ddc3da..79dabfc 100644 --- a/vision/first-party-source-plugins.md +++ b/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 -redirect/DNS enforcement and source-specific CDN allowlists are still pending. +Parser-result URL validation and bounded worker messages were added next. The +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