165 lines
10 KiB
JavaScript
165 lines
10 KiB
JavaScript
'use strict';
|
|
|
|
const https = require('node:https');
|
|
const fs = require('node:fs');
|
|
const fsp = fs.promises;
|
|
const crypto = require('node:crypto');
|
|
const { normalizeWorkIdentity } = require('../scraper/lib/work-identity');
|
|
const { pipeline } = require('node:stream/promises');
|
|
const { Transform } = require('node:stream');
|
|
const { PeeringState, fp, requirePeer, grantStatus } = require('./state');
|
|
|
|
const MAX_JSON = 8 * 1024 * 1024;
|
|
const MAX_PAGE = 30 * 1024 * 1024;
|
|
const agents = new Map();
|
|
const downloadSlots = new Map();
|
|
function remoteError(code, message, retryAfter) {
|
|
const error = new Error(`Peer HTTP ${code}: ${String(message).slice(0, 300)}`);
|
|
error.status = code;
|
|
error.retryAfter = retryAfter;
|
|
return error;
|
|
}
|
|
function connection(store, id, needGrant = true, requireReceive = false) {
|
|
const state = store.load(), peer = requirePeer(state, id), identity = store.identity();
|
|
if (!identity) throw new Error('Generate local peering identity first');
|
|
if (needGrant && !peer.paired) throw new Error('Verify peer identity first');
|
|
if (requireReceive && !peer.receive) throw new Error('Downloads from this peer are disabled');
|
|
const token = store.outbound()[id];
|
|
if (needGrant && !token) throw new Error('Import the peer grant first');
|
|
const cacheKey = `${id}:${fp(peer.cert)}:${identity.fingerprint}`;
|
|
let agent = agents.get(cacheKey);
|
|
if (!agent) {
|
|
agent = new https.Agent({ keepAlive: true, maxSockets: Math.max(1, peer.download?.concurrent || 2), minVersion: 'TLSv1.3',
|
|
cert: identity.cert, key: identity.key, ca: peer.cert, rejectUnauthorized: true,
|
|
checkServerIdentity(_host, cert) {
|
|
if (!cert.raw || fp(cert.raw) !== peer.fingerprint) return new Error('Peer certificate fingerprint changed');
|
|
return undefined; // The out-of-band pin replaces a hostname CA hierarchy.
|
|
} });
|
|
agents.set(cacheKey, agent);
|
|
}
|
|
return { peer, agent, token };
|
|
}
|
|
function request(store, id, endpoint, options = {}) {
|
|
const { peer, agent, token } = connection(store, id, !options.pairing && !options.access, !!options.requireReceive);
|
|
const base = new URL(peer.url), target = new URL(endpoint, base);
|
|
if (target.origin !== base.origin || !target.pathname.startsWith('/peer/v1/')) throw new Error('Invalid peer endpoint');
|
|
return new Promise((resolve, reject) => {
|
|
const headers = { accept: options.binary ? 'image/*' : 'application/json',
|
|
...(options.pairing ? { 'x-invitation-id': peer.invitationId || '', 'x-invitation-code': peer.invitationCode || '' }
|
|
: options.access ? {} : { authorization: `Bearer ${token}` }),
|
|
...options.headers };
|
|
const req = https.request(target, { method: 'GET', agent, headers, timeout: 30000 }, res => resolve(res));
|
|
req.on('timeout', () => req.destroy(new Error('Peer request timed out')));
|
|
req.on('error', reject); req.end();
|
|
});
|
|
}
|
|
async function readResponse(res, limit) {
|
|
const chunks = []; let size = 0;
|
|
for await (const chunk of res) { size += chunk.length; if (size > limit) throw new Error('Peer response exceeds limit'); chunks.push(chunk); }
|
|
return Buffer.concat(chunks);
|
|
}
|
|
async function jsonRequest(store, id, endpoint, options = {}) {
|
|
const res = await request(store, id, endpoint, options);
|
|
const data = await readResponse(res, MAX_JSON);
|
|
let value;
|
|
try { value = JSON.parse(data.toString('utf8')); } catch { throw new Error('Peer returned invalid JSON'); }
|
|
if (res.statusCode !== 200) throw remoteError(res.statusCode, value.error || 'Request failed', res.headers['retry-after']);
|
|
return value;
|
|
}
|
|
async function probe(store, id) {
|
|
const peer = requirePeer(store.load(), id);
|
|
if (Date.parse(peer.invitationExpiresAt) <= Date.now() && !peer.paired) throw new Error('Peer invitation expired');
|
|
const hello = await jsonRequest(store, id, '/peer/v1/hello', { pairing: !peer.paired });
|
|
if (hello.version !== 1 || hello.fingerprint !== peer.fingerprint) throw new Error('Peer hello identity mismatch');
|
|
const state = store.load(), paired = requirePeer(state, id); paired.paired = true;
|
|
delete paired.invitationId; delete paired.invitationCode; delete paired.invitationExpiresAt;
|
|
const file = store.stateFile(), temp = `${file}.${crypto.randomBytes(6).toString('hex')}.tmp`;
|
|
fs.writeFileSync(temp, JSON.stringify(state, null, 2) + '\n', { mode: 0o600 }); fs.renameSync(temp, file);
|
|
return { ok: true, id: hello.id };
|
|
}
|
|
async function access(store, id) {
|
|
const result = await jsonRequest(store, id, '/peer/v1/access', { access: true });
|
|
if (result.version !== 1 || typeof result.paired !== 'boolean' || !['none', 'active', 'paused', 'expired'].includes(result.grant?.status)
|
|
|| !Array.isArray(result.grant?.scopes)) throw new Error('Invalid peer access status');
|
|
return result;
|
|
}
|
|
async function cover(store, id, shareId) {
|
|
if (!/^m[0-9a-f]{24}$/.test(shareId)) throw new Error('Invalid peer manga ID');
|
|
const res = await request(store, id, `/peer/v1/covers/${shareId}`, { binary: true });
|
|
if (res.statusCode !== 200) {
|
|
const body = await readResponse(res, 4096).catch(() => Buffer.alloc(0));
|
|
let message; try { message = JSON.parse(body.toString()).error; } catch { message = 'Cover unavailable'; }
|
|
throw remoteError(res.statusCode, message, res.headers['retry-after']);
|
|
}
|
|
const mime = String(res.headers['content-type'] || '').split(';')[0];
|
|
if (!/^image\/(?:jpeg|png|webp|gif|avif|bmp)$/.test(mime)) throw new Error('Peer returned an unsupported cover type');
|
|
const buffer = await readResponse(res, 5 * 1024 * 1024);
|
|
if (!buffer.length) throw new Error('Peer returned an empty cover');
|
|
return { buffer, mime, etag: res.headers.etag || null };
|
|
}
|
|
async function catalog(store, id) {
|
|
const items = []; let cursor = null;
|
|
for (let page = 0; page < 100; page++) {
|
|
const result = await jsonRequest(store, id, `/peer/v1/catalog${cursor ? `?cursor=${encodeURIComponent(cursor)}` : ''}`);
|
|
if (!Array.isArray(result.items) || result.items.length > 50) throw new Error('Invalid peer catalog');
|
|
for (const item of result.items) if (item.workIdentity != null && !normalizeWorkIdentity(item.workIdentity)) throw new Error('Invalid work identity in peer catalog');
|
|
items.push(...result.items); cursor = result.nextCursor;
|
|
if (!cursor) return items;
|
|
}
|
|
throw new Error('Peer catalog exceeds page limit');
|
|
}
|
|
async function manga(store, id, shareId) {
|
|
if (!/^m[0-9a-f]{24}$/.test(shareId)) throw new Error('Invalid manga ID');
|
|
const chapters = []; let cursor = null, first;
|
|
for (let page = 0; page < 100; page++) {
|
|
const result = await jsonRequest(store, id, `/peer/v1/manga/${shareId}${cursor ? `?cursor=${encodeURIComponent(cursor)}` : ''}`);
|
|
if (result.id !== shareId || !Array.isArray(result.chapters) || result.chapters.length > 100) throw new Error('Invalid peer manga manifest');
|
|
if (result.workIdentity != null && !normalizeWorkIdentity(result.workIdentity)) throw new Error('Invalid work identity in peer manga manifest');
|
|
first ||= result; chapters.push(...result.chapters); cursor = result.nextCursor;
|
|
if (!cursor) return { id: shareId, title: first.title, ...first, chapters };
|
|
}
|
|
throw new Error('Peer manga exceeds chapter limit');
|
|
}
|
|
async function chapter(store, id, shareId, chapterId) {
|
|
if (!/^m[0-9a-f]{24}$/.test(shareId) || !/^c[0-9a-f]{32}$/.test(chapterId)) throw new Error('Invalid chapter ID');
|
|
const result = await jsonRequest(store, id, `/peer/v1/manga/${shareId}/chapters/${chapterId}`);
|
|
if (result.id !== chapterId || !Array.isArray(result.pages) || !result.pages.length || result.pages.length > 500) throw new Error('Invalid peer chapter manifest');
|
|
for (const page of result.pages) if (!/^b[0-9a-f]{32}$/.test(page.id) || !Number.isSafeInteger(page.bytes) || page.bytes < 1 || page.bytes > MAX_PAGE
|
|
|| !/^[0-9a-f]{64}$/.test(page.sha256) || !/^image\/(?:jpeg|png|webp|gif|avif|bmp)$/.test(page.mime)) throw new Error('Invalid peer page');
|
|
return result;
|
|
}
|
|
async function downloadPage(store, id, page, file, shouldStop) {
|
|
const verify = async () => {
|
|
const digest = crypto.createHash('sha256');
|
|
for await (const chunk of fs.createReadStream(file)) digest.update(chunk);
|
|
return digest.digest('hex') === page.sha256;
|
|
};
|
|
let stat = await fsp.stat(file).catch(() => null);
|
|
if (stat?.size === page.bytes && await verify()) return { reused: true };
|
|
if (stat?.size === page.bytes) { await fsp.rm(file, { force: true }); stat = null; }
|
|
if (stat && stat.size > page.bytes) { await fsp.rm(file, { force: true }); stat = null; }
|
|
const start = stat?.size || 0;
|
|
if (shouldStop?.()) throw new Error('Peer transfer paused');
|
|
const res = await request(store, id, `/peer/v1/blobs/${page.id}`, { binary: true, requireReceive: true,
|
|
headers: start ? { range: `bytes=${start}-`, ...(page.etag ? { 'if-match': page.etag } : {}) } : {} });
|
|
if (res.statusCode !== (start ? 206 : 200)) {
|
|
const body = await readResponse(res, 4096).catch(() => Buffer.alloc(0));
|
|
let message; try { message = JSON.parse(body.toString()).error; } catch { message = 'Transfer failed'; }
|
|
throw remoteError(res.statusCode, message, res.headers['retry-after']);
|
|
}
|
|
if (start && !String(res.headers['content-range'] || '').startsWith(`bytes ${start}-`)) throw new Error('Peer returned an invalid range');
|
|
if (Number(res.headers['content-length']) !== page.bytes - start) throw new Error('Peer returned an unexpected page size');
|
|
const rate = requirePeer(store.load(), id).download?.bytesPerSecond || 2 * 1024 * 1024;
|
|
const throttle = new Transform({ transform(chunk, _encoding, callback) {
|
|
const now = Date.now(), ready = Math.max(now, downloadSlots.get(id) || now) + chunk.length * 1000 / rate;
|
|
downloadSlots.set(id, ready);
|
|
setTimeout(() => callback(null, chunk), Math.max(0, ready - now));
|
|
} });
|
|
await pipeline(res, throttle, fs.createWriteStream(file, { flags: start ? 'a' : 'w', mode: 0o600 }));
|
|
stat = await fsp.stat(file);
|
|
if (stat.size !== page.bytes || !await verify()) { await fsp.rm(file, { force: true }); throw new Error('Peer page checksum mismatch'); }
|
|
return { reused: false };
|
|
}
|
|
|
|
module.exports = { connection, jsonRequest, probe, access, catalog, manga, cover, chapter, downloadPage };
|