Analytics: Annahme gehaertet

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

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); });

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;
 }

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 });

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();
+});