openvibers/openvibe.sources Commits
Per-actor rate limits on Sources' /api/v1 (roadmap WS-R task 4; openvibe-sdk v0.12.0 is now a dependency). server/api/actor-limits.js counts each request by the principal that passed the route's capability guard (req.principal.sub: svc:news, app:app_…), before any work. Past a limit a route answers 429 problem+json rate_limited with Retry-After, logs one [limits] line and counts sources_rate_limited_total{limit,window}.
Changes
@@@ -17,6 +17,12 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000#OV_NETWORK_ISSUER=https://openvibe.network#OV_NETWORK_PUBLIC_KEY=++# Per-actor limits (server/api/actor-limits.js): API reads one app or module (app:…, mod:…) may make per++# minute and per hour; past them 429 rate_limited with Retry-After. First-party services (svc:…) are not++# counted on reads; writes (sources.source.manage) have their own numbers (README).++#SOURCES_LIMITS_MINUTE=120++#SOURCES_LIMITS_HOUR=3000++# This service's own principal (svc:sources) — publishes sources.* events (events.event.publish).OV_OAUTH_CLIENT_ID=sourcesOV_OAUTH_CLIENT_SECRET=@@@ -129,6 +129,26 @@ it contains. The capability ids (first proposed in [docs/capabilities-proposal/]are released in `openvibe-contracts` v0.12.0 (this repo pins v0.33.0); [server/auth.js](server/auth.js)decides them with the contracts grant rule.++### Per-actor limits++++Every API route also limits the principal that passed its capability guard, before any work:++`server/api/actor-limits.js`, openvibe-sdk/limits, roadmap WS-R task 4. Past a limit: `429`++problem+json `rate_limited` with `Retry-After`, one `[limits]` log line and++`sources_rate_limited_total{limit,window}`.++++| Routes | Per principal, a minute / an hour |++|---|---|++| Reads (`sources.source.read`, `sources.item.read`) by an app or module (`app:…`, `mod:…`) | `SOURCES_LIMITS_MINUTE` / `SOURCES_LIMITS_HOUR` (120 / 3000) |++| `POST`, `PATCH`, `DELETE /api/v1/sources[/:key]` | 30 / 300 |++| `POST /api/v1/sources/:key/fetch` | 10 / 100 |++| `POST /api/v1/sources/:key/items` (manual) | 60 / 1200 |++| `DELETE /api/v1/items/:id` (a takedown may cover many items) | 120 / 3000 |++++A first-party service (`svc:…`) is not counted on reads: News, Reviews and Trade pull cursor pages++for all their readers, and Wiki asks about every item its editors cite. Sources' own workers (the++scheduler, fetches, the outbox relay) run in-process and never pass through HTTP. Never limited:++`/api/health`, `/api/ready`, `/release.json`, `/metrics` and the home page. `test/actor-limits.test.js`.++## Events (transactional outbox → OpenVibe.Events when `EVENTS_URL` is set)- `sources.item.created`, `sources.item.updated` (with the previous content hash), `sources.item.removed` (with the reason)@@@ -14,6 +14,7 @@"express": "^4.21.2","fast-xml-parser": "^4.5.3","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"},"devDependencies": {@@@ -1005,6 +1006,15 @@"node": ">=20"}},++ "node_modules/openvibe-sdk": {++ "version": "0.12.0",++ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0",++ "integrity": "sha512-iG1h905O1f7y/PnP8xLA6GCVzXEBDfSaBj2tc1ucNreWg2sT/AJWosb8NW0WRMK5WS4sIQ6tMsprmDLj/fjjeQ==",++ "license": "MIT",++ "engines": {++ "node": ">=20"++ }++ },"node_modules/openvibe-shared": {"version": "1.22.0","resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",@@@ -28,6 +28,7 @@"express": "^4.21.2","fast-xml-parser": "^4.5.3","openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",++ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0","openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"},"engines": {nnew file mode 100644@@@ -0,0 +1,88 @@++'use strict';++/**++ * Per-actor rate limits on /api/v1 (roadmap WS-R task 4; openvibe-sdk/limits).++ *++ * Every API route takes a service token and one capability (auth.js); the limits count requests by++ * the principal that passed that guard (req.principal.sub: svc:news, app:app_…), before the route does++ * any work. Sources' own workers (the scheduler, fetches, the outbox relay) run in this process and++ * never go through HTTP, so nothing here can slow them.++ *++ * Reads (sources.source.read, sources.item.read) take SOURCES_LIMITS_MINUTE / SOURCES_LIMITS_HOUR,++ * 120 and 3000, for an app or module (app:…, mod:…). A first-party service (svc:…) is not counted on++ * reads: News, Reviews and Trade pull cursor pages for all their readers, and Wiki asks about every++ * item its editors cite, so one budget for the whole service would refuse real people's work. The API++ * answers only on the production host (nginx serves the public host nothing but health, ready and++ * release.json).++ *++ * Writes (sources.source.manage, staff tooling) are counted for every principal, each with its own++ * budget below.++ *++ * Past a limit the route answers 429 problem+json `rate_limited` with Retry-After; the refusal is logged++ * once and counted in sources_rate_limited_total{limit,window}. Counters live in this process: a++ * restart forgets them. Never limited: /api/health, /api/ready, /release.json, /metrics and the home++ * page.++ */++const { createActorLimiter, defaultActor } = require('openvibe-sdk/limits');++++const FIRST_PARTY = /^svc:/;++++function actor(req) {++ const p = req.principal;++ if (p && typeof p.sub === 'string' && p.sub) return p.sub;++ return defaultActor(req);++}++++/** Counted on reads: anything but a first-party service (an app or module acting as itself). */++function countedRead(req) {++ const p = req.principal;++ return !(p && FIRST_PARTY.test(String(p.sub)));++}++++/** The writes, each with its numbers per principal (a minute, an hour). */++const BUDGETS = {++ // Creating, changing or deleting a source is a staff decision recorded with its terms and robots++ // notes: a form every few seconds at most.++ 'sources.source.manage': { minute: 30, hour: 300 },++ // A manual fetch requests every endpoint of the source now (the per-host spacing still applies):++ // the scheduler already polls each source, so a person's "fetch now" is a rare nudge.++ 'sources.source.fetch': { minute: 10, hour: 100 },++ // A manual item (a code or record a publisher stated, entered with its evidence URL) is stored,++ // revisioned and announced to subscribers: one a second.++ 'sources.item.create': { minute: 60, hour: 1200 },++ // A removal (takedown, licence, error) hides the item everywhere and emits an event. A publisher's++ // takedown may cover many items, so two a second.++ 'sources.item.remove': { minute: 120, hour: 3000 },++};++++/**++ * limits(name, own) middleware, plus limits.reads(name) (the defaults on a counted read) and++ * limits.budget(name) (one of BUDGETS).++ */++function createActorLimits({ config, now = () => Date.now(), registry = null, log = console }) {++ const refused = registry++ ? registry.counter({ name: 'sources_rate_limited_total', help: 'Requests refused 429 by a per-actor limit, by limit name and window', labelNames: ['limit', 'window'] })++ : null;++ const limiter = createActorLimiter({++ limits: { minute: config.limits.minute, hour: config.limits.hour },++ actor,++ now,++ onLimited(e) {++ // The actor is a principal (or an address), never a token.++ log.warn(`[limits] ${e.name}: ${e.actor} refused, over ${e.limit} per ${e.window}`);++ if (refused) refused.inc({ limit: e.name, window: e.window });++ },++ });++ limiter.reads = (name) => {++ const limit = limiter(name);++ return function actorReadLimit(req, res, next) { return countedRead(req) ? limit(req, res, next) : next(); };++ };++ const budgets = new Map(Object.entries(BUDGETS).map(([name, own]) => [name, limiter(name, own)]));++ limiter.budget = (name) => {++ const m = budgets.get(name);++ if (!m) throw new Error(`limits: no budget named ${name}`);++ return m;++ };++ return limiter;++}++++module.exports = { createActorLimits, actor, countedRead, BUDGETS };@@@ -47,10 +47,12 @@ function manualItem(body) {};}--function itemsRouter({ db, registry, items, auth, relay, now }) {++function itemsRouter({ db, registry, items, auth, relay, now, limits }) {const router = express.Router();-- const read = auth.requireCap(CAPS.items);-- const manage = auth.requireCap(CAPS.manage);++ // Each guard, then the principal's per-actor limit (api/actor-limits.js), before any work.++ const read = [auth.requireCap(CAPS.items), limits.reads('sources.read')];++ const create = [auth.requireCap(CAPS.manage), limits.budget('sources.item.create')];++ const remove = [auth.requireCap(CAPS.manage), limits.budget('sources.item.remove')];router.get('/api/v1/items', read, (req, res) => {const source = req.query.source ? String(req.query.source) : null;@@@ -79,7 +81,7 @@ function itemsRouter({ db, registry, items, auth, relay, now }) {res.json({ item, source: h ? { key: item.source_key, status: h.status, stale: h.stale, last_success_at: h.last_success_at } : null });});-- router.post('/api/v1/sources/:key/items', manage, (req, res) => {++ router.post('/api/v1/sources/:key/items', create, (req, res) => {const row = registry.get(req.params.key);if (!row) return http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such source', ctx: req.ov });if (row.type !== 'manual') return http.sendProblem(res, 409, 'sources.not_manual', { detail: 'items of fetched sources come from their fetches only', ctx: req.ov });@@@ -96,7 +98,7 @@ function itemsRouter({ db, registry, items, auth, relay, now }) {res.status(counts.created ? 201 : 200).json({ outcome, item: items.get(saved.id) });});-- router.delete('/api/v1/items/:id', manage, (req, res) => {++ router.delete('/api/v1/items/:id', remove, (req, res) => {const reason = toText(req.body && req.body.reason, 300);if (!reason) return http.sendProblem(res, 422, 'sources.reason_required', { detail: 'a removal needs a reason (takedown, licence, error…)', ctx: req.ov });const found = items.get(String(req.params.id));@@@ -32,10 +32,12 @@ function runView(r) {};}--function sourcesRouter({ db, registry, ingest, auth }) {++function sourcesRouter({ db, registry, ingest, auth, limits }) {const router = express.Router();-- const read = auth.requireCap(CAPS.read);-- const manage = auth.requireCap(CAPS.manage);++ // Each guard, then the principal's per-actor limit (api/actor-limits.js), before any work.++ const read = [auth.requireCap(CAPS.read), limits.reads('sources.read')];++ const manage = [auth.requireCap(CAPS.manage), limits.budget('sources.source.manage')];++ const fetchNow = [auth.requireCap(CAPS.manage), limits.budget('sources.source.fetch')];const fail = (res, req, err) => {if (err instanceof RegistryError) {@@@ -112,7 +114,7 @@ function sourcesRouter({ db, registry, ingest, auth }) {} catch (err) { fail(res, req, err); }});-- router.post('/api/v1/sources/:key/fetch', manage, async (req, res, next) => {++ router.post('/api/v1/sources/:key/fetch', fetchNow, async (req, res, next) => {try {const out = await ingest.run(req.params.key, { trigger: 'manual' });if (!out) return notFound(res, req);@@@ -8,6 +8,7 @@ const { createRelease } = require('openvibe-shared/release');const { createSourcesReadiness, registerSourcesGauges } = require('./observability');const { sourcesRouter } = require('./api/sources');const { itemsRouter } = require('./api/items');++const { createActorLimits } = require('./api/actor-limits');const pkg = require('../package.json');const TEXT_INDEX = [@@@ -62,7 +63,7 @@ Source code: <a href="https://github.com/OpenVibers/OpenVibe.Sources">OpenVibers</html>`;--function createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log = console }) {++function createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log = console, limitsNow = null }) {const app = express();app.disable('x-powered-by');app.set('trust proxy', 'loopback');@@@ -92,8 +93,11 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, keys,// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).release.mount(app, { registry: metrics.registry });-- app.use(sourcesRouter({ db, registry, ingest, auth }));-- app.use(itemsRouter({ db, registry, items, auth, relay, now }));++ // Per-actor limits (api/actor-limits.js), counted by the principal once a route's capability guard++ // passed. limitsNow: the limiter's clock (tests; default the wall clock, not the ingest clock).++ const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log });++ app.use(sourcesRouter({ db, registry, ingest, auth, limits }));++ app.use(itemsRouter({ db, registry, items, auth, relay, now, limits }));// The public host (sources.openvibe.network) shows only this page, health, readiness and// /release.json: Sources is internal. Browsers get a short honest HTML page, other clients the@@@ -33,6 +33,12 @@ function load(env = process.env) {issuer: strip(env.OV_NETWORK_ISSUER || env.OV_NETWORK_URL || 'https://openvibe.network'),networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,audience: 'openvibe.sources',++ // Per-actor limits (server/api/actor-limits.js, roadmap WS-R task 4): the API reads one app or++ // module may make per minute and per hour. Writes set their own numbers there.++ limits: {++ minute: Math.max(1, int(env.SOURCES_LIMITS_MINUTE, 120)),++ hour: Math.max(1, int(env.SOURCES_LIMITS_HOUR, 3000)),++ },oauth: {clientId: env.OV_OAUTH_CLIENT_ID || 'sources',clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',@@@ -21,7 +21,7 @@ const { createOutbox, createRelay } = require('./events/outbox');const { createKeyStore, createAuth } = require('./auth');const { createApp } = require('./app');--async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true } = {}) {++async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true, limitsNow = null } = {}) {config = config || load();const db = openDb(config.dbPath);const outbox = createOutbox(db, { source: config.serviceId, now });@@@ -43,7 +43,7 @@ async function start({ config, now = () => Date.now(), fetchImpl = globalThis.feconst scheduler = createScheduler({ db, ingest, config, now, log });const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });const auth = createAuth({ config, keys });-- const app = createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log });++ const app = createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log, limitsNow });const keyLoaded = keys.start().catch(() => null);relay.start();nnew file mode 100644@@@ -0,0 +1,107 @@++'use strict';++/**++ * Per-actor rate limits (server/api/actor-limits.js, roadmap WS-R task 4): past its limit one principal++ * gets 429 problem+json `rate_limited` with Retry-After, before the route does any work, while another++ * still passes; the window reopens on the clock. A first-party service is not counted on reads (it++ * pulls for all its readers and editors); callers without a token get 401, never 429. Writes have their++ * own budgets. Health, ready, release.json and metrics are never limited; refusals are logged (no++ * token) and counted.++ */++const assert = require('assert');++const crypto = require('crypto');++const { serviceAuth } = require('openvibe-contracts');++const { boot, request, serviceToken, sourceDef, suite, ISSUER, privateKey, silent } = require('./helpers');++++const t = suite('actor-limits');++// The limiter's clock: 15 s into a minute, so the minute window has 45 s left. Reads: 3 a minute.++let clock = Date.UTC(2026, 8, 27, 12, 0, 15);++const lines = [];++let svc;++const api = (method, p, opts = {}) => request(svc.base, method, p, opts);++++function appToken(id, cap) {++ const now = Math.floor(Date.now() / 1000);++ return serviceAuth.signServiceToken({++ iss: ISSUER, sub: `app:app_${id}`, actor_type: 'app', aud: ['openvibe.sources'], cap, ns: [], iat: now, exp: now + 300,++ jti: `tok_${crypto.randomBytes(8).toString('hex')}`, project_id: 'prj_01J8ZQ4Y7N3M2K1H0G9F8E7D6C', env: 'production',++ }, privateKey);++}++const APP = appToken('01J8ZQ4Y7N3M2K1H0G9F8E7D6C', ['sources.source.read', 'sources.item.read']);++const OTHER_APP = appToken('01J8ZQ4Y7N3M2K1H0G9F8E7D6D', ['sources.source.read']);++const NEWS = serviceToken('news', ['sources.source.read', 'sources.item.read']);++const ADMIN = serviceToken('network', ['sources.source.manage']);++const STAFF_TOOL = serviceToken('admin-tools', ['sources.source.manage']);++++t('boot', async () => {++ svc = await boot({ env: { SOURCES_LIMITS_MINUTE: '3', SOURCES_LIMITS_HOUR: '100' }, limitsNow: () => clock, log: { ...silent, warn: (m) => lines.push(String(m)) } });++});++++t('an app reads 3 a minute, then 429 rate_limited with Retry-After; another app passes', async () => {++ for (let i = 0; i < 3; i++) assert.strictEqual((await api('GET', '/api/v1/sources', { token: APP })).status, 200, `read ${i + 1}`);++ const r = await api('GET', '/api/v1/items', { token: APP });++ assert.strictEqual(r.status, 429, r.text);++ assert.strictEqual(r.headers.get('retry-after'), '45');++ assert.match(r.headers.get('content-type'), /^application\/problem\+json/);++ assert.deepStrictEqual([r.body.code, r.body.status, r.body.retry_after_seconds], ['rate_limited', 429, 45]);++ assert.ok(r.body.detail.includes('sources.read'), r.body.detail);++ assert.strictEqual((await api('GET', '/api/v1/sources', { token: OTHER_APP })).status, 200, 'another app still passes');++});++++t('a first-party service is not counted on reads; no token is 401, never 429', async () => {++ for (let i = 0; i < 8; i++) {++ assert.strictEqual((await api('GET', '/api/v1/items?category=news', { token: NEWS })).status, 200, `service read ${i + 1}`);++ assert.strictEqual((await api('GET', '/api/v1/sources')).status, 401);++ }++});++++t('the next minute opens the window again', async () => {++ clock += 45 * 1000;++ assert.strictEqual((await api('GET', '/api/v1/sources', { token: APP })).status, 200);++});++++t('source changes: 30 a minute per principal; nothing is stored past it; another staff tool passes', async () => {++ clock = Date.UTC(2026, 8, 27, 12, 5, 0);++ const count = () => svc.db.prepare('SELECT COUNT(*) AS n FROM sources').get().n;++ for (let i = 0; i < 30; i++) {++ const r = await api('POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: `limits-${i}`, enabled: false, endpoints: ['https://example.org/feed'] }) });++ assert.strictEqual(r.status, 201, `source ${i + 1}: ${r.text}`);++ }++ const before = count();++ const r = await api('POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: 'one-too-many', enabled: false, endpoints: ['https://example.org/feed'] }) });++ assert.deepStrictEqual([r.status, r.body.code, r.headers.get('retry-after')], [429, 'rate_limited', '60']);++ assert.strictEqual(count(), before, 'nothing stored');++ const other = await api('POST', '/api/v1/sources', { token: STAFF_TOOL, body: sourceDef({ key: 'other-tool', enabled: false, endpoints: ['https://example.org/feed'] }) });++ assert.strictEqual(other.status, 201, other.text);++});++++t('fetch now has its own, tighter budget (10 a minute)', async () => {++ await api('POST', '/api/v1/sources', { token: STAFF_TOOL, body: { key: 'manual-codes', name: 'Codes', type: 'manual', category: 'coupons', terms_note: 'test', enabled: true } });++ for (let i = 0; i < 10; i++) assert.strictEqual((await api('POST', '/api/v1/sources/manual-codes/fetch', { token: ADMIN })).status, 409, `fetch ${i + 1}`);++ const r = await api('POST', '/api/v1/sources/manual-codes/fetch', { token: ADMIN });++ assert.deepStrictEqual([r.status, r.body.code], [429, 'rate_limited']);++});++++t('health, ready, release.json, metrics and the home page are never limited', async () => {++ for (let i = 0; i < 6; i++) {++ assert.strictEqual((await api('GET', '/api/health')).status, 200);++ assert.notStrictEqual((await api('GET', '/api/ready')).status, 429);++ assert.strictEqual((await api('GET', '/release.json')).status, 200);++ assert.strictEqual((await api('GET', '/metrics')).status, 200);++ assert.strictEqual((await api('GET', '/')).status, 200);++ }++});++++t('refusals are logged (the principal, never a token) and counted in sources_rate_limited_total', async () => {++ assert.ok(lines.includes('[limits] sources.read: app:app_01J8ZQ4Y7N3M2K1H0G9F8E7D6C refused, over 3 per minute'), lines.join('\n'));++ assert.ok(lines.includes('[limits] sources.source.manage: svc:network refused, over 30 per minute'), lines.join('\n'));++ assert.ok(!lines.some((l) => /Bearer|eyJ/.test(l)), 'no token in the log');++ const m = (await api('GET', '/metrics')).text;++ const found = m.split('\n').filter((l) => l.includes('sources_rate_limited_total')).join('\n');++ assert.ok(/sources_rate_limited_total\{limit="sources.read",window="minute"\} 1/.test(m), found);++ assert.ok(/sources_rate_limited_total\{limit="sources.source.manage",window="minute"\} 1/.test(m), found);++ assert.ok(/sources_rate_limited_total\{limit="sources.source.fetch",window="minute"\} 1/.test(m), found);++});++++t('stop', async () => { await svc.stop(); });++++t.run();@@@ -39,8 +39,11 @@ function tmpDir() {return d;}--/** Boot Sources; the scheduler is off unless worker: 'on'. Loopback is allowlisted for the stubs. */--async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now } = {}) {++/**++ * Boot Sources; the scheduler is off unless worker: 'on'. Loopback is allowlisted for the stubs.++ * log: Sources' logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).++ */++async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now, log = silent, limitsNow = null } = {}) {const dir = tmpDir();const config = load({NODE_ENV: 'test',@@@ -54,7 +57,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now } =SOURCES_TICK_MS: '50',...env,});-- const h = await start({ config, log: silent, lookupImpl, tokenClient, ...(now ? { now } : {}) });++ const h = await start({ config, log, lookupImpl, tokenClient, limitsNow, ...(now ? { now } : {}) });const base = `http://127.0.0.1:${h.server.address().port}`;return { ...h, base, dir, async stop() { await h.close(); } };}
@@ -17,6 +17,12 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
@@ -17,6 +17,12 @@ OV_NETWORK_INTERNAL_URL=http://127.0.0.1:4000
#OV_NETWORK_ISSUER=https://openvibe.network
#OV_NETWORK_ISSUER=https://openvibe.network
#OV_NETWORK_PUBLIC_KEY=
#OV_NETWORK_PUBLIC_KEY=
+# Per-actor limits (server/api/actor-limits.js): API reads one app or module (app:…, mod:…) may make per
+# minute and per hour; past them 429 rate_limited with Retry-After. First-party services (svc:…) are not
+# counted on reads; writes (sources.source.manage) have their own numbers (README).
+#SOURCES_LIMITS_MINUTE=120
+#SOURCES_LIMITS_HOUR=3000
+
# This service's own principal (svc:sources) — publishes sources.* events (events.event.publish).
# This service's own principal (svc:sources) — publishes sources.* events (events.event.publish).
OV_OAUTH_CLIENT_ID=sources
OV_OAUTH_CLIENT_ID=sources
OV_OAUTH_CLIENT_SECRET=
OV_OAUTH_CLIENT_SECRET=
@@ -129,6 +129,26 @@ it contains. The capability ids (first proposed in [docs/capabilities-proposal/]
@@ -129,6 +129,26 @@ it contains. The capability ids (first proposed in [docs/capabilities-proposal/]
are released in `openvibe-contracts` v0.12.0 (this repo pins v0.33.0); [server/auth.js](server/auth.js)
are released in `openvibe-contracts` v0.12.0 (this repo pins v0.33.0); [server/auth.js](server/auth.js)
decides them with the contracts grant rule.
decides them with the contracts grant rule.
+### Per-actor limits
+
+Every API route also limits the principal that passed its capability guard, before any work:
+`server/api/actor-limits.js`, openvibe-sdk/limits, roadmap WS-R task 4. Past a limit: `429`
+problem+json `rate_limited` with `Retry-After`, one `[limits]` log line and
+`sources_rate_limited_total{limit,window}`.
+
+| Routes | Per principal, a minute / an hour |
+|---|---|
+| Reads (`sources.source.read`, `sources.item.read`) by an app or module (`app:…`, `mod:…`) | `SOURCES_LIMITS_MINUTE` / `SOURCES_LIMITS_HOUR` (120 / 3000) |
+| `POST`, `PATCH`, `DELETE /api/v1/sources[/:key]` | 30 / 300 |
+| `POST /api/v1/sources/:key/fetch` | 10 / 100 |
+| `POST /api/v1/sources/:key/items` (manual) | 60 / 1200 |
+| `DELETE /api/v1/items/:id` (a takedown may cover many items) | 120 / 3000 |
+
+A first-party service (`svc:…`) is not counted on reads: News, Reviews and Trade pull cursor pages
+for all their readers, and Wiki asks about every item its editors cite. Sources' own workers (the
+scheduler, fetches, the outbox relay) run in-process and never pass through HTTP. Never limited:
+`/api/health`, `/api/ready`, `/release.json`, `/metrics` and the home page. `test/actor-limits.test.js`.
+
## Events (transactional outbox → OpenVibe.Events when `EVENTS_URL` is set)
## Events (transactional outbox → OpenVibe.Events when `EVENTS_URL` is set)
- `sources.item.created`, `sources.item.updated` (with the previous content hash), `sources.item.removed` (with the reason)
- `sources.item.created`, `sources.item.updated` (with the previous content hash), `sources.item.removed` (with the reason)
@@ -14,6 +14,7 @@
@@ -14,6 +14,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
},
},
"devDependencies": {
"devDependencies": {
@@ -1005,6 +1006,15 @@
@@ -1005,6 +1006,15 @@
"node": ">=20"
"node": ">=20"
}
}
},
},
+ "node_modules/openvibe-sdk": {
+ "version": "0.12.0",
+ "resolved": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0",
+ "integrity": "sha512-iG1h905O1f7y/PnP8xLA6GCVzXEBDfSaBj2tc1ucNreWg2sT/AJWosb8NW0WRMK5WS4sIQ6tMsprmDLj/fjjeQ==",
+ "license": "MIT",
+ "engines": {
+ "node": ">=20"
+ }
+ },
"node_modules/openvibe-shared": {
"node_modules/openvibe-shared": {
"version": "1.22.0",
"version": "1.22.0",
"resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
"resolved": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0",
@@ -28,6 +28,7 @@
@@ -28,6 +28,7 @@
"express": "^4.21.2",
"express": "^4.21.2",
"fast-xml-parser": "^4.5.3",
"fast-xml-parser": "^4.5.3",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
"openvibe-contracts": "https://codeload.github.com/OpenVibers/OpenVibe.Contracts/tar.gz/refs/tags/v0.49.0",
+ "openvibe-sdk": "https://codeload.github.com/OpenVibers/OpenVibe.SDK/tar.gz/refs/tags/v0.12.0",
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
"openvibe-shared": "https://codeload.github.com/OpenVibers/OpenVibe.Shared/tar.gz/refs/tags/v1.22.0"
},
},
"engines": {
"engines": {
new file mode 100644
new file mode 100644
@@ -0,0 +1,88 @@
@@ -0,0 +1,88 @@
+'use strict';
+/**
+ * Per-actor rate limits on /api/v1 (roadmap WS-R task 4; openvibe-sdk/limits).
+ *
+ * Every API route takes a service token and one capability (auth.js); the limits count requests by
+ * the principal that passed that guard (req.principal.sub: svc:news, app:app_…), before the route does
+ * any work. Sources' own workers (the scheduler, fetches, the outbox relay) run in this process and
+ * never go through HTTP, so nothing here can slow them.
+ *
+ * Reads (sources.source.read, sources.item.read) take SOURCES_LIMITS_MINUTE / SOURCES_LIMITS_HOUR,
+ * 120 and 3000, for an app or module (app:…, mod:…). A first-party service (svc:…) is not counted on
+ * reads: News, Reviews and Trade pull cursor pages for all their readers, and Wiki asks about every
+ * item its editors cite, so one budget for the whole service would refuse real people's work. The API
+ * answers only on the production host (nginx serves the public host nothing but health, ready and
+ * release.json).
+ *
+ * Writes (sources.source.manage, staff tooling) are counted for every principal, each with its own
+ * budget below.
+ *
+ * Past a limit the route answers 429 problem+json `rate_limited` with Retry-After; the refusal is logged
+ * once and counted in sources_rate_limited_total{limit,window}. Counters live in this process: a
+ * restart forgets them. Never limited: /api/health, /api/ready, /release.json, /metrics and the home
+ * page.
+ */
+const { createActorLimiter, defaultActor } = require('openvibe-sdk/limits');
+
+const FIRST_PARTY = /^svc:/;
+
+function actor(req) {
+ const p = req.principal;
+ if (p && typeof p.sub === 'string' && p.sub) return p.sub;
+ return defaultActor(req);
+}
+
+/** Counted on reads: anything but a first-party service (an app or module acting as itself). */
+function countedRead(req) {
+ const p = req.principal;
+ return !(p && FIRST_PARTY.test(String(p.sub)));
+}
+
+/** The writes, each with its numbers per principal (a minute, an hour). */
+const BUDGETS = {
+ // Creating, changing or deleting a source is a staff decision recorded with its terms and robots
+ // notes: a form every few seconds at most.
+ 'sources.source.manage': { minute: 30, hour: 300 },
+ // A manual fetch requests every endpoint of the source now (the per-host spacing still applies):
+ // the scheduler already polls each source, so a person's "fetch now" is a rare nudge.
+ 'sources.source.fetch': { minute: 10, hour: 100 },
+ // A manual item (a code or record a publisher stated, entered with its evidence URL) is stored,
+ // revisioned and announced to subscribers: one a second.
+ 'sources.item.create': { minute: 60, hour: 1200 },
+ // A removal (takedown, licence, error) hides the item everywhere and emits an event. A publisher's
+ // takedown may cover many items, so two a second.
+ 'sources.item.remove': { minute: 120, hour: 3000 },
+};
+
+/**
+ * limits(name, own) middleware, plus limits.reads(name) (the defaults on a counted read) and
+ * limits.budget(name) (one of BUDGETS).
+ */
+function createActorLimits({ config, now = () => Date.now(), registry = null, log = console }) {
+ const refused = registry
+ ? registry.counter({ name: 'sources_rate_limited_total', help: 'Requests refused 429 by a per-actor limit, by limit name and window', labelNames: ['limit', 'window'] })
+ : null;
+ const limiter = createActorLimiter({
+ limits: { minute: config.limits.minute, hour: config.limits.hour },
+ actor,
+ now,
+ onLimited(e) {
+ // The actor is a principal (or an address), never a token.
+ log.warn(`[limits] ${e.name}: ${e.actor} refused, over ${e.limit} per ${e.window}`);
+ if (refused) refused.inc({ limit: e.name, window: e.window });
+ },
+ });
+ limiter.reads = (name) => {
+ const limit = limiter(name);
+ return function actorReadLimit(req, res, next) { return countedRead(req) ? limit(req, res, next) : next(); };
+ };
+ const budgets = new Map(Object.entries(BUDGETS).map(([name, own]) => [name, limiter(name, own)]));
+ limiter.budget = (name) => {
+ const m = budgets.get(name);
+ if (!m) throw new Error(`limits: no budget named ${name}`);
+ return m;
+ };
+ return limiter;
+}
+
+module.exports = { createActorLimits, actor, countedRead, BUDGETS };
@@ -47,10 +47,12 @@ function manualItem(body) {
@@ -47,10 +47,12 @@ function manualItem(body) {
};
};
}
}
-function itemsRouter({ db, registry, items, auth, relay, now }) {
+function itemsRouter({ db, registry, items, auth, relay, now, limits }) {
const router = express.Router();
const router = express.Router();
- const read = auth.requireCap(CAPS.items);
- const manage = auth.requireCap(CAPS.manage);
+ // Each guard, then the principal's per-actor limit (api/actor-limits.js), before any work.
+ const read = [auth.requireCap(CAPS.items), limits.reads('sources.read')];
+ const create = [auth.requireCap(CAPS.manage), limits.budget('sources.item.create')];
+ const remove = [auth.requireCap(CAPS.manage), limits.budget('sources.item.remove')];
router.get('/api/v1/items', read, (req, res) => {
router.get('/api/v1/items', read, (req, res) => {
const source = req.query.source ? String(req.query.source) : null;
const source = req.query.source ? String(req.query.source) : null;
@@ -79,7 +81,7 @@ function itemsRouter({ db, registry, items, auth, relay, now }) {
@@ -79,7 +81,7 @@ function itemsRouter({ db, registry, items, auth, relay, now }) {
res.json({ item, source: h ? { key: item.source_key, status: h.status, stale: h.stale, last_success_at: h.last_success_at } : null });
res.json({ item, source: h ? { key: item.source_key, status: h.status, stale: h.stale, last_success_at: h.last_success_at } : null });
});
});
- router.post('/api/v1/sources/:key/items', manage, (req, res) => {
+ router.post('/api/v1/sources/:key/items', create, (req, res) => {
const row = registry.get(req.params.key);
const row = registry.get(req.params.key);
if (!row) return http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such source', ctx: req.ov });
if (!row) return http.sendProblem(res, 404, 'sources.not_found', { detail: 'no such source', ctx: req.ov });
if (row.type !== 'manual') return http.sendProblem(res, 409, 'sources.not_manual', { detail: 'items of fetched sources come from their fetches only', ctx: req.ov });
if (row.type !== 'manual') return http.sendProblem(res, 409, 'sources.not_manual', { detail: 'items of fetched sources come from their fetches only', ctx: req.ov });
@@ -96,7 +98,7 @@ function itemsRouter({ db, registry, items, auth, relay, now }) {
@@ -96,7 +98,7 @@ function itemsRouter({ db, registry, items, auth, relay, now }) {
res.status(counts.created ? 201 : 200).json({ outcome, item: items.get(saved.id) });
res.status(counts.created ? 201 : 200).json({ outcome, item: items.get(saved.id) });
});
});
- router.delete('/api/v1/items/:id', manage, (req, res) => {
+ router.delete('/api/v1/items/:id', remove, (req, res) => {
const reason = toText(req.body && req.body.reason, 300);
const reason = toText(req.body && req.body.reason, 300);
if (!reason) return http.sendProblem(res, 422, 'sources.reason_required', { detail: 'a removal needs a reason (takedown, licence, error…)', ctx: req.ov });
if (!reason) return http.sendProblem(res, 422, 'sources.reason_required', { detail: 'a removal needs a reason (takedown, licence, error…)', ctx: req.ov });
const found = items.get(String(req.params.id));
const found = items.get(String(req.params.id));
@@ -32,10 +32,12 @@ function runView(r) {
@@ -32,10 +32,12 @@ function runView(r) {
};
};
}
}
-function sourcesRouter({ db, registry, ingest, auth }) {
+function sourcesRouter({ db, registry, ingest, auth, limits }) {
const router = express.Router();
const router = express.Router();
- const read = auth.requireCap(CAPS.read);
- const manage = auth.requireCap(CAPS.manage);
+ // Each guard, then the principal's per-actor limit (api/actor-limits.js), before any work.
+ const read = [auth.requireCap(CAPS.read), limits.reads('sources.read')];
+ const manage = [auth.requireCap(CAPS.manage), limits.budget('sources.source.manage')];
+ const fetchNow = [auth.requireCap(CAPS.manage), limits.budget('sources.source.fetch')];
const fail = (res, req, err) => {
const fail = (res, req, err) => {
if (err instanceof RegistryError) {
if (err instanceof RegistryError) {
@@ -112,7 +114,7 @@ function sourcesRouter({ db, registry, ingest, auth }) {
@@ -112,7 +114,7 @@ function sourcesRouter({ db, registry, ingest, auth }) {
} catch (err) { fail(res, req, err); }
} catch (err) { fail(res, req, err); }
});
});
- router.post('/api/v1/sources/:key/fetch', manage, async (req, res, next) => {
+ router.post('/api/v1/sources/:key/fetch', fetchNow, async (req, res, next) => {
try {
try {
const out = await ingest.run(req.params.key, { trigger: 'manual' });
const out = await ingest.run(req.params.key, { trigger: 'manual' });
if (!out) return notFound(res, req);
if (!out) return notFound(res, req);
@@ -8,6 +8,7 @@ const { createRelease } = require('openvibe-shared/release');
@@ -8,6 +8,7 @@ const { createRelease } = require('openvibe-shared/release');
const { createSourcesReadiness, registerSourcesGauges } = require('./observability');
const { createSourcesReadiness, registerSourcesGauges } = require('./observability');
const { sourcesRouter } = require('./api/sources');
const { sourcesRouter } = require('./api/sources');
const { itemsRouter } = require('./api/items');
const { itemsRouter } = require('./api/items');
+const { createActorLimits } = require('./api/actor-limits');
const pkg = require('../package.json');
const pkg = require('../package.json');
const TEXT_INDEX = [
const TEXT_INDEX = [
@@ -62,7 +63,7 @@ Source code: <a href="https://github.com/OpenVibers/OpenVibe.Sources">OpenVibers
@@ -62,7 +63,7 @@ Source code: <a href="https://github.com/OpenVibers/OpenVibe.Sources">OpenVibers
</html>
</html>
`;
`;
-function createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log = console }) {
+function createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log = console, limitsNow = null }) {
const app = express();
const app = express();
app.disable('x-powered-by');
app.disable('x-powered-by');
app.set('trust proxy', 'loopback');
app.set('trust proxy', 'loopback');
@@ -92,8 +93,11 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, keys,
@@ -92,8 +93,11 @@ function createApp({ config, db, registry, items, ingest, scheduler, auth, keys,
// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
// GET /release.json (ADR-016) and POST /release-metrics (open tabs' update reports into /metrics).
release.mount(app, { registry: metrics.registry });
release.mount(app, { registry: metrics.registry });
- app.use(sourcesRouter({ db, registry, ingest, auth }));
- app.use(itemsRouter({ db, registry, items, auth, relay, now }));
+ // Per-actor limits (api/actor-limits.js), counted by the principal once a route's capability guard
+ // passed. limitsNow: the limiter's clock (tests; default the wall clock, not the ingest clock).
+ const limits = createActorLimits({ config, now: limitsNow || (() => Date.now()), registry: metrics.registry, log });
+ app.use(sourcesRouter({ db, registry, ingest, auth, limits }));
+ app.use(itemsRouter({ db, registry, items, auth, relay, now, limits }));
// The public host (sources.openvibe.network) shows only this page, health, readiness and
// The public host (sources.openvibe.network) shows only this page, health, readiness and
// /release.json: Sources is internal. Browsers get a short honest HTML page, other clients the
// /release.json: Sources is internal. Browsers get a short honest HTML page, other clients the
@@ -33,6 +33,12 @@ function load(env = process.env) {
@@ -33,6 +33,12 @@ function load(env = process.env) {
issuer: strip(env.OV_NETWORK_ISSUER || env.OV_NETWORK_URL || 'https://openvibe.network'),
issuer: strip(env.OV_NETWORK_ISSUER || env.OV_NETWORK_URL || 'https://openvibe.network'),
networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,
networkPublicKey: env.OV_NETWORK_PUBLIC_KEY ? env.OV_NETWORK_PUBLIC_KEY.replace(/\\n/g, '\n') : null,
audience: 'openvibe.sources',
audience: 'openvibe.sources',
+ // Per-actor limits (server/api/actor-limits.js, roadmap WS-R task 4): the API reads one app or
+ // module may make per minute and per hour. Writes set their own numbers there.
+ limits: {
+ minute: Math.max(1, int(env.SOURCES_LIMITS_MINUTE, 120)),
+ hour: Math.max(1, int(env.SOURCES_LIMITS_HOUR, 3000)),
+ },
oauth: {
oauth: {
clientId: env.OV_OAUTH_CLIENT_ID || 'sources',
clientId: env.OV_OAUTH_CLIENT_ID || 'sources',
clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',
clientSecret: env.OV_OAUTH_CLIENT_SECRET || '',
@@ -21,7 +21,7 @@ const { createOutbox, createRelay } = require('./events/outbox');
@@ -21,7 +21,7 @@ const { createOutbox, createRelay } = require('./events/outbox');
const { createKeyStore, createAuth } = require('./auth');
const { createKeyStore, createAuth } = require('./auth');
const { createApp } = require('./app');
const { createApp } = require('./app');
-async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true } = {}) {
+async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fetch, tokenClient, lookupImpl, log = console, listen = true, limitsNow = null } = {}) {
config = config || load();
config = config || load();
const db = openDb(config.dbPath);
const db = openDb(config.dbPath);
const outbox = createOutbox(db, { source: config.serviceId, now });
const outbox = createOutbox(db, { source: config.serviceId, now });
@@ -43,7 +43,7 @@ async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fe
@@ -43,7 +43,7 @@ async function start({ config, now = () => Date.now(), fetchImpl = globalThis.fe
const scheduler = createScheduler({ db, ingest, config, now, log });
const scheduler = createScheduler({ db, ingest, config, now, log });
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const keys = createKeyStore({ urls: [config.networkInternalUrl, config.networkUrl], pem: config.networkPublicKey, fetchImpl, log });
const auth = createAuth({ config, keys });
const auth = createAuth({ config, keys });
- const app = createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log });
+ const app = createApp({ config, db, registry, items, ingest, scheduler, auth, keys, outbox, relay, now, log, limitsNow });
const keyLoaded = keys.start().catch(() => null);
const keyLoaded = keys.start().catch(() => null);
relay.start();
relay.start();
new file mode 100644
new file mode 100644
@@ -0,0 +1,107 @@
@@ -0,0 +1,107 @@
+'use strict';
+/**
+ * Per-actor rate limits (server/api/actor-limits.js, roadmap WS-R task 4): past its limit one principal
+ * gets 429 problem+json `rate_limited` with Retry-After, before the route does any work, while another
+ * still passes; the window reopens on the clock. A first-party service is not counted on reads (it
+ * pulls for all its readers and editors); callers without a token get 401, never 429. Writes have their
+ * own budgets. Health, ready, release.json and metrics are never limited; refusals are logged (no
+ * token) and counted.
+ */
+const assert = require('assert');
+const crypto = require('crypto');
+const { serviceAuth } = require('openvibe-contracts');
+const { boot, request, serviceToken, sourceDef, suite, ISSUER, privateKey, silent } = require('./helpers');
+
+const t = suite('actor-limits');
+// The limiter's clock: 15 s into a minute, so the minute window has 45 s left. Reads: 3 a minute.
+let clock = Date.UTC(2026, 8, 27, 12, 0, 15);
+const lines = [];
+let svc;
+const api = (method, p, opts = {}) => request(svc.base, method, p, opts);
+
+function appToken(id, cap) {
+ const now = Math.floor(Date.now() / 1000);
+ return serviceAuth.signServiceToken({
+ iss: ISSUER, sub: `app:app_${id}`, actor_type: 'app', aud: ['openvibe.sources'], cap, ns: [], iat: now, exp: now + 300,
+ jti: `tok_${crypto.randomBytes(8).toString('hex')}`, project_id: 'prj_01J8ZQ4Y7N3M2K1H0G9F8E7D6C', env: 'production',
+ }, privateKey);
+}
+const APP = appToken('01J8ZQ4Y7N3M2K1H0G9F8E7D6C', ['sources.source.read', 'sources.item.read']);
+const OTHER_APP = appToken('01J8ZQ4Y7N3M2K1H0G9F8E7D6D', ['sources.source.read']);
+const NEWS = serviceToken('news', ['sources.source.read', 'sources.item.read']);
+const ADMIN = serviceToken('network', ['sources.source.manage']);
+const STAFF_TOOL = serviceToken('admin-tools', ['sources.source.manage']);
+
+t('boot', async () => {
+ svc = await boot({ env: { SOURCES_LIMITS_MINUTE: '3', SOURCES_LIMITS_HOUR: '100' }, limitsNow: () => clock, log: { ...silent, warn: (m) => lines.push(String(m)) } });
+});
+
+t('an app reads 3 a minute, then 429 rate_limited with Retry-After; another app passes', async () => {
+ for (let i = 0; i < 3; i++) assert.strictEqual((await api('GET', '/api/v1/sources', { token: APP })).status, 200, `read ${i + 1}`);
+ const r = await api('GET', '/api/v1/items', { token: APP });
+ assert.strictEqual(r.status, 429, r.text);
+ assert.strictEqual(r.headers.get('retry-after'), '45');
+ assert.match(r.headers.get('content-type'), /^application\/problem\+json/);
+ assert.deepStrictEqual([r.body.code, r.body.status, r.body.retry_after_seconds], ['rate_limited', 429, 45]);
+ assert.ok(r.body.detail.includes('sources.read'), r.body.detail);
+ assert.strictEqual((await api('GET', '/api/v1/sources', { token: OTHER_APP })).status, 200, 'another app still passes');
+});
+
+t('a first-party service is not counted on reads; no token is 401, never 429', async () => {
+ for (let i = 0; i < 8; i++) {
+ assert.strictEqual((await api('GET', '/api/v1/items?category=news', { token: NEWS })).status, 200, `service read ${i + 1}`);
+ assert.strictEqual((await api('GET', '/api/v1/sources')).status, 401);
+ }
+});
+
+t('the next minute opens the window again', async () => {
+ clock += 45 * 1000;
+ assert.strictEqual((await api('GET', '/api/v1/sources', { token: APP })).status, 200);
+});
+
+t('source changes: 30 a minute per principal; nothing is stored past it; another staff tool passes', async () => {
+ clock = Date.UTC(2026, 8, 27, 12, 5, 0);
+ const count = () => svc.db.prepare('SELECT COUNT(*) AS n FROM sources').get().n;
+ for (let i = 0; i < 30; i++) {
+ const r = await api('POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: `limits-${i}`, enabled: false, endpoints: ['https://example.org/feed'] }) });
+ assert.strictEqual(r.status, 201, `source ${i + 1}: ${r.text}`);
+ }
+ const before = count();
+ const r = await api('POST', '/api/v1/sources', { token: ADMIN, body: sourceDef({ key: 'one-too-many', enabled: false, endpoints: ['https://example.org/feed'] }) });
+ assert.deepStrictEqual([r.status, r.body.code, r.headers.get('retry-after')], [429, 'rate_limited', '60']);
+ assert.strictEqual(count(), before, 'nothing stored');
+ const other = await api('POST', '/api/v1/sources', { token: STAFF_TOOL, body: sourceDef({ key: 'other-tool', enabled: false, endpoints: ['https://example.org/feed'] }) });
+ assert.strictEqual(other.status, 201, other.text);
+});
+
+t('fetch now has its own, tighter budget (10 a minute)', async () => {
+ await api('POST', '/api/v1/sources', { token: STAFF_TOOL, body: { key: 'manual-codes', name: 'Codes', type: 'manual', category: 'coupons', terms_note: 'test', enabled: true } });
+ for (let i = 0; i < 10; i++) assert.strictEqual((await api('POST', '/api/v1/sources/manual-codes/fetch', { token: ADMIN })).status, 409, `fetch ${i + 1}`);
+ const r = await api('POST', '/api/v1/sources/manual-codes/fetch', { token: ADMIN });
+ assert.deepStrictEqual([r.status, r.body.code], [429, 'rate_limited']);
+});
+
+t('health, ready, release.json, metrics and the home page are never limited', async () => {
+ for (let i = 0; i < 6; i++) {
+ assert.strictEqual((await api('GET', '/api/health')).status, 200);
+ assert.notStrictEqual((await api('GET', '/api/ready')).status, 429);
+ assert.strictEqual((await api('GET', '/release.json')).status, 200);
+ assert.strictEqual((await api('GET', '/metrics')).status, 200);
+ assert.strictEqual((await api('GET', '/')).status, 200);
+ }
+});
+
+t('refusals are logged (the principal, never a token) and counted in sources_rate_limited_total', async () => {
+ assert.ok(lines.includes('[limits] sources.read: app:app_01J8ZQ4Y7N3M2K1H0G9F8E7D6C refused, over 3 per minute'), lines.join('\n'));
+ assert.ok(lines.includes('[limits] sources.source.manage: svc:network refused, over 30 per minute'), lines.join('\n'));
+ assert.ok(!lines.some((l) => /Bearer|eyJ/.test(l)), 'no token in the log');
+ const m = (await api('GET', '/metrics')).text;
+ const found = m.split('\n').filter((l) => l.includes('sources_rate_limited_total')).join('\n');
+ assert.ok(/sources_rate_limited_total\{limit="sources.read",window="minute"\} 1/.test(m), found);
+ assert.ok(/sources_rate_limited_total\{limit="sources.source.manage",window="minute"\} 1/.test(m), found);
+ assert.ok(/sources_rate_limited_total\{limit="sources.source.fetch",window="minute"\} 1/.test(m), found);
+});
+
+t('stop', async () => { await svc.stop(); });
+
+t.run();
@@ -39,8 +39,11 @@ function tmpDir() {
@@ -39,8 +39,11 @@ function tmpDir() {
return d;
return d;
}
}
-/** Boot Sources; the scheduler is off unless worker: 'on'. Loopback is allowlisted for the stubs. */
-async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now } = {}) {
+/**
+ * Boot Sources; the scheduler is off unless worker: 'on'. Loopback is allowlisted for the stubs.
+ * log: Sources' logger (default silent); limitsNow: the per-actor limiter's clock (default the wall clock).
+ */
+async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now, log = silent, limitsNow = null } = {}) {
const dir = tmpDir();
const dir = tmpDir();
const config = load({
const config = load({
NODE_ENV: 'test',
NODE_ENV: 'test',
@@ -54,7 +57,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now } =
@@ -54,7 +57,7 @@ async function boot({ env = {}, worker = 'off', lookupImpl, tokenClient, now } =
SOURCES_TICK_MS: '50',
SOURCES_TICK_MS: '50',
...env,
...env,
});
});
- const h = await start({ config, log: silent, lookupImpl, tokenClient, ...(now ? { now } : {}) });
+ const h = await start({ config, log, lookupImpl, tokenClient, limitsNow, ...(now ? { now } : {}) });
const base = `http://127.0.0.1:${h.server.address().port}`;
const base = `http://127.0.0.1:${h.server.address().port}`;
return { ...h, base, dir, async stop() { await h.close(); } };
return { ...h, base, dir, async stop() { await h.close(); } };
}
}