diff --git a/analytics/src/index.js b/analytics/src/index.js index 4f422ca..5dbe737 100644 --- a/analytics/src/index.js +++ b/analytics/src/index.js @@ -12,9 +12,10 @@ const db = openDb(join(config.dataDir, 'analytics.db')); const geo = loadGeo(config.geoFile || join(config.dataDir, 'geo.csv')); if (geo.size) console.log(`Analytics: ${geo.size} Länder-Bereiche geladen`); +const publicServer = createPublicServer({ config, db, geo, salt: createSalt() }); const servers = [ - ['öffentlich', createPublicServer({ config, db, geo, salt: createSalt() }), config.port], - ['intern (Admin)', createAdminServer({ config, db }), config.adminPort], + ['öffentlich', publicServer, config.port], + ['intern (Admin)', createAdminServer({ config, db, onSitesChanged: () => publicServer.invalidateSites() }), config.adminPort], ]; for (const [name, server, port] of servers) { server.on('error', (e) => { console.error(`Analytics-Listener ${name} (Port ${port}):`, e.message); process.exit(1); }); diff --git a/analytics/src/ingest.js b/analytics/src/ingest.js index aceac4a..f51c265 100644 --- a/analytics/src/ingest.js +++ b/analytics/src/ingest.js @@ -8,26 +8,28 @@ const HOST = /^[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?(\.[a-z0-9]([a-z0-9-]{0,61}[a-z const LIMIT = 120; const HOUR = 3600 * 1000; const MINUTE = 60 * 1000; -const FIELDS = { +const FIELDS = Object.assign(Object.create(null), { pv: ['t', 'site', 'path', 'ref', 'vp'], bn: ['t', 'site', 'a'], en: ['t', 'site', 'path', 'vp', 'sec', 'scroll'], ck: ['t', 'site', 'path', 'vp', 'x', 'y', 'sel'], ct: ['t', 'site', 'path', 'kind', 'host'], -}; +}); +const SITES_TTL = 30 * 1000; + +function tooBig() { return Object.assign(new Error('zu groß'), { status: 413 }); } function readBody(req) { return new Promise((resolve, reject) => { + if (Number(req.headers['content-length']) > MAX_BODY) return reject(tooBig()); const chunks = []; let size = 0; - let tooBig = false; req.on('data', (c) => { - if (tooBig) return; size += c.length; - if (size > MAX_BODY) { tooBig = true; chunks.length = 0; return; } + if (size > MAX_BODY) { chunks.length = 0; reject(tooBig()); return; } chunks.push(c); }); - req.on('end', () => (tooBig ? reject(Object.assign(new Error('zu groß'), { status: 413 })) : resolve(Buffer.concat(chunks).toString('utf8')))); + req.on('end', () => resolve(Buffer.concat(chunks).toString('utf8'))); req.on('error', reject); }); } @@ -39,8 +41,17 @@ function clientIp(req) { return ip.replace(/^::ffff:/i, ''); } +// Nur https ohne expliziten Port zählt; alles andere ergibt '' (kein Treffer). function hostOf(value) { if (typeof value !== 'string' || !value) return ''; + try { + const u = new URL(value); + return u.protocol === 'https:' && !u.port ? u.hostname.toLowerCase() : ''; + } catch { return ''; } +} + +// Für cleanRef: Host einer beliebigen Referrer-URL (auch http, mit Port). +function anyHostOf(value) { try { return new URL(value).hostname.toLowerCase(); } catch { return ''; } } @@ -59,7 +70,7 @@ function cleanHost(h) { function cleanRef(ref, domains) { if (typeof ref !== 'string' || !ref) return null; - const host = cleanHost(ref.includes('://') ? hostOf(ref) : ref); + const host = cleanHost(ref.includes('://') ? anyHostOf(ref) : ref); return host && !domains.includes(host) ? host : null; } @@ -70,6 +81,8 @@ const clamp = (v, lo, hi) => Math.min(hi, Math.max(lo, v)); export function createIngest({ db, geo = { lookup: () => '' }, salt, now = Date.now }) { const getSite = db.prepare('SELECT domains, active FROM sites WHERE siteKey = ?'); const activeSites = db.prepare('SELECT domains FROM sites WHERE active = 1'); + let hostCache = null; // { at, hosts: Set } für Preflight-Prüfungen + let hostCacheStale = true; const insEvent = db.prepare(`INSERT INTO events_raw (site, ts, type, path, vp, ref, country, device, visitor, sec, scroll, kind, host) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`); const insClick = db.prepare('INSERT INTO clicks (site, ts, path, vp, x, y, sel) VALUES (?, ?, ?, ?, ?, ?, ?)'); @@ -79,6 +92,17 @@ export function createIngest({ db, geo = { lookup: () => '' }, salt, now = Date. const originHost = (req) => hostOf(req.headers.origin) || (req.headers.origin ? '' : hostOf(req.headers.referer)); const parseDomains = (s) => { try { return JSON.parse(s); } catch { return []; } }; + function activeHosts() { + const t = now(); + if (hostCacheStale || !hostCache || t - hostCache.at >= SITES_TTL || t < hostCache.at) { + const hosts = new Set(); + for (const s of activeSites.all()) for (const d of parseDomains(s.domains)) hosts.add(d); + hostCache = { at: t, hosts }; + hostCacheStale = false; + } + return hostCache.hosts; + } + function limited(key) { const t = now(); const w = Math.floor(t / MINUTE); @@ -96,7 +120,8 @@ export function createIngest({ db, geo = { lookup: () => '' }, salt, now = Date. res.setHeader('Vary', 'Origin'); } - function store(ev, site, domains, req, ip, ua) { + function store(ev, site, domains, ip, ua) { + if (typeof ev.t !== 'string') return; const spec = FIELDS[ev.t]; if (!spec) return; const e = {}; @@ -144,13 +169,20 @@ export function createIngest({ db, geo = { lookup: () => '' }, salt, now = Date. } // Liefert true, wenn die Anfrage /a/e betraf. - return async function handle(req, res, pathname) { + async function handle(req, res, pathname) { if (pathname !== '/a/e') return false; - const done = (status) => { res.writeHead(status); res.end(); return true; }; + const done = (status) => { + if (status === 413) { // Rest des Bodys nicht abwarten: Antwort senden, danach Verbindung kappen + res.writeHead(413, { Connection: 'close' }); + res.end(() => req.destroy()); + return true; + } + res.writeHead(status); res.end(); return true; + }; if (req.method === 'OPTIONS') { const host = originHost(req); const origin = req.headers.origin; - if (host && origin && activeSites.all().some((s) => parseDomains(s.domains).includes(host))) { + if (host && origin && activeHosts().has(host)) { cors(res, origin); res.setHeader('Access-Control-Allow-Methods', 'POST, OPTIONS'); res.setHeader('Access-Control-Allow-Headers', 'Content-Type'); @@ -174,7 +206,10 @@ export function createIngest({ db, geo = { lookup: () => '' }, salt, now = Date. if (classifyUa(ua).bot) return done(204); const ip = clientIp(req); if (limited(salt.rateKey(ev.site, ip))) return done(204); - try { store(ev, ev.site, domains, req, ip, ua); } catch (e) { console.error('Analytics-Annahme:', e.message); } + try { store(ev, ev.site, domains, ip, ua); } catch (e) { console.error('Analytics-Annahme:', e.message); } return done(204); - }; + } + + handle.invalidateSites = () => { hostCacheStale = true; }; + return handle; } diff --git a/analytics/src/server.js b/analytics/src/server.js index b30b093..8b484c9 100644 --- a/analytics/src/server.js +++ b/analytics/src/server.js @@ -55,9 +55,9 @@ function validateSite(b) { return { site: { projectId: b.projectId, domains: b.domains, active: b.active } }; } -function createHandlerServer({ config, db, role, geo, salt, now }) { +function createHandlerServer({ config, db, role, geo, salt, now, onSitesChanged }) { const ingest = role === 'public' ? createIngest({ db, geo, salt: salt || createSalt({ now }), now }) : null; - return http.createServer(async (req, res) => { + const server = http.createServer(async (req, res) => { try { const url = new URL(req.url, 'http://x'); const parts = url.pathname.split('/').filter(Boolean); @@ -85,6 +85,7 @@ function createHandlerServer({ config, db, role, geo, salt, now }) { db.exec('ROLLBACK'); throw e; } + onSitesChanged?.(); return send(res, 200, { ok: true }); } const v = validateSite(await readJson(req, MAX_ADMIN_BODY)); @@ -96,6 +97,7 @@ function createHandlerServer({ config, db, role, geo, salt, now }) { ON CONFLICT(siteKey) DO UPDATE SET projectId = excluded.projectId, domains = excluded.domains, active = excluded.active, offSince = excluded.offSince`) .run(key, projectId, JSON.stringify(domains), active ? 1 : 0, offSince); + onSitesChanged?.(); return send(res, 200, { ok: true }); } return send(res, 404, { error: 'Nicht gefunden' }); @@ -105,9 +107,12 @@ function createHandlerServer({ config, db, role, geo, salt, now }) { if (!res.headersSent) send(res, status, { error: status === 500 ? 'Interner Fehler' : e.message }); } }); + server.invalidateSites = ingest ? ingest.invalidateSites : () => {}; + return server; } // Öffentlicher Listener: /a/health und /a/e (Erfassung); Admin-Routen gibt es hier nicht. export const createPublicServer = ({ config, db, geo, salt, now }) => createHandlerServer({ config, db, role: 'public', geo, salt, now }); // Interner Listener für die Website-Verwaltung. -export const createAdminServer = ({ config, db }) => createHandlerServer({ config, db, role: 'admin' }); +// onSitesChanged: wird nach jeder Änderung an Websites gerufen (public.invalidateSites()). +export const createAdminServer = ({ config, db, onSitesChanged }) => createHandlerServer({ config, db, role: 'admin', onSitesChanged }); diff --git a/analytics/test/ingest.test.js b/analytics/test/ingest.test.js index e7b9ef1..0f2e27f 100644 --- a/analytics/test/ingest.test.js +++ b/analytics/test/ingest.test.js @@ -1,7 +1,8 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; import { openDb } from '../src/db.js'; -import { createPublicServer } from '../src/server.js'; +import net from 'node:net'; +import { createPublicServer, createAdminServer } from '../src/server.js'; import { classifyUa } from '../src/ua.js'; import { createSalt } from '../src/salt.js'; @@ -16,16 +17,17 @@ async function setup({ now = () => Date.UTC(2026, 9, 1, 10, 30, 45), geo = { loo db.prepare('INSERT INTO sites (siteKey, projectId, domains, active) VALUES (?,?,?,1)') .run(KEY, 'p1', JSON.stringify([HOST, 'www.kunde.de'])); const server = createPublicServer({ config: {}, db, geo, now }); + const invalidate = () => server.invalidateSites(); await new Promise((r) => server.listen(0, '127.0.0.1', r)); const base = `http://127.0.0.1:${server.address().port}`; - const post = (ev, { origin = ORIGIN, ua = UA, ip = IP, referer, raw, type = 'text/plain' } = {}) => { - const headers = { 'Content-Type': type, 'User-Agent': ua, 'X-Forwarded-For': `1.2.3.4, ${ip}` }; + const post = (ev, { origin = ORIGIN, ua = UA, ip = IP, left = '1.2.3.4', referer, raw, type = 'text/plain' } = {}) => { + const headers = { 'Content-Type': type, 'User-Agent': ua, 'X-Forwarded-For': `${left}, ${ip}` }; if (origin) headers.Origin = origin; if (referer) headers.Referer = referer; return fetch(base + '/a/e', { method: 'POST', headers, body: raw ?? JSON.stringify(ev) }); }; const rows = () => db.prepare('SELECT * FROM events_raw ORDER BY id').all(); - return { db, post, rows, base, close: () => server.close() }; + return { db, post, rows, base, server, invalidate, close: () => server.close() }; } const pv = (o = {}) => ({ t: 'pv', site: KEY, path: '/', vp: 'd', ...o }); @@ -249,3 +251,102 @@ test('Datenschutz: weder IP noch User-Agent irgendwo in der Datenbank', async () for (const needle of [IP, '1.2.3.4', 'GEHEIMERUA', 'Mozilla', 'iPhone']) assert.ok(!dump.includes(needle), needle); t.close(); }); + +test('t: nur eigene Feldlisten, kein Prototyp-Zugriff, keine Logzeile', async () => { + const t = await setup(); + const errs = []; + const orig = console.error; + console.error = (...a) => errs.push(a.join(' ')); + try { + for (const tt of ['__proto__', 'constructor', 'toString', 'hasOwnProperty', 'valueOf']) { + assert.equal((await t.post(null, { raw: JSON.stringify({ t: tt, site: KEY, path: '/', vp: 'd' }) })).status, 204); + } + assert.equal((await t.post(null, { raw: `{"t":"__proto__","site":"${KEY}","path":"/"}` })).status, 204); + assert.equal((await t.post({ t: ['pv'], site: KEY, path: '/', vp: 'd' })).status, 204); + assert.equal((await t.post({ t: { toString: 1 }, site: KEY, path: '/', vp: 'd' })).status, 204); + } finally { console.error = orig; } + assert.equal(t.rows().length, 0); + assert.deepEqual(errs, []); + t.close(); +}); + +test('Origin/Referer: nur https ohne Port zählt', async () => { + const t = await setup(); + for (const origin of [`http://${HOST}`, `https://${HOST}:8080`, `https://evil-${HOST}`, `https://${HOST}.evil.com`, `ftp://${HOST}`]) { + const res = await t.post(pv(), { origin }); + assert.equal(res.headers.get('access-control-allow-origin'), null, origin); + const pre = await fetch(t.base + '/a/e', { method: 'OPTIONS', headers: { Origin: origin } }); + assert.equal(pre.headers.get('access-control-allow-origin'), null, origin); + } + await t.post(pv(), { origin: null, referer: `http://${HOST}/x` }); + await t.post(pv(), { origin: null, referer: `https://${HOST}:8080/x` }); + assert.equal(t.rows().length, 0); + await t.post(pv(), { origin: null, referer: `https://${HOST}/x` }); + assert.equal(t.rows().length, 1); + t.close(); +}); + +function rawRequest(port, head, body = '') { + return new Promise((resolve) => { + const s = net.connect(port, '127.0.0.1'); + let buf = ''; + s.on('connect', () => s.write(head + '\r\n\r\n' + body)); + s.on('data', (d) => { buf += d; }); + s.on('close', () => resolve(buf)); + s.on('error', () => resolve(buf)); + setTimeout(() => { s.destroy(); }, 1500); + }); +} + +test('Body: früh 413 bei Content-Length > 4096, Abbruch bei gestreamtem Überlauf', async () => { + const t = await setup(); + const port = t.server.address().port; + const head = `POST /a/e HTTP/1.1\r\nHost: x\r\nOrigin: ${ORIGIN}\r\nContent-Type: text/plain`; + // Content-Length 1 MB angekündigt, aber nichts gesendet: Antwort kommt trotzdem sofort + const early = await rawRequest(port, `${head}\r\nContent-Length: 1048576`); + assert.match(early, /^HTTP\/1\.1 413/); + // chunked ohne Content-Length: über 4 KB ⇒ 413 und Verbindung wird beendet + const chunk = 'x'.repeat(3000); + const body = `${chunk.length.toString(16)}\r\n${chunk}\r\n`.repeat(3); + const streamed = await rawRequest(port, `${head}\r\nTransfer-Encoding: chunked`, body); + assert.match(streamed, /^HTTP\/1\.1 413/); + assert.equal(t.rows().length, 0); + t.close(); +}); + +test('Preflight: Hostliste gecacht (30 s), invalidateSites() und Admin-Änderung wirken sofort', async () => { + let now = Date.UTC(2026, 9, 1, 10, 0, 0); + const t = await setup({ now: () => now }); + const pre = async () => (await fetch(t.base + '/a/e', { method: 'OPTIONS', headers: { Origin: ORIGIN } })).headers.get('access-control-allow-origin'); + assert.equal(await pre(), ORIGIN); + t.db.prepare('UPDATE sites SET active = 0').run(); + assert.equal(await pre(), ORIGIN); // noch aus dem Cache + now += 31000; + assert.equal(await pre(), null); // Cache abgelaufen + t.db.prepare('UPDATE sites SET active = 1').run(); + assert.equal(await pre(), null); + t.invalidate(); + assert.equal(await pre(), ORIGIN); + // Admin-Listener meldet Änderungen an den öffentlichen Listener + const admin = createAdminServer({ config: { adminToken: 'geheim' }, db: t.db, onSitesChanged: () => t.invalidate() }); + await new Promise((r) => admin.listen(0, '127.0.0.1', r)); + const res = await fetch(`http://127.0.0.1:${admin.address().port}/a/admin/sites/${KEY}`, { + method: 'PUT', headers: { Authorization: 'Bearer geheim', 'Content-Type': 'application/json' }, + body: JSON.stringify({ projectId: 'p1', domains: [HOST], active: false }), + }); + assert.equal(res.status, 200); + assert.equal(await pre(), null); + admin.close(); + t.close(); +}); + +test('X-Forwarded-For: rechtester Eintrag zählt für das Mengenlimit', async () => { + const t = await setup(); + for (let i = 0; i < 120; i++) await t.post(pv(), { left: `10.0.0.${i % 250}` }); // links variiert, rechts gleich + assert.equal(t.rows().length, 120); + await t.post(pv(), { left: '10.9.9.9' }); // gleicher Bucket ⇒ verworfen + assert.equal(t.rows().length, 120); + await t.post(pv(), { left: '1.2.3.4', ip: '198.51.100.9' }); // rechts anders ⇒ eigener Bucket + assert.equal(t.rows().length, 121); + t.close(); +});