diff --git a/docs/peering.md b/docs/peering.md new file mode 100644 index 0000000..ab0b8da --- /dev/null +++ b/docs/peering.md @@ -0,0 +1,138 @@ +# Server peering (v1) + +Status: implemented, tested in isolated local fixtures, and enabled for a local +two-instance pilot on 2026-10-10. It has **not** been deployed to the VM. See the broader design and +future ideas in [`vision/server-peering.md`](../vision/server-peering.md). + +Peering is an opt-in, pull-only transfer between MangaCircus servers. It does not +exchange accounts, reading history, ratings, notes, or scraper credentials. The +publishing gateway reads a library mount, and only the existing scraper publishes +received chapters. No database server or second library writer is introduced. + +## Roles and direction + +The admin UI is **Settings → Peering**, and all users can open the **Peers** catalog +page. Each server may have many peers. For every peer, its admin independently sets: + +- **Serve approved manga**: expose only individually checked library titles to that + peer. Off is the default and invalidates the current serving grant. +- **Allow downloads**: queue and pull chapters from that peer. Off is the default; + a verified pairing and imported grant still allow browsing its approved catalog. +- **Let readers queue**: ordinary readers may request transfers; otherwise only an + admin may queue. Browsing a permitted catalog does not disclose the local user to + the remote server. +- Upload streams, bytes/s and bytes/day, and receiving pages in flight and bytes/s. + Defaults are two in-flight pages and 2 MiB/s in each direction per peer. The + publisher's caps remain authoritative. A gateway-wide four-stream ceiling applies. + +Serving and receiving are separate directions. Pairing by itself exposes no title. + +## Deployment + +The optional `peer-gateway` Compose profile publishes port 8093 on **127.0.0.1** by +default. It uses the same image as reader/scraper, a read-only library mount, a +read-only peering policy/identity mount, and its own writable usage-counter mount. +The gateway needs an external-facing Docker network for its host port to work on +Docker Desktop; it also joins the private backend. It has no reader account database, +scraper queue, Docker socket, or writable library mount. It must never be scaled +past one instance for the same usage-counter directory. + +1. Build the image and start the normal stack as documented in + [`compose-deployment.md`](compose-deployment.md). Generate the local certificate + in **Settings → Peering**. Keep `/data/peering/identity.key` private and backed up. +2. Start the optional service with `docker compose --profile peering up -d + peer-gateway`. Verify `docker compose --profile peering ps` shows healthy and + `https://127.0.0.1:8093/health` responds. `/health` is deliberately unauthenticated + and contains only `{ "ok": true }`. +3. Give the other server a reachable HTTPS endpoint for port 8093. A private + Tailscale TCP forward is recommended for the pilot; the default loopback binding + is not remotely reachable. Do not expose port 8093 on the public internet merely + to make pairing work. Use a stable tailnet DNS name, not a changing IP. Keep + scraper (8091) and Sentinel (8092) private. + +The `PEER_BIND`, `PEER_PORT`, `PEER_DATA_PATH`, and `PEER_USAGE_PATH` Compose variables +control the bind and persistent paths. The gateway's internal port remains 8093. +OpenSSL is bundled in the app image for one-time P-256 self-signed certificate +generation. Each certificate is valid for one year. Expired/replaced identities +require administrator-supervised re-pairing; there is no automatic key rotation yet. + +## Pairing and transfer + +1. Each admin generates an identity, creates a 48-hour invitation, and exchanges the + JSON outside the app. The JSON contains a public certificate, SHA-256 fingerprint, + URL, invitation ID and one-use code—**not** the private key. Compare the + fingerprints through a separate trusted channel. +2. Both admins import the other's invitation and press **Test & verify pairing**. + The HTTPS handshake presents client and server certificates, checks their exact + pins and dates, and consumes the one-use invitation code. Both sides must probe. +3. A publisher checks specific titles under that peer, enables **Serve**, issues a + 90-day random grant, and sends the token privately to the receiving admin. The + receiving admin imports the grant and enables **Allow downloads**. Reverse + direction needs its own share selection and grant. The server stores only a hash + of a grant it issued; the receiver stores its imported token server-side. +4. A local user browses the catalog, selects missing/newest/last-five or an exact + chapter, and queues it through the normal scraper. The peer adapter downloads up + to the configured pages in flight, validates size and SHA-256, resumes a partial + page with HTTP Range, checkpoints between batches, and atomically publishes each + complete chapter into the ordinary library. Existing-chapter policy is applied + before download. Partial transfers remain in the hidden scraper work folder. + +The publisher sees the receiving **server** certificate and transfer requests, not +the human reader. It can still observe requested titles, times and byte counts. + +## Peer v1 HTTP interface + +All content endpoints require TLS 1.3 client certificate pinning, a paired/serving +peer, and a currently valid certificate-bound bearer grant. There is no insecure +HTTP fallback and no path-based file browsing. + +| Request | Response | +| --- | --- | +| `GET /peer/v1/hello` with one-use invitation headers | Protocol version and server identity fingerprint; no catalog | +| `GET /peer/v1/catalog?cursor=N` | Up to 50 approved opaque title IDs, title/year/status, next cursor | +| `GET /peer/v1/manga/{id}?cursor=N` | Up to 100 complete chapter IDs and light metadata | +| `GET /peer/v1/manga/{id}/chapters/{id}` | Ordered page IDs, MIME, byte size, SHA-256 and ETag | +| `GET /peer/v1/blobs/{id}` | One image, optional `Range: bytes=N-` and `If-Match`; 206 on resume | + +Unknown/unshared IDs return 404. Revoked serving returns 403; bad/missing grants +return 401; caps return 429 and `Retry-After`. No source-site URL, local filesystem +path, or user record is sent in protocol responses. Individual complete chapter +manifests are discovered from final `chapter.json` folders; an unfinished folder is +not shared. Catalog and chapter lists are short-lived in-memory caches. Turning off +serving blocks subsequent requests immediately, though an already streaming image +may finish its current chunk/response. + +The initial implementation sends original image bytes; it does not transcode them. +See [`vision/server-peering-compression-test.md`](../vision/server-peering-compression-test.md) +for the measured reason. The transport accepts a small amount of controlled page +parallelism, not unbounded connections. + +## Tests and safe operation + +`npm run test:peer` creates disposable A/B identities, gateways and libraries and +tests mTLS pairing, no-cert/bad-grant rejection, private catalog, per-title sharing, +manifest hashes, a resumed range response, scraper publication, duplicate skip and +revocation. `npm test` includes it. An isolated Compose acceptance is: + +```powershell +$env:APP_IMAGE = 'mangacircus:local' # replace with the image just built +$env:QA_PORT = '8188' +$env:QA_PEER_PORT = '8193' +node tools/test-compose.js --peer-gateway --peer-ui +``` + +It makes disposable library/data, four containers, a certificate and a browser +session, then tears them down. It never mounts the live library/queue. + +For live rollout, wait for a scraper checkpoint, back up the library and data +(including the peering identity, policy, outbound grants, gateway usage and hidden +work folders), and deploy reader/scraper/gateway from one immutable image digest. +There must remain exactly one queue-owning scraper and one reader database writer. +Peering is optional: stop the `peer-gateway` profile to close incoming service +without affecting reading or existing local downloads. Do not delete the identity +to recover from a network error; that would break every peer pin. + +Current limits: no automatic certificate renewal, invitation cancellation UI, +peer cover-image endpoint, multi-peer performance pilot, VM deployment, or public +internet exposure review. The gateway has no automated hoster-interaction logic; +source scraping remains subject to the existing source-safety policy. diff --git a/peering/adapter.js b/peering/adapter.js new file mode 100644 index 0000000..99bcec7 --- /dev/null +++ b/peering/adapter.js @@ -0,0 +1,110 @@ +'use strict'; + +const fs = require('node:fs').promises; +const path = require('node:path'); +const crypto = require('node:crypto'); +const { PeeringState, requirePeer } = require('./state'); +const client = require('./client'); +const { ArchiveControlError } = require('../scraper/lib/control'); +const { selectByScope } = require('../scraper/lib/scope'); +const { scanLibrary, findManga, shouldDownload } = require('../scraper/lib/library-inventory'); +const { publishManifest } = require('../scraper/lib/publisher'); +const { writeJsonAtomic } = require('../scraper/lib/json-file'); +const { sleep } = require('../scraper/lib/delay'); + +const dataRoot = () => path.resolve(process.env.PEER_DATA || path.join(process.env.DATA || 'data', 'peering')); +const store = () => new PeeringState(dataRoot()); +const infoFor = value => { + try { + const u = new URL(String(value)); + if (u.protocol !== 'peer:' || u.username || u.password || u.search || u.hash || !/^p[0-9a-f]{24}$/.test(u.hostname) + || !/^\/m[0-9a-f]{24}$/.test(u.pathname)) return null; + const peer = requirePeer(store().load(), u.hostname); + return { source: `peer-${peer.id}`, id: `${peer.id}-${u.pathname.slice(1)}`, peerId: peer.id, + shareId: u.pathname.slice(1), title: peer.alias, url: `peer://${peer.id}${u.pathname}` }; + } catch { return null; } +}; +function refsOf(result) { + return result.chapters.map(ref => ({ id: ref.id, number: ref.number, volume: ref.volume, + book: { number: Number(ref.volume) || 1 }, title: ref.title, pages: ref.pageCount })); +} +async function discover(info) { + const result = await client.manga(store(), info.peerId, info.shareId); + return { title: result.title, chapters: refsOf(result), availability: { selectedChapters: result.chapters.length, coverage: 'shared-selection' } }; +} +async function run(info, item, options) { + const stateStore = store(), peer = requirePeer(stateStore.load(), info.peerId); + if (!peer.paired || !peer.receive) throw new Error('Downloads from this peer are disabled'); + const stop = () => { const reason = options.shouldStop?.(); if (reason) throw new ArchiveControlError(reason === true ? 'paused' : String(reason)); }; + stop(); + const remote = await client.manga(stateStore, info.peerId, info.shareId); + const catalog = { title: remote.title, chapters: refsOf(remote), availability: { selectedChapters: remote.chapters.length, coverage: 'shared-selection' } }; + if (options.onCatalog) await options.onCatalog(catalog); + const sourceUrl = info.url, source = info.source; + const workRoot = path.join(options.libraryRoot, `.mcscraper-work-${info.id}`); + const stateFile = path.join(workRoot, 'job.json'); + await fs.mkdir(workRoot, { recursive: true }); + let job = { version: 1, publicationMode: 'incremental', source, seriesUrl: sourceUrl, completed: [], jobCompleted: 0, + status: 'running', started: new Date().toISOString(), updated: new Date().toISOString(), total: 0, current: null }; + try { const prior = JSON.parse(await fs.readFile(stateFile, 'utf8')); if (prior.seriesUrl === sourceUrl) job = { ...prior, status: 'running' }; } + catch (e) { if (e.code !== 'ENOENT') throw e; } + const record = async () => { job.updated = new Date().toISOString(); await writeJsonAtomic(stateFile, job); }; + const inventory = await scanLibrary(options.libraryRoot); + let manga = findManga(inventory, { source, seriesId: info.id, sourceUrl, libraryMangaId: item.libraryMangaId }); + if (item.libraryMangaId && !manga) throw new Error(`Library manga target not found: ${item.libraryMangaId}`); + const selected = selectByScope(catalog.chapters.filter(ref => shouldDownload(manga, ref, source, item.existingPolicy)), item.scope); + job.total = selected.length; job.jobCompleted = 0; job.current = null; await record(); + let publishedCount = 0; + const series = { title: remote.title, direction: ['rtl', 'ltr', 'ttb'].includes(remote.direction) ? remote.direction : 'rtl', + sources: [{ name: `Peer ${peer.alias}`, url: sourceUrl }], + metadata: { title: remote.title }, scraper: { ids: { [source]: info.id } } }; + for (let index = 0; index < selected.length; index++) { + stop(); + const ref = selected[index]; + if (!shouldDownload(manga, ref, source, item.existingPolicy)) { job.jobCompleted++; await record(); continue; } + const chapter = await client.chapter(stateStore, info.peerId, info.shareId, ref.id); + if (String(chapter.number) !== String(ref.number) || String(chapter.volume) !== String(ref.volume)) throw new Error('Peer chapter identity changed'); + const digest = crypto.createHash('sha256').update(JSON.stringify(chapter.pages.map(p => [p.id, p.sha256]))).digest('hex').slice(0, 20); + const stage = path.join(workRoot, 'pages', `${ref.id}-${digest}`); + await fs.mkdir(stage, { recursive: true }); + job.current = { number: ref.number, sourceId: ref.id, position: index + 1, total: selected.length, + downloadedPages: 0, expectedPages: chapter.pages.length }; + await record(); + const pages = chapter.pages.map((page, i) => ({ ...page, path: path.join(stage, `${String(i + 1).padStart(4, '0')}.${page.mime.split('/')[1] === 'jpeg' ? 'jpg' : page.mime.split('/')[1]}`) })); + const concurrency = Math.min(4, Math.max(1, Number(requirePeer(stateStore.load(), info.peerId).download?.concurrent || 2))); + for (let offset = 0; offset < pages.length; offset += concurrency) { + stop(); + const batch = pages.slice(offset, offset + concurrency); + await Promise.all(batch.map(page => client.downloadPage(stateStore, info.peerId, page, page.path))); + job.current.downloadedPages = Math.min(pages.length, offset + batch.length); + await record(); + stop(); + } + const publication = await publishManifest({ version: 1, series, chapters: [{ + number: ref.number, title: ref.title, book: { number: Number(ref.volume) || 1 }, + source: { name: `Peer ${peer.alias} ${peer.id.slice(1, 7)}`, url: sourceUrl }, + pages: pages.map(page => ({ path: page.path })), + }] }, { libraryRoot: options.libraryRoot, ...(manga ? { mangaDir: manga.path } : {}) }); + if (publication.results[0].status === 'published') { + publishedCount++; + if (!manga) manga = { id: path.basename(publication.mangaDir), path: publication.mangaDir, chapters: [], meta: {} }; + manga.chapters.push({ book: String(Number(ref.volume) || 1), number: String(ref.number), sourceKey: source }); + if (options.onPublished) await options.onPublished({ mangaDir: publication.mangaDir, chapter: ref.number }); + } + job.jobCompleted++; job.lastChapter = ref.number; job.current = null; await record(); + // Only a fully verified and published chapter loses its resumable page cache. + await fs.rm(stage, { recursive: true, force: true }); + if (index < selected.length - 1) { stop(); await sleep(Math.max(500, Number(options.settings.chapterMinMs || 500))); } + } + job.status = 'complete'; job.current = null; await record(); + return { finalPath: manga?.path || null, publishedCount, selectedCount: selected.length }; +} +function adapterForPeer(value) { + const info = infoFor(value); + if (!info) return null; + const adapter = { id: info.source, label: `Peer ${info.peerId.slice(1, 7)}`, + async discover() { return discover(info); }, + async run(item, options) { return run(info, item, options); } }; + return { adapter, info }; +} +module.exports = { adapterForPeer, infoFor, discover, run }; diff --git a/peering/client.js b/peering/client.js new file mode 100644 index 0000000..bf5a39b --- /dev/null +++ b/peering/client.js @@ -0,0 +1,146 @@ +'use strict'; + +const https = require('node:https'); +const fs = require('node:fs'); +const fsp = fs.promises; +const crypto = require('node:crypto'); +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.binary); + 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 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'); + 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'); + first ||= result; chapters.push(...result.chapters); cursor = result.nextCursor; + if (!cursor) return { id: shareId, title: first.title, direction: first.direction, 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, 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, chapter, downloadPage }; diff --git a/peering/gateway.js b/peering/gateway.js new file mode 100644 index 0000000..dce367c --- /dev/null +++ b/peering/gateway.js @@ -0,0 +1,236 @@ +#!/usr/bin/env node +'use strict'; + +const https = require('node:https'); +const fs = require('node:fs'); +const fsp = fs.promises; +const path = require('node:path'); +const crypto = require('node:crypto'); +const { once } = require('node:events'); +const { PeeringState, fp, hash, requirePeer, grantStatus } = require('./state'); +const { parseKey, parseSpecialFolder } = require('../chapter-identity'); + +const DATA = path.resolve(process.env.PEER_DATA || path.join(process.env.DATA || 'data', 'peering')); +const LIBRARY = path.resolve(process.env.LIBRARY || 'library'); +const HOST = process.env.PEER_HOST || '127.0.0.1'; +const PORT = Number(process.env.PEER_PORT || 8093); +const peering = new PeeringState(DATA); +const identity = peering.identity(); +if (!identity) throw new Error('Generate peering identity in the admin UI before starting the gateway'); +const MAX_PAGE = 30 * 1024 * 1024; +const MIME = { '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.webp': 'image/webp', '.gif': 'image/gif', '.avif': 'image/avif', '.bmp': 'image/bmp' }; +const usageFile = path.resolve(process.env.PEER_USAGE_FILE || path.join(DATA, 'gateway-usage.json')); +fs.mkdirSync(path.dirname(usageFile), { recursive: true, mode: 0o700 }); +const usage = (() => { try { return JSON.parse(fs.readFileSync(usageFile, 'utf8')); } catch (e) { if (e.code === 'ENOENT') return { day: '', peers: {}, usedInvites: [] }; throw e; } })(); +const active = new Map(); +const uploadSlots = new Map(); +const GLOBAL_STREAMS = Math.max(1, Math.min(32, Number(process.env.PEER_GLOBAL_STREAMS || 4))); +const indexCache = new Map(); +const blobCache = new Map(); +const sha = s => crypto.createHash('sha256').update(s).digest('hex'); +const opaque = s => sha(s).slice(0, 32); +function saveUsage() { const tmp = `${usageFile}.tmp`; fs.writeFileSync(tmp, JSON.stringify(usage), { mode: 0o600 }); fs.renameSync(tmp, usageFile); } +function fail(code, message, retryAfter) { const e = new Error(message); e.code = code; e.retryAfter = retryAfter; throw e; } +function json(res, code, value, extra = {}) { + res.writeHead(code, { 'content-type': 'application/json; charset=utf-8', 'cache-control': 'no-store', ...extra }); + res.end(JSON.stringify(value)); +} +function peerOf(req) { + const cert = req.socket.getPeerCertificate(true); + if (!cert?.raw || Date.parse(cert.valid_to) <= Date.now() || Date.parse(cert.valid_from) > Date.now()) fail(401, 'Valid client certificate required'); + const fingerprint = fp(cert.raw); + const state = peering.load(); + const peer = state.peers.find(p => p.fingerprint === fingerprint); + if (!peer || fp(peer.cert) !== fingerprint) fail(401, 'Unrecognized client certificate'); + return { state, peer }; +} +function grant(req, peer, scope = 'catalog') { + const status = grantStatus(peer); + if (!peer.paired || status.status !== 'active') fail(403, `Peering grant ${status.status}`); + const match = /^Bearer ([A-Za-z0-9_-]{40,60})$/.exec(String(req.headers.authorization || '')); + if (!match) fail(401, 'Grant required'); + const got = Buffer.from(hash(match[1]), 'hex'), want = Buffer.from(peer.grantHash, 'hex'); + if (got.length !== want.length || !crypto.timingSafeEqual(got, want)) fail(401, 'Invalid grant'); + if (!status.scopes.includes(scope)) fail(403, `Grant does not allow ${scope}`); +} +function quota(peer, bytes = 0, catalog = false) { + const now = new Date(), day = now.toISOString().slice(0, 10), minute = Math.floor(Date.now() / 60000); + if (usage.day !== day) { usage.day = day; usage.peers = {}; saveUsage(); } + const entry = usage.peers[peer.id] ||= { bytes: 0, minute, catalogs: 0 }; + if (entry.minute !== minute) { entry.minute = minute; entry.catalogs = 0; } + if (catalog && entry.catalogs >= peer.limits.catalogPerMinute) fail(429, 'Catalog request limit reached', 60); + if (entry.bytes + bytes > peer.limits.bytesPerDay) fail(429, 'Daily upload limit reached', 3600); + if (catalog) entry.catalogs++; + if (bytes) entry.bytes += bytes; // Reserve before streaming; an interrupted stream never resets quota. + saveUsage(); +} +function safeDir(parent, name) { + if (typeof name !== 'string' || !name || name.startsWith('.') || name.includes('/') || name.includes('\\') || name === '..') return null; + const file = path.join(parent, name); + if (!file.startsWith(parent + path.sep)) return null; + try { if (!fs.lstatSync(file).isDirectory()) return null; } catch { return null; } + return file; +} +async function dirs(root) { + try { return (await fsp.readdir(root, { withFileTypes: true })).filter(e => e.isDirectory() && !e.isSymbolicLink() && !e.name.startsWith('.')).map(e => e.name).sort(); } + catch { return []; } +} +async function jsonFile(file) { try { return JSON.parse(await fsp.readFile(file, 'utf8')); } catch { return {}; } } +async function mangaIndex(peer, share) { + const key = `${peer.id}:${share.id}`; + const old = indexCache.get(key); + if (old && Date.now() - old.at < 30000) return old; + const mangaDir = safeDir(LIBRARY, share.mangaId); + if (!mangaDir) fail(404, 'Manga not found'); + const meta = await jsonFile(path.join(mangaDir, 'manga.json')); + const chapters = [], byId = new Map(); + for (const bookName of await dirs(mangaDir)) { + const book = parseKey(bookName); + if (!book) continue; + const bookDir = safeDir(mangaDir, bookName); + for (const chapterName of await dirs(bookDir)) { + const parsed = parseSpecialFolder(chapterName) || parseKey(chapterName); + if (!parsed) continue; + const chapterDir = safeDir(bookDir, chapterName); + // A chapter without final metadata may still be a manually written partial. + if (!fs.existsSync(path.join(chapterDir, 'chapter.json'))) continue; + const entries = await fsp.readdir(chapterDir, { withFileTypes: true }); + const pages = entries.filter(e => e.isFile() && MIME[path.extname(e.name).toLowerCase()]).map(e => e.name).sort(); + if (!pages.length) continue; + const id = `c${opaque(`${share.id}/${bookName}/${chapterName}`)}`; + const chapterMeta = await jsonFile(path.join(chapterDir, 'chapter.json')); + const item = { id, number: parsed.key, volume: book.key, title: String(chapterMeta.title || parsed.title || '').slice(0, 200), + pageCount: pages.length, chapterDir, pages }; + chapters.push(item); byId.set(id, item); + if (chapters.length > 10000) fail(413, 'Shared manga chapter limit exceeded'); + } + } + const result = { at: Date.now(), mangaDir, meta, chapters, byId }; + indexCache.set(key, result); + return result; +} +function permittedShare(peer, id) { + const share = (peer.shares || []).find(s => s.id === id); + if (!share) fail(404, 'Manga not found'); + return share; +} +async function chapterManifest(peer, share, chapterId) { + const index = await mangaIndex(peer, share), chapter = index.byId.get(chapterId); + if (!chapter) fail(404, 'Chapter not found'); + const pages = []; + for (const name of chapter.pages) { + const file = path.join(chapter.chapterDir, name), stat = await fsp.lstat(file); + if (!stat.isFile() || stat.size < 1 || stat.size > MAX_PAGE) fail(413, 'Page size outside transfer limits'); + const digest = crypto.createHash('sha256'); + for await (const chunk of fs.createReadStream(file)) digest.update(chunk); + const id = `b${opaque(`${share.id}/${chapter.id}/${name}`)}`; + const etag = `"${id}-${stat.size}-${Math.round(stat.mtimeMs)}"`; + pages.push({ id, bytes: stat.size, sha256: digest.digest('hex'), mime: MIME[path.extname(name).toLowerCase()], etag }); + blobCache.set(`${peer.id}:${id}`, { file, chapterDir: chapter.chapterDir, shareId: share.id, chapterId, + size: stat.size, mtimeMs: stat.mtimeMs, at: Date.now() }); + if (pages.length > 500) fail(413, 'Chapter page limit exceeded'); + } + return { id: chapter.id, number: chapter.number, volume: chapter.volume, title: chapter.title, pages }; +} +async function streamBlob(req, res, peer, id) { + const item = blobCache.get(`${peer.id}:${id}`); + if (!item || Date.now() - item.at > 30 * 60 * 1000) fail(404, 'Refresh chapter manifest'); + permittedShare(peer, item.shareId); + if (!fs.existsSync(path.join(item.chapterDir, 'chapter.json'))) fail(404, 'Chapter no longer complete'); + const stat = await fsp.lstat(item.file).catch(() => null); + if (!stat?.isFile() || stat.size !== item.size || stat.mtimeMs !== item.mtimeMs) fail(412, 'Page changed; refresh manifest'); + const etag = `"${id}-${stat.size}-${Math.round(stat.mtimeMs)}"`; + if (req.headers['if-match'] && req.headers['if-match'] !== etag) fail(412, 'Page changed; refresh manifest'); + const match = /^bytes=(\d+)-$/.exec(String(req.headers.range || '')); + if (req.headers.range && !match) fail(416, 'Unsupported range'); + const start = match ? Number(match[1]) : 0; + if (!Number.isSafeInteger(start) || start >= stat.size) fail(416, 'Range outside page'); + const count = stat.size - start; + if ((active.get(peer.id) || 0) >= peer.limits.concurrent) fail(429, 'Too many transfers', 2); + if ([...active.values()].reduce((a, b) => a + b, 0) >= GLOBAL_STREAMS) fail(429, 'Server transfer limit reached', 2); + quota(peer, count); + active.set(peer.id, (active.get(peer.id) || 0) + 1); + try { + res.writeHead(start ? 206 : 200, { + 'content-type': MIME[path.extname(item.file).toLowerCase()], 'content-length': count, + 'accept-ranges': 'bytes', etag, + ...(start ? { 'content-range': `bytes ${start}-${stat.size - 1}/${stat.size}` } : {}), + 'cache-control': 'no-store', + }); + const stream = fs.createReadStream(item.file, { start, highWaterMark: 64 * 1024 }); + for await (const chunk of stream) { + if (req.destroyed || res.destroyed) break; + // Reserve each chunk on a shared per-peer timeline, including simultaneous streams. + const now = Date.now(), ready = Math.max(now, uploadSlots.get(peer.id) || now) + chunk.length * 1000 / peer.limits.bytesPerSecond; + uploadSlots.set(peer.id, ready); + if (ready > now) await new Promise(resolve => setTimeout(resolve, ready - now)); + if (!res.write(chunk)) await once(res, 'drain'); + } + res.end(); + } finally { active.set(peer.id, Math.max(0, (active.get(peer.id) || 1) - 1)); } +} + +const server = https.createServer({ key: identity.key, cert: identity.cert, requestCert: true, rejectUnauthorized: false, minVersion: 'TLSv1.3' }, async (req, res) => { + try { + const url = new URL(req.url, 'https://peer'); + if (req.method !== 'GET') fail(405, 'GET only'); + if (url.pathname === '/health') return json(res, 200, { ok: true }); + const { state, peer } = peerOf(req); + if (url.pathname === '/peer/v1/hello') { + if (!peer.paired) { + const invitation = state.invitations.find(i => i.id === req.headers['x-invitation-id']); + const code = String(req.headers['x-invitation-code'] || ''); + if (!invitation || Date.parse(invitation.expiresAt) <= Date.now() || invitation.codeHash !== hash(code) + || usage.usedInvites.includes(invitation.id)) fail(403, 'Valid unused invitation required'); + usage.usedInvites.push(invitation.id); usage.usedInvites = usage.usedInvites.slice(-1000); saveUsage(); + } + return json(res, 200, { version: 1, id: identity.id, fingerprint: identity.fingerprint }); + } + if (url.pathname === '/peer/v1/access') { + if (!peer.paired) fail(403, 'Verify peer identity first'); + return json(res, 200, { version: 1, paired: true, name: peering.load().name, + certificateExpiresAt: new crypto.X509Certificate(identity.cert).validTo, + grant: grantStatus(peer) }); + } + const requiredScope = url.pathname.startsWith('/peer/v1/blobs/') ? 'download' : 'catalog'; + grant(req, peer, requiredScope); + if (url.pathname === '/peer/v1/catalog') { + quota(peer, 0, true); + const cursor = Number(url.searchParams.get('cursor') || 0); + if (!Number.isSafeInteger(cursor) || cursor < 0) fail(400, 'Invalid cursor'); + const shares = peer.shares || [], items = []; + for (const share of shares.slice(cursor, cursor + 50)) { + const mangaDir = safeDir(LIBRARY, share.mangaId); + if (!mangaDir) continue; + const meta = await jsonFile(path.join(mangaDir, 'manga.json')); + items.push({ id: share.id, title: String(meta.title || share.mangaId).slice(0, 200), + year: Number(meta.year || meta.startDate?.year) || null, status: String(meta.status || 'unknown').slice(0, 40) }); + } + return json(res, 200, { version: 1, items, nextCursor: cursor + 50 < shares.length ? String(cursor + 50) : null }); + } + const mangaMatch = /^\/peer\/v1\/manga\/(m[0-9a-f]{24})$/.exec(url.pathname); + if (mangaMatch) { + quota(peer, 0, true); + const share = permittedShare(peer, mangaMatch[1]), index = await mangaIndex(peer, share); + const cursor = Number(url.searchParams.get('cursor') || 0); + if (!Number.isSafeInteger(cursor) || cursor < 0) fail(400, 'Invalid cursor'); + return json(res, 200, { id: share.id, title: String(index.meta.title || share.mangaId).slice(0, 200), + direction: ['rtl', 'ltr', 'ttb'].includes(index.meta.direction) ? index.meta.direction : 'rtl', + chapters: index.chapters.slice(cursor, cursor + 100).map(({ id, number, volume, title, pageCount }) => ({ id, number, volume, title, pageCount })), + nextCursor: cursor + 100 < index.chapters.length ? String(cursor + 100) : null }); + } + const chapterMatch = /^\/peer\/v1\/manga\/(m[0-9a-f]{24})\/chapters\/(c[0-9a-f]{32})$/.exec(url.pathname); + if (chapterMatch) { + quota(peer, 0, true); + return json(res, 200, await chapterManifest(peer, permittedShare(peer, chapterMatch[1]), chapterMatch[2])); + } + const blobMatch = /^\/peer\/v1\/blobs\/(b[0-9a-f]{32})$/.exec(url.pathname); + if (blobMatch) return await streamBlob(req, res, peer, blobMatch[1]); + fail(404, 'Not found'); + } catch (error) { + if (!res.headersSent) json(res, Number(error.code) || 500, { error: Number(error.code) ? error.message : 'Internal error' }, error.retryAfter ? { 'retry-after': String(error.retryAfter) } : {}); + else res.destroy(); + if (!Number(error.code)) console.error('[peer-gateway]', error.message); + } +}); +server.listen(PORT, HOST, () => console.log(`[peer-gateway] listening on ${HOST}:${PORT}`)); diff --git a/peering/state.js b/peering/state.js new file mode 100644 index 0000000..492601e --- /dev/null +++ b/peering/state.js @@ -0,0 +1,242 @@ +'use strict'; + +const fs = require('node:fs'); +const path = require('node:path'); +const crypto = require('node:crypto'); +const { execFileSync } = require('node:child_process'); + +const MAX_PEERS = 64; +const INVITE_MS = 48 * 60 * 60 * 1000; +const GRANT_MS = 90 * 24 * 60 * 60 * 1000; +const NAME_RE = /^[A-Za-z0-9]{1,25}$/; +const GRANT_SCOPES = new Set(['catalog', 'download']); +const fp = pem => new crypto.X509Certificate(pem).fingerprint256.replaceAll(':', '').toLowerCase(); +const hash = value => crypto.createHash('sha256').update(value).digest('hex'); +const peerId = pem => `p${fp(pem).slice(0, 24)}`; +const own = (root, name) => path.join(root, name); +function readJson(file, fallback) { + try { return JSON.parse(fs.readFileSync(file, 'utf8')); } + catch (e) { if (e.code === 'ENOENT') return fallback; throw e; } +} +function atomic(file, value) { + const temp = `${file}.${crypto.randomBytes(6).toString('hex')}.tmp`; + fs.writeFileSync(temp, JSON.stringify(value, null, 2) + '\n', { mode: 0o600 }); + fs.renameSync(temp, file); +} +function requirePeer(state, id) { + const peer = state.peers.find(p => p.id === id); + if (!peer) throw new Error('Peer not found'); + return peer; +} +function validUrl(value) { + const url = new URL(String(value)); + if (url.protocol !== 'https:' || url.username || url.password || url.search || url.hash || url.pathname !== '/') throw new Error('Peer URL must be an HTTPS origin'); + if (!url.hostname || url.hostname.length > 253) throw new Error('Invalid peer hostname'); + return url.origin; +} +function publicPeer(peer, outbound) { + const grantValid = !!peer.grantHash && Date.parse(peer.grantExpiresAt) > Date.now(); + return { + id: peer.id, alias: peer.alias, url: peer.url, fingerprint: peer.fingerprint, + paired: !!peer.paired, serve: grantValid && !peer.servingPaused, receive: !!peer.receive, + grantStatus: !peer.grantHash ? 'none' : !grantValid ? 'expired' : peer.servingPaused ? 'paused' : 'active', + grantScopes: peer.grantHash ? scopesFor(peer) : [], grantIssuedAt: peer.grantIssuedAt || null, + readerQueue: !!peer.readerQueue, shares: peer.shares || [], + limits: peer.limits, download: peer.download, + grantExpiresAt: peer.grantExpiresAt || null, + hasOutboundGrant: !!outbound?.[peer.id], + invitationExpiresAt: peer.invitationExpiresAt || null, + }; +} +function serverName(value) { + const name = String(value ?? '').trim(); + if (!NAME_RE.test(name)) throw new Error('Server name must contain 1–25 letters or numbers'); + return name; +} +function normalizeScopes(value) { + const scopes = [...new Set((Array.isArray(value) ? value : []).map(String))].filter(scope => GRANT_SCOPES.has(scope)); + if (scopes.includes('download') && !scopes.includes('catalog')) scopes.push('catalog'); + return scopes; +} +function scopesFor(peer) { + return Array.isArray(peer.grantScopes) ? normalizeScopes(peer.grantScopes) : ['catalog', 'download']; +} +function grantStatus(peer) { + if (!peer.grantHash) return { status: 'none', scopes: [], issuedAt: null, expiresAt: null }; + const scopes = scopesFor(peer), expiresAt = peer.grantExpiresAt || null; + const status = !expiresAt || Date.parse(expiresAt) <= Date.now() ? 'expired' : peer.servingPaused ? 'paused' : 'active'; + return { status, scopes: status === 'expired' ? [] : scopes, issuedAt: peer.grantIssuedAt || null, expiresAt }; +} +function normalizeLimits(input = {}, previous = {}) { + const defaults = { concurrent: 2, bytesPerSecond: 2 * 1024 * 1024, bytesPerDay: 10 * 1024 ** 3, catalogPerMinute: 30 }; + const result = { ...defaults, ...previous }; + for (const [key, min, max] of [['concurrent', 1, 8], ['bytesPerSecond', 65536, 100 * 1024 * 1024], + ['bytesPerDay', 1024 * 1024, 10 * 1024 ** 4], ['catalogPerMinute', 1, 600]]) { + if (input[key] === undefined) continue; + const value = Number(input[key]); + if (!Number.isSafeInteger(value) || value < min || value > max) throw new Error(`Invalid ${key}`); + result[key] = value; + } + return result; +} +function normalizeDownload(input = {}, previous = {}) { + const result = { concurrent: 2, bytesPerSecond: 2 * 1024 * 1024, ...previous }; + for (const [key, min, max] of [['concurrent', 1, 4], ['bytesPerSecond', 65536, 100 * 1024 * 1024]]) { + if (input[key] === undefined) continue; + const value = Number(input[key]); + if (!Number.isSafeInteger(value) || value < min || value > max) throw new Error(`Invalid ${key}`); + result[key] = value; + } + return result; +} + +class PeeringState { + constructor(root) { this.root = path.resolve(root); } + stateFile() { return own(this.root, 'state.json'); } + outboundFile() { return own(this.root, 'outbound.json'); } + certFile() { return own(this.root, 'identity.crt'); } + keyFile() { return own(this.root, 'identity.key'); } + load() { return readJson(this.stateFile(), { version: 1, name: '', peers: [], invitations: [] }); } + outbound() { return readJson(this.outboundFile(), {}); } + identity() { + if (!fs.existsSync(this.certFile()) || !fs.existsSync(this.keyFile())) return null; + const cert = fs.readFileSync(this.certFile(), 'utf8'); + return { cert, key: fs.readFileSync(this.keyFile(), 'utf8'), fingerprint: fp(cert), id: peerId(cert) }; + } + setName(value) { + const state = this.load(); state.name = serverName(value); atomic(this.stateFile(), state); + return state.name; + } + generateIdentity(name) { + fs.mkdirSync(this.root, { recursive: true, mode: 0o700 }); + if (this.identity()) throw new Error('Identity already exists; rotate it through a separate recovery procedure'); + const state = this.load(); + const cleanName = serverName(name || state.name); + if (state.name && state.name !== cleanName) throw new Error('Save this server name before generating its certificate'); + const openssl = process.env.OPENSSL_BIN || 'openssl'; + const key = this.keyFile(), cert = this.certFile(); + try { + execFileSync(openssl, ['req', '-x509', '-newkey', 'ec', '-pkeyopt', 'ec_paramgen_curve:P-256', '-sha256', '-nodes', + '-keyout', key, '-out', cert, '-days', '365', '-subj', `/CN=${cleanName}`], { stdio: 'pipe', timeout: 15000 }); + fs.chmodSync(key, 0o600); fs.chmodSync(cert, 0o600); + state.name = cleanName; atomic(this.stateFile(), state); + return { id: peerId(fs.readFileSync(cert, 'utf8')), fingerprint: fp(fs.readFileSync(cert, 'utf8')), name: cleanName }; + } catch (error) { + for (const file of [key, cert]) { try { fs.rmSync(file, { force: true }); } catch {} } + throw new Error(`Could not generate peering identity: ${error.message}`); + } + } + createInvitation(url) { + const identity = this.identity(); + if (!identity) throw new Error('Generate a server identity first'); + const origin = validUrl(url); + const state = this.load(); + const id = crypto.randomUUID(), code = crypto.randomBytes(32).toString('base64url'); + const expiresAt = new Date(Date.now() + INVITE_MS).toISOString(); + state.invitations = (state.invitations || []).filter(item => Date.parse(item.expiresAt) > Date.now()).slice(-50); + state.invitations.push({ id, codeHash: hash(code), expiresAt }); + atomic(this.stateFile(), state); + return { version: 1, name: state.name, url: origin, cert: identity.cert, + fingerprint: identity.fingerprint, invitationId: id, code, expiresAt }; + } + importInvitation(card) { + if (!card || card.version !== 1 || typeof card.cert !== 'string' || card.cert.length > 10000) throw new Error('Invalid invitation'); + if (Date.parse(card.expiresAt) <= Date.now() || Date.parse(card.expiresAt) > Date.now() + INVITE_MS + 60000) throw new Error('Invitation expired or invalid'); + if (!crypto.randomUUID || !/^[0-9a-f-]{36}$/i.test(String(card.invitationId))) throw new Error('Invalid invitation ID'); + if (!/^[A-Za-z0-9_-]{40,60}$/.test(String(card.code))) throw new Error('Invalid invitation code'); + const cert = new crypto.X509Certificate(card.cert); + const name = serverName(card.name); + if (Date.parse(cert.validTo) <= Date.now() || Date.parse(cert.validFrom) > Date.now()) throw new Error('Peer certificate is not valid now'); + const fingerprint = fp(card.cert); + if (fingerprint !== String(card.fingerprint).replaceAll(':', '').toLowerCase()) throw new Error('Invitation fingerprint mismatch'); + if (this.identity()?.fingerprint === fingerprint) throw new Error('Cannot pair with this server itself'); + const state = this.load(); + if (state.peers.length >= MAX_PEERS && !state.peers.some(p => p.fingerprint === fingerprint)) throw new Error('Peer limit reached'); + let peer = state.peers.find(p => p.fingerprint === fingerprint); + if (!peer) { + peer = { id: peerId(card.cert), alias: name, url: validUrl(card.url), + cert: card.cert, fingerprint, paired: false, servingPaused: false, receive: false, readerQueue: false, + shares: [], limits: normalizeLimits(), download: normalizeDownload() }; + state.peers.push(peer); + } else if (peer.url !== validUrl(card.url)) { + // A changed endpoint needs a new admin-reviewed invitation; never silently redirect traffic. + peer.url = validUrl(card.url); peer.paired = false; peer.servingPaused = false; peer.receive = false; + delete peer.grantHash; delete peer.grantScopes; delete peer.grantExpiresAt; delete peer.grantIssuedAt; + } + peer.invitationId = card.invitationId; peer.invitationCode = card.code; peer.invitationExpiresAt = card.expiresAt; + atomic(this.stateFile(), state); + return publicPeer(peer, this.outbound()); + } + updatePeer(id, patch) { + const state = this.load(), peer = requirePeer(state, id); + if (patch.alias !== undefined) peer.alias = serverName(patch.alias); + if (patch.limits !== undefined) peer.limits = normalizeLimits(patch.limits, peer.limits); + if (patch.download !== undefined) peer.download = normalizeDownload(patch.download, peer.download); + if (patch.readerQueue !== undefined) peer.readerQueue = !!patch.readerQueue; + if (patch.receive !== undefined) peer.receive = !!patch.receive && !!peer.paired; + if (patch.servingPaused !== undefined) peer.servingPaused = !!patch.servingPaused; + // Accept the legacy field without revoking credentials: false now means pause. + if (patch.serve !== undefined) { + if (patch.serve && !peer.paired) throw new Error('Verify pairing before enabling serving'); + peer.servingPaused = !patch.serve; + } + if (patch.grantScopes !== undefined) { + if (!peer.grantHash) throw new Error('Issue a grant before changing its permissions'); + const scopes = normalizeScopes(patch.grantScopes); + if (!scopes.length) throw new Error('Select at least one grant permission or revoke the grant'); + peer.grantScopes = scopes; + } + atomic(this.stateFile(), state); + return publicPeer(peer, this.outbound()); + } + setShare(id, mangaId, enabled) { + const state = this.load(), peer = requirePeer(state, id); + if (typeof mangaId !== 'string' || !mangaId || mangaId.startsWith('.') || mangaId.includes('/') || mangaId.includes('\\')) throw new Error('Invalid manga ID'); + peer.shares ||= []; + const existing = peer.shares.find(s => s.mangaId === mangaId); + if (enabled && !existing) peer.shares.push({ mangaId, id: `m${crypto.randomBytes(12).toString('hex')}` }); + if (!enabled) peer.shares = peer.shares.filter(s => s.mangaId !== mangaId); + atomic(this.stateFile(), state); + return publicPeer(peer, this.outbound()); + } + issueGrant(id, scopes = ['catalog', 'download']) { + const state = this.load(), peer = requirePeer(state, id); + if (!peer.paired) throw new Error('Verify pairing before issuing a grant'); + const selected = normalizeScopes(scopes); + if (!selected.length) throw new Error('Select at least one grant permission'); + const token = crypto.randomBytes(32).toString('base64url'); + peer.grantHash = hash(token); + peer.grantScopes = selected; peer.grantIssuedAt = new Date().toISOString(); peer.servingPaused = false; + peer.grantExpiresAt = new Date(Math.min(Date.now() + GRANT_MS, Date.parse(new crypto.X509Certificate(peer.cert).validTo), Date.parse(new crypto.X509Certificate(this.identity().cert).validTo))).toISOString(); + atomic(this.stateFile(), state); + return { peerId: peer.id, token, expiresAt: peer.grantExpiresAt }; + } + revokeGrant(id) { + const state = this.load(), peer = requirePeer(state, id); + for (const key of ['grantHash', 'grantScopes', 'grantIssuedAt', 'grantExpiresAt']) delete peer[key]; + peer.servingPaused = false; atomic(this.stateFile(), state); + return publicPeer(peer, this.outbound()); + } + acceptGrant(id, token) { + const state = this.load(), peer = requirePeer(state, id); + if (!/^[A-Za-z0-9_-]{40,60}$/.test(String(token))) throw new Error('Invalid grant token'); + const outbound = this.outbound(); outbound[id] = String(token); atomic(this.outboundFile(), outbound); + return publicPeer(peer, outbound); + } + removePeer(id) { + const state = this.load(); requirePeer(state, id); + state.peers = state.peers.filter(p => p.id !== id); atomic(this.stateFile(), state); + const outbound = this.outbound(); delete outbound[id]; atomic(this.outboundFile(), outbound); + } + publicState() { + const state = this.load(), identity = this.identity(), outbound = this.outbound(); + return { identity: identity ? { id: identity.id, fingerprint: identity.fingerprint, name: state.name, + expiresAt: new crypto.X509Certificate(identity.cert).validTo } : null, + peers: state.peers.map(p => publicPeer(p, outbound)), + invitations: (state.invitations || []).filter(i => Date.parse(i.expiresAt) > Date.now()).map(i => ({ expiresAt: i.expiresAt })), + certificateExpiresAt: identity ? new crypto.X509Certificate(identity.cert).validTo : null }; + } +} + +module.exports = { PeeringState, fp, hash, peerId, validUrl, requirePeer, normalizeLimits, normalizeDownload, + serverName, normalizeScopes, scopesFor, grantStatus }; diff --git a/peering/test/peering.test.js b/peering/test/peering.test.js new file mode 100644 index 0000000..01aded0 --- /dev/null +++ b/peering/test/peering.test.js @@ -0,0 +1,137 @@ +'use strict'; +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const fsp = fs.promises; +const os = require('node:os'); +const path = require('node:path'); +const net = require('node:net'); +const https = require('node:https'); +const { spawn } = require('node:child_process'); +const { PeeringState } = require('../state'); +const client = require('../client'); + +const ROOT = path.resolve(__dirname, '../..'); +const OPENSSL = process.env.OPENSSL_BIN || (process.platform === 'win32' ? 'C:\\Program Files\\Git\\usr\\bin\\openssl.exe' : 'openssl'); +const PNG = Buffer.from('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9Y9Zr5sAAAAASUVORK5CYII=', 'base64'); +async function port() { + const server = net.createServer(); await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + const value = server.address().port; await new Promise(resolve => server.close(resolve)); return value; +} +async function health(value) { + return new Promise((resolve, reject) => { + https.get({ hostname: '127.0.0.1', port: value, path: '/health', rejectUnauthorized: false }, res => { + res.resume(); res.on('end', () => resolve(res.statusCode)); + }).on('error', reject); + }); +} +async function gateway(data, library, value, usage) { + const child = spawn(process.execPath, ['peering/gateway.js'], { cwd: ROOT, env: { ...process.env, + PEER_DATA: data, LIBRARY: library, PEER_PORT: String(value), PEER_HOST: '127.0.0.1', PEER_USAGE_FILE: usage }, stdio: ['ignore', 'pipe', 'pipe'] }); + let error = ''; child.stderr.on('data', chunk => { error += chunk.toString(); }); + for (let i = 0; i < 50; i++) { + if (child.exitCode !== null) throw new Error(`Gateway exited: ${error}`); + try { if (await health(value) === 200) return child; } catch {} + await new Promise(resolve => setTimeout(resolve, 100)); + } + child.kill(); throw new Error(`Gateway failed to start: ${error}`); +} +async function stop(child) { + if (child.exitCode !== null) return; + child.kill('SIGTERM'); await new Promise(resolve => { child.once('exit', resolve); setTimeout(resolve, 3000).unref(); }); +} + +test('mTLS pairing, opt-in catalog, peer queue publication, and revocation', { timeout: 30000 }, async t => { + const root = await fsp.mkdtemp(path.join(os.tmpdir(), 'mangacircus-peering-')); + const dataA = path.join(root, 'a-data'), dataB = path.join(root, 'b-data'); + const libA = path.join(root, 'a-library'), libB = path.join(root, 'b-library'); + await Promise.all([dataA, dataB, libA, libB].map(dir => fsp.mkdir(dir, { recursive: true }))); + t.after(async () => { await fsp.rm(root, { recursive: true, force: true }); }); + const a = new PeeringState(dataA), b = new PeeringState(dataB); + process.env.OPENSSL_BIN = OPENSSL; + a.generateIdentity('ServerA'); b.generateIdentity('ServerB'); + const beforeInvalidInvite = a.load().invitations.length; + assert.throws(() => a.createInvitation('https://localhost:8080/path'), /HTTPS origin/); + assert.throws(() => a.createInvitation('localhost:8080'), /HTTPS origin/); + assert.equal(a.load().invitations.length, beforeInvalidInvite, 'Invalid URLs must not record unusable invitations'); + const portA = await port(), portB = await port(); + const inviteA = a.createInvitation(`https://127.0.0.1:${portA}/`); + const inviteB = b.createInvitation(`https://127.0.0.1:${portB}/`); + const inA = a.importInvitation(inviteB), inB = b.importInvitation(inviteA); + assert.equal(inA.paired, false); + assert.equal(a.importInvitation(inviteB).id, inA.id, 'Importing one certificate must not duplicate a peer'); + assert.throws(() => a.importInvitation({ ...inviteB, fingerprint: '0'.repeat(64) }), /fingerprint mismatch/); + assert.throws(() => a.importInvitation({ ...inviteB, expiresAt: new Date(Date.now() - 1000).toISOString() }), /expired/); + await assert.rejects(() => client.catalog(b, inB.id), /Verify peer identity first/); + const mangaDir = path.join(libA, 'Sample Manga'); + const chapterDir = path.join(mangaDir, 'v01', 'c0001 [Local]'); + await fsp.mkdir(chapterDir, { recursive: true }); + await fsp.writeFile(path.join(mangaDir, 'manga.json'), JSON.stringify({ title: 'Sample Manga', direction: 'ltr' })); + await fsp.writeFile(path.join(chapterDir, 'chapter.json'), JSON.stringify({ title: 'The first chapter', source: { name: 'Local' } })); + await fsp.writeFile(path.join(chapterDir, '0001.png'), PNG); + const ga = await gateway(dataA, libA, portA, path.join(root, 'usage-a.json')); + const gb = await gateway(dataB, libB, portB, path.join(root, 'usage-b.json')); + t.after(async () => { await stop(ga); await stop(gb); }); + await client.probe(b, inB.id); + await client.probe(a, inA.id); + assert.equal(a.publicState().peers[0].paired, true); + a.setShare(inA.id, 'Sample Manga', true); + const c = new PeeringState(path.join(root, 'c-data')); + c.generateIdentity('ServerC'); + const inC = a.importInvitation(c.createInvitation('https://third-peer.example:8093/')); + assert.equal(a.publicState().peers.length, 2); + assert.deepEqual(a.publicState().peers.find(p => p.id === inC.id).shares, []); + assert.equal(a.publicState().peers.find(p => p.id === inA.id).shares.length, 1); + a.updatePeer(inA.id, { serve: true }); + const grant = a.issueGrant(inA.id, ['catalog']); + b.acceptGrant(inB.id, grant.token); + const listing = await client.catalog(b, inB.id); + assert.equal(listing.length, 1); + assert.equal(listing[0].title, 'Sample Manga'); + assert.equal(JSON.stringify(listing).includes('Sample Manga\\v01'), false); + const remote = await client.manga(b, inB.id, listing[0].id); + assert.equal(remote.chapters.length, 1); + const remoteAccess = await client.access(b, inB.id); + assert.equal(remoteAccess.grant.status, 'active'); + assert.deepEqual(remoteAccess.grant.scopes, ['catalog']); + assert.throws(() => client.connection(b, inB.id, true, true), /Downloads from this peer are disabled/); + b.updatePeer(inB.id, { receive: true }); + const manifest = await client.chapter(b, inB.id, listing[0].id, remote.chapters[0].id); + assert.equal(manifest.pages.length, 1); + assert.equal(manifest.pages[0].sha256, require('node:crypto').createHash('sha256').update(PNG).digest('hex')); + await assert.rejects(() => client.downloadPage(b, inB.id, manifest.pages[0], path.join(root, 'denied.png')), + /Peer HTTP 403: Grant does not allow download/); + a.updatePeer(inA.id, { grantScopes: ['catalog', 'download'] }); + assert.deepEqual((await client.access(b, inB.id)).grant.scopes, ['catalog', 'download'], 'Permission edits apply without exchanging a new token'); + await assert.rejects(() => client.jsonRequest(b, inB.id, '/peer/v1/manga/m000000000000000000000000'), /Peer HTTP 404/); + await assert.rejects(() => client.jsonRequest(b, inB.id, '/peer/v1/catalog', { headers: { authorization: 'Bearer ' + 'x'.repeat(43) } }), /Peer HTTP 401/); + const noCert = await new Promise((resolve, reject) => https.get({ hostname: '127.0.0.1', port: portA, + path: '/peer/v1/catalog', rejectUnauthorized: false }, res => { res.resume(); res.on('end', () => resolve(res.statusCode)); }).on('error', reject)); + assert.equal(noCert, 401); + const page = manifest.pages[0], { agent } = client.connection(b, inB.id); + const partial = await new Promise((resolve, reject) => https.get(`https://127.0.0.1:${portA}/peer/v1/blobs/${page.id}`, + { agent, headers: { authorization: `Bearer ${b.outbound()[inB.id]}`, range: 'bytes=4-', 'if-match': page.etag } }, res => { + const chunks = []; res.on('data', chunk => chunks.push(chunk)); res.on('end', () => resolve({ status: res.statusCode, body: Buffer.concat(chunks), range: res.headers['content-range'] })); + }).on('error', reject)); + assert.equal(partial.status, 206); + assert.equal(partial.range, `bytes 4-${PNG.length - 1}/${PNG.length}`); + assert.deepEqual(partial.body, PNG.subarray(4)); + const { infoFor, run } = require('../adapter'); + process.env.PEER_DATA = dataB; + const uri = `peer://${inB.id}/${listing[0].id}`; + const info = infoFor(uri); + const result = await run(info, { scope: { mode: 'missing' }, existingPolicy: 'skip-any' }, + { libraryRoot: libB, settings: { chapterMinMs: 500 }, shouldStop: () => false }); + assert.equal(result.publishedCount, 1); + assert.equal(fs.existsSync(path.join(libB, 'Sample Manga', 'v01', 'c0001 The first chapter [Peer ServerA ' + inB.id.slice(1, 7) + ']', '0001.png')), true); + const second = await run(info, { scope: { mode: 'missing' }, existingPolicy: 'skip-any' }, + { libraryRoot: libB, settings: { chapterMinMs: 500 }, shouldStop: () => false }); + assert.equal(second.publishedCount, 0); + a.updatePeer(inA.id, { servingPaused: true }); + assert.equal((await client.access(b, inB.id)).grant.status, 'paused'); + await assert.rejects(() => client.catalog(b, inB.id), /Peer HTTP 403/); + assert.equal(a.publicState().peers.find(p => p.id === inA.id).grantStatus, 'paused', 'Pausing must retain the grant'); + a.revokeGrant(inA.id); + assert.equal(a.publicState().peers.find(p => p.id === inA.id).grantStatus, 'none'); + await assert.rejects(() => client.chapter(b, inB.id, listing[0].id, remote.chapters[0].id), /Peer HTTP 403/); +}); diff --git a/tools/test-peer-ui.js b/tools/test-peer-ui.js new file mode 100644 index 0000000..7ceb9d3 --- /dev/null +++ b/tools/test-peer-ui.js @@ -0,0 +1,72 @@ +'use strict'; +const assert = require('node:assert/strict'); +const fs = require('node:fs'), os = require('node:os'), path = require('node:path'); +const { PeeringState } = require('../peering/state'); + +module.exports = async function testPeerUI(url, cookie) { + const { chromium } = require('playwright'); + const browser = await chromium.launch({ channel: process.env.QA_BROWSER_CHANNEL || 'msedge', headless: true }); + const context = await browser.newContext({ viewport: { width: 1280, height: 900 } }); + const [name, ...value] = cookie.split('='); + await context.addCookies([{ name, value: value.join('='), url }]); + const page = await context.newPage(), errors = []; + page.on('pageerror', error => errors.push(error.message)); + try { + await page.goto(url + '/peers.html'); + await page.getByRole('heading', { name: 'Peer catalogs', exact: true }).waitFor(); + let state = await page.evaluate(() => fetch('/api/admin/peering').then(response => response.json())); + if (!state.identity) { + await page.getByRole('button', { name: 'Generate local certificate' }).click(); + await page.getByText('Identity generated').waitFor(); + state = await page.evaluate(() => fetch('/api/admin/peering').then(response => response.json())); + } + assert.match(state.identity.fingerprint, /^[0-9a-f]{64}$/); + await page.getByLabel('Reachable peer-gateway URL').fill('http://localhost:8080'); + await page.getByRole('button', { name: 'Create 48-hour invitation' }).click(); + assert.match(await page.getByLabel('Reachable peer-gateway URL').evaluate(el => el.validationMessage), /HTTPS peer-gateway origin/); + await page.getByLabel('Reachable peer-gateway URL').fill('https://peer-test.example:8093'); + await page.getByRole('button', { name: 'Create 48-hour invitation' }).click(); + await page.waitForFunction(() => document.querySelector('textarea[aria-label="Invitation to exchange"]')?.value.includes('fingerprint')); + const invitation = JSON.parse(await page.getByLabel('Invitation to exchange').inputValue()); + assert.equal(invitation.url, 'https://peer-test.example:8093'); + assert.equal(invitation.fingerprint, state.identity.fingerprint); + await page.getByLabel('Other server invitation').fill('{bad json'); + await page.getByRole('button', { name: 'Import invitation' }).click(); + await page.getByText('Invalid invitation JSON').waitFor(); + const peerDir = fs.mkdtempSync(path.join(os.tmpdir(), 'mc-peer-ui-')); + try { + if (process.platform === 'win32' && !process.env.OPENSSL_BIN) process.env.OPENSSL_BIN = 'C:\\Program Files\\Git\\usr\\bin\\openssl.exe'; + const peer = new PeeringState(peerDir); + peer.generateIdentity('UIFixturePeer'); + const fixtureInvitation = peer.createInvitation('https://fixture-peer.example:8093'); + await page.getByLabel('Other server invitation').fill(JSON.stringify(fixtureInvitation)); + await page.getByRole('button', { name: 'Import invitation' }).click(); + await page.getByRole('heading', { name: 'UI fixture peer' }).waitFor(); + const card = page.locator('.peer-card').filter({ has: page.getByRole('heading', { name: 'UI fixture peer' }) }); + await card.getByText('Transfer limits').click(); + await card.getByText('Approved manga (0)').click(); + await card.getByText('Access grant').click(); + await card.getByLabel('Moon Garden').check(); + await card.getByText('Approved manga (1)').waitFor(); + assert.equal(await card.locator('details[open]').count(), 3, 'Saving a share must preserve all expanded sections'); + } finally { fs.rmSync(peerDir, { recursive: true, force: true }); } + assert.equal(errors.length, 0, `Browser errors: ${errors.join('; ')}`); + await page.setViewportSize({ width: 390, height: 844 }); + assert.ok(await page.evaluate(() => document.documentElement.scrollWidth <= innerWidth), 'Peering page overflows on mobile'); + // An ordinary reader may browse, but cannot operate certificate controls. + const created = await page.evaluate(async () => { + const response = await fetch('/api/admin/user', { method: 'POST', headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ name: 'peer-reader-qa', password: 'disposable-reader-test', admin: false }) }); + return response.status; + }); + assert.equal(created, 200); + await page.evaluate(() => fetch('/api/logout', { method: 'POST', headers: { 'content-type': 'application/json' }, body: '{}' })); + await page.evaluate(() => fetch('/api/login', { method: 'POST', headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ name: 'peer-reader-qa', password: 'disposable-reader-test' }) })); + await page.reload(); + await page.getByRole('heading', { name: 'Peer catalogs', exact: true }).waitFor(); + assert.equal(await page.getByRole('heading', { name: 'Server identity & invitations' }).count(), 0); + assert.equal((await page.evaluate(() => fetch('/api/admin/peering').then(response => response.status))), 403); + assert.equal(errors.length, 0, `Browser errors: ${errors.join('; ')}`); + } finally { await browser.close(); } +};