diff --git a/toolchain/aespatchercron/README.md b/toolchain/aespatchercron/README.md index b56336b508..3572ece0a9 100644 --- a/toolchain/aespatchercron/README.md +++ b/toolchain/aespatchercron/README.md @@ -48,6 +48,8 @@ client analytics. Missing credentials and unsupported queries fail visibly. function exception prefixes on the server. Raw lines never leave Lith. * `lith-errors`: Lith's bounded in-memory error ring, fetched over loopback and reduced to function counts before transport. It resets when Lith restarts. +* Cloudflare and `lith-access`: hourly host/status aggregates, aligned to the + same UTC window, from Cloudflare analytics and Lith's retained Caddy logs. * `chat-clock` and `chat-system`: bounded recent public, nondeleted, nonmuted messages; only selected malfunction reports become private evidence. * PostHog: read-only, AC-bound aggregate piece events and endpoint status counts @@ -105,6 +107,48 @@ reports the number of manual-review diagnostics as well as worker candidates. Collector provenance hashes all collection/validation modules and the exact remote program sent to Lith. +## Cloudflare and Lith correlation + +Cloudflare reads use the existing vault credential resolver in +[`../domains/cloudflare.mjs`](../domains/cloudflare.mjs). No key is copied into +runtime state or source. `AESPATCHER_CLOUDFLARE_API_TOKEN` can supply a scoped +read token instead. Requests are limited to zone lookup and GraphQL analytics; +this collector never changes DNS, cache rules, security settings or deployment. + +The default zones are `aesthetic.computer`, `laklok.com`, `kidlisp.com`, +`notepat.com`, `oskiewar.com` and `nopaint.art`, with their reviewed hostname +aliases. `cloudflare.zones` can select up to eight known studio properties; +client domains are rejected. An inaccessible zone is reported individually. +Set `cloudflare.enabled: false` to disable this comparison. + +The reader requests `httpRequestsAdaptiveGroups` for eyeball requests, grouped +by UTC hour, hostname, edge status and origin status. Its returned counts are +already [estimates](https://developers.cloudflare.com/analytics/graphql-api/sampling/); +they are never multiplied by the sampling interval again. Each zone is bounded +to 1,000 groups, with overflow reported. Bots and direct origin traffic make +these request populations different from the piece-run/user telemetry. + +Caddy logs are reduced on Lith before transport: no paths, query strings, IPs, +headers, Ray IDs or log lines leave the host. Reads include the active log and +up to three rotated logs, capped at 256 MiB of decoded data and 45 seconds. +Retention gaps, malformed rows and read limits mark origin coverage incomplete. +Both layers use precisely the aggregate report's `[start, end)` window. + +`latest.json` includes `correlation.buckets`, matched by exact hostname and UTC +hour. Each bucket retains separate edge estimates and observed Lith counts; +an unobserved side is `null`, never a fabricated zero. The collector distinguishes +co-occurring edge/Lith 5xx, origin 5xx reported by Cloudflare, and edge failures +without a recorded origin response. Origin status zero does not prove Cloudflare +caused a failure: connection failures and other paths may lack an origin HTTP +response. Cached responses, other origins, sampling and ingestion delay also +prevent subtracting these totals or treating a mismatch as lost requests. + +These findings enter the manual-review count, not the automatic piece worker. +Timing overlap cannot authorize infrastructure changes or establish causality. +The `collect` command prints flagged buckets; all buckets remain in the private +evidence file for inspection. Lith, boot/piece telemetry and public reports +remain the primary product evidence; PostHog is optional context. + ## PostHog read access Set `posthog.projectId` and `posthog.organizationId` in private `config.json`, @@ -273,6 +317,9 @@ English/Danish report routing, moderation filters, redaction, instruction-like reports, public quotation rejection, real boot schema, bounded log coverage, client exclusion and source failures. PostHog tests cover project binding, aggregate query shape, missing read permissions and response validation. +Edge tests cover zone/host boundaries, estimated counts, partial source failure, +exact-hour joins, manual-only findings, rotated logs, retention gaps and +server-side removal of request details. The worktree integration test explicitly skips when the host performance guard returns exit 75; it does not bypass the guard. No test invokes a model or creates a remote PR. diff --git a/toolchain/aespatchercron/aespatchercron.mjs b/toolchain/aespatchercron/aespatchercron.mjs index 6555eca959..643894d813 100755 --- a/toolchain/aespatchercron/aespatchercron.mjs +++ b/toolchain/aespatchercron/aespatchercron.mjs @@ -72,7 +72,10 @@ async function main() { const evidence = await collect(config.repo, config); await save(join(home, "latest.json"), evidence); return { report: evidence.report, signals: { coverage: evidence.signals.coverage, leads: evidence.signals.leads, metrics: evidence.signals.metrics, - privateReportCount: evidence.signals.privateReports.length }, saved: join(home, "latest.json") }; + privateReportCount: evidence.signals.privateReports.length }, correlation: evidence.correlation && { + start: evidence.correlation.start, end: evidence.correlation.end, bucketSeconds: evidence.correlation.bucketSeconds, + buckets: evidence.correlation.buckets.length, findings: evidence.correlation.buckets.filter(row => row.assessment !== "no-error-signal"), + }, saved: join(home, "latest.json") }; } if (command === "scan") return scan(config, rest[0]); if (command === "tick") { diff --git a/toolchain/aespatchercron/collect.mjs b/toolchain/aespatchercron/collect.mjs index cf6fb51f71..4998bcaee4 100644 --- a/toolchain/aespatchercron/collect.mjs +++ b/toolchain/aespatchercron/collect.mjs @@ -5,6 +5,7 @@ import { collectDiagnostics, journalNames, errorNames, functionConsumers } from import { userReports } from "./reports.mjs"; import { emptySignals, mergeSignals, validateSignals } from "./signals.mjs"; import { collectPostHog } from "./posthog.mjs"; +import { collectEdge } from "./edge.mjs"; export async function catalog(repo) { const prefix = "system/public/aesthetic.computer/disks/"; @@ -114,10 +115,15 @@ export async function collect(repo, config) { parsed = { signals, report: { format: "aespatcher.signals.v1", start: new Date(Date.parse(opts.end) - opts.hours * 3600000).toISOString(), end: opts.end, minimum: opts.minimum, runs: [], transitions: [], visits: [], unavailable: ["runs", "piece-runs", "account-activity", "visits"], truncated: [] } }; } - const signals = validateSignals(mergeSignals(parsed.signals, await collectPostHog(repo, config)), routes); + const window = { start: parsed.report.start, end: parsed.report.end }; + const [posthog, edge] = await Promise.all([ + collectPostHog(repo, { ...config, collection: { ...config.collection, end: opts.end } }), + collectEdge(config, window), + ]); + const signals = validateSignals(mergeSignals(parsed.signals, posthog, edge.signals), routes); const modules = {}; - for (const file of ["collect.mjs", "diagnostics.mjs", "reports.mjs", "signals.mjs", "posthog.mjs"]) + for (const file of ["collect.mjs", "diagnostics.mjs", "reports.mjs", "signals.mjs", "posthog.mjs", "edge.mjs", "lith-access.mjs", "../domains/cloudflare.mjs"]) modules[file] = hash(await readFile(new URL(file, import.meta.url))); - return { report: parsed.report, signals, provenance: { revision: await git(repo, "rev-parse", "HEAD"), + return { report: parsed.report, signals, correlation: edge.correlation, provenance: { revision: await git(repo, "rev-parse", "HEAD"), collector: hash(JSON.stringify(modules)), modules, remotePayload: hash(source), at: new Date().toISOString() } }; } diff --git a/toolchain/aespatchercron/config.mjs b/toolchain/aespatchercron/config.mjs index 6df4b76c0c..f75377f593 100644 --- a/toolchain/aespatchercron/config.mjs +++ b/toolchain/aespatchercron/config.mjs @@ -9,7 +9,8 @@ export function defaults(repo) { return { repo, fetch: true, collection: { hours: 24, minimum: 3, limit: 100, chatLimit: 200, logLimit: 2000 }, lith: { host: "root@lith.aesthetic.computer", root: "/opt/ac", identity: join(repo, "aesthetic-computer-vault/home/.ssh/id_rsa") }, bounds: { perDay: 2, maxActive: 4, maxFiles: 4, maxLines: 200 }, - worker: { executable: "codex", timeoutMs: 900000 }, posthog: { enabled: true, projectId: null, organizationId: null }, hosting: { kind: "unconfigured" } }; + worker: { executable: "codex", timeoutMs: 900000 }, cloudflare: { enabled: true }, + posthog: { enabled: true, projectId: null, organizationId: null }, hosting: { kind: "unconfigured" } }; } export function validateConfig(config) { diff --git a/toolchain/aespatchercron/edge.mjs b/toolchain/aespatchercron/edge.mjs new file mode 100644 index 0000000000..180a81484a --- /dev/null +++ b/toolchain/aespatchercron/edge.mjs @@ -0,0 +1,162 @@ +import { credentials } from "../domains/cloudflare.mjs"; +import { VISIT_PROPERTIES, visitGroup } from "../../system/public/aesthetic.computer/lib/visit-model.mjs"; +import { run } from "./io.mjs"; +import { collectAccessLogs } from "./lith-access.mjs"; + +export const edgeZones = ["aesthetic.computer", "laklok.com", "kidlisp.com", "notepat.com", "oskiewar.com", "nopaint.art"]; +const api = "https://api.cloudflare.com/client/v4"; +const hour = value => new Date(Math.floor(Date.parse(value) / 3600000) * 3600000).toISOString(); +const integer = value => Number.isSafeInteger(value) && value >= 0; + +export function edgeScope(config, window) { + const zones = config.cloudflare?.zones || edgeZones; + if (!Array.isArray(zones) || !zones.length || zones.length > 8 || new Set(zones).size !== zones.length || + zones.some(zone => !Object.hasOwn(VISIT_PROPERTIES, zone) || visitGroup(zone) !== "studio") || + !Number.isFinite(Date.parse(window.start)) || !Number.isFinite(Date.parse(window.end)) || + Date.parse(window.end) <= Date.parse(window.start) || Date.parse(window.end) - Date.parse(window.start) > 168 * 3600000) + throw new Error("Invalid edge correlation scope"); + return { ...window, zones, hosts: [...new Set(zones.flatMap(zone => [zone, ...VISIT_PROPERTIES[zone]]))], limit: 1000 }; +} + +async function readCloudflare(path, headers, fetchImpl, body) { + const response = await fetchImpl(api + path, { method: body ? "POST" : "GET", headers, redirect: "error", + body: body ? JSON.stringify(body) : undefined, signal: AbortSignal.timeout(20000) }); + if (!response.ok) throw new Error("Cloudflare read unavailable"); + const chunks = []; let bytes = 0; + for await (const chunk of response.body) { + bytes += chunk.byteLength; + if (bytes > 2 * 1024 * 1024) throw new Error("Cloudflare response exceeds bound"); + chunks.push(chunk); + } + return JSON.parse(Buffer.concat(chunks).toString("utf8")); +} + +export function edgeQuery(zoneId, hosts, scope) { + if (!/^[a-f0-9]{32}$/.test(zoneId) || !hosts.length || hosts.some(host => !scope.hosts.includes(host))) throw new Error("Invalid edge query binding"); + return `query { viewer { zones(filter: {zoneTag: ${JSON.stringify(zoneId)}}) { + httpRequestsAdaptiveGroups(limit: ${scope.limit + 1}, orderBy: [datetimeHour_ASC], filter: { + datetime_geq: ${JSON.stringify(scope.start)}, datetime_lt: ${JSON.stringify(scope.end)}, + clientRequestHTTPHost_in: ${JSON.stringify(hosts)}, requestSource: "eyeball" + }) { count dimensions { datetimeHour clientRequestHTTPHost edgeResponseStatus originResponseStatus } avg { sampleInterval } } + } } }`; +} + +export function edgeRows(result, hosts, scope) { + if (result.errors?.length || result.data?.viewer?.zones?.length !== 1) throw new Error("Cloudflare query unavailable"); + const rows = result.data.viewer.zones[0].httpRequestsAdaptiveGroups; + if (!Array.isArray(rows) || rows.length > scope.limit + 1) throw new Error("Invalid edge groups"); + const buckets = new Map(); + for (const row of rows.slice(0, scope.limit)) { + const d = row.dimensions, interval = row.avg?.sampleInterval; + if (!d || !hosts.includes(d.clientRequestHTTPHost) || !integer(row.count) || !integer(d.edgeResponseStatus) || + (d.edgeResponseStatus !== 0 && (d.edgeResponseStatus < 100 || d.edgeResponseStatus > 599)) || !integer(d.originResponseStatus) || + (d.originResponseStatus !== 0 && (d.originResponseStatus < 100 || d.originResponseStatus > 599)) || + !Number.isFinite(interval) || interval < 1 || !Number.isFinite(Date.parse(d.datetimeHour)) || + Date.parse(d.datetimeHour) !== Date.parse(hour(d.datetimeHour)) || + Date.parse(d.datetimeHour) < Date.parse(hour(scope.start)) || Date.parse(d.datetimeHour) >= Date.parse(scope.end)) + throw new Error("Invalid edge aggregate"); + const host = d.clientRequestHTTPHost, time = hour(d.datetimeHour), key = `${host}:${time}`; + const bucket = buckets.get(key) || { host, hour: time, requests: 0, errors5xx: 0, origin5xx: 0, withoutOrigin5xx: 0, afterOriginSuccess5xx: 0, sampled: false }; + bucket.requests += row.count; + if (d.edgeResponseStatus >= 500) { + bucket.errors5xx += row.count; + if (d.originResponseStatus === 0) bucket.withoutOrigin5xx += row.count; + if (d.originResponseStatus >= 200 && d.originResponseStatus < 400) bucket.afterOriginSuccess5xx += row.count; + } + if (d.originResponseStatus >= 500) bucket.origin5xx += row.count; + bucket.sampled ||= interval > 1; + if (![bucket.requests, bucket.errors5xx, bucket.origin5xx].every(integer)) throw new Error("Edge count exceeds bound"); + buckets.set(key, bucket); + } + return { rows: [...buckets.values()], groups: rows.length, truncated: rows.length > scope.limit }; +} + +export async function collectCloudflare(scope, { fetchImpl = globalThis.fetch, auth = credentials, env = process.env } = {}) { + const token = env.AESPATCHER_CLOUDFLARE_API_TOKEN; + const { email, apiKey } = token ? {} : auth(); + if (!token && (!email || !apiKey)) return { rows: [], coverage: scope.zones.map(zone => ({ zone, status: "unavailable", reason: "missing-credential" })) }; + const headers = { "content-type": "application/json", ...(token ? { Authorization: `Bearer ${token}` } : { "X-Auth-Email": email, "X-Auth-Key": apiKey }) }; + const zones = await Promise.all(scope.zones.map(async zone => { + try { + const result = await readCloudflare(`/zones?name=${encodeURIComponent(zone)}`, headers, fetchImpl); + const matches = result.success && Array.isArray(result.result) ? result.result.filter(row => row.name === zone && row.status === "active") : []; + if (matches.length !== 1) return { rows: [], coverage: { zone, status: "unavailable", reason: "zone-not-accessible" } }; + const hosts = [zone, ...VISIT_PROPERTIES[zone]]; + const groups = edgeRows(await readCloudflare("/graphql", headers, fetchImpl, { query: edgeQuery(matches[0].id, hosts, scope) }), hosts, scope); + return { rows: groups.rows, coverage: { zone, status: "available", groups: groups.groups, truncated: groups.truncated, estimated: true } }; + } catch { return { rows: [], coverage: { zone, status: "unavailable", reason: "analytics-read-unavailable" } }; } + })); + return { rows: zones.flatMap(zone => zone.rows), coverage: zones.map(zone => zone.coverage) }; +} + +export function validateAccess(data, scope) { + const fields = ["host", "hour", "requests", "errors5xx"]; + if (data.status !== "available" || !Array.isArray(data.rows) || data.rows.length > 2000 || !integer(data.scanned) || typeof data.truncated !== "boolean" || + data.rows.some(row => Object.keys(row).length !== fields.length || !fields.every(field => Object.hasOwn(row, field)) || + !scope.hosts.includes(row.host) || !Number.isFinite(Date.parse(row.hour)) || row.hour !== hour(row.hour) || + Date.parse(row.hour) < Date.parse(hour(scope.start)) || Date.parse(row.hour) >= Date.parse(scope.end) || + !integer(row.requests) || !integer(row.errors5xx) || row.errors5xx > row.requests)) throw new Error("Invalid Lith access aggregates"); + return { status: "available", rows: data.rows, scanned: data.scanned, truncated: data.truncated }; +} + +async function readLithAccess(config, scope) { + if (!/^[a-zA-Z0-9_.@-]+$/.test(config.lith?.host || "")) throw new Error("Invalid Lith host"); + const program = `import { readdir, lstat } from "node:fs/promises"; import { createReadStream } from "node:fs"; import { createGunzip } from "node:zlib"; + (${collectAccessLogs.toString()})(${JSON.stringify(scope)}).then(result => process.stdout.write(JSON.stringify(result))).catch(() => { process.exitCode = 1; });`; + try { + const result = await run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=10", ...(config.lith.identity ? ["-i", config.lith.identity] : []), + config.lith.host, "node --input-type=module"], { input: program, timeout: 90000 }); + if (result.code !== 0) throw new Error("Access log unavailable"); + return validateAccess(JSON.parse(result.stdout), scope); + } catch { return { status: "unavailable", rows: [], reason: "access-log-read-unavailable" }; } +} + +export function correlateEdge(scope, cloudflare, access, minimum = 3) { + const buckets = new Map(); + for (const row of cloudflare.rows) buckets.set(`${row.host}:${row.hour}`, { host: row.host, hour: row.hour, edge: { ...row }, lith: null }); + for (const row of access.rows) { + const key = `${row.host}:${row.hour}`, bucket = buckets.get(key) || { host: row.host, hour: row.hour, edge: null, lith: null }; + bucket.lith = { requests: row.requests, errors5xx: row.errors5xx }; buckets.set(key, bucket); + } + const signals = { format: "aespatcher.supplement.v1", coverage: [], metrics: [], leads: [], privateReports: [] }; + for (const entry of cloudflare.coverage) signals.coverage.push({ source: "cloudflare", status: entry.status, reason: `${entry.zone}: ${entry.reason || "adaptive estimates; host/hour comparison only"}`, + ...(entry.status === "available" ? { scanned: entry.groups, truncated: entry.truncated } : {}) }); + signals.coverage.push({ source: "lith-access", status: access.status, ...(access.status === "available" + ? { scanned: access.scanned, truncated: access.truncated, reason: "Retained Caddy logs; direct traffic and edge traffic differ." } : { reason: access.reason }) }); + const findings = new Map(), totals = new Map(); + for (const bucket of buckets.values()) { + const { edge, lith, host } = bucket; + bucket.assessment = edge?.errors5xx >= minimum && lith?.errors5xx >= minimum ? "edge-and-lith-errors" + : edge?.origin5xx >= minimum ? "origin-error-at-edge" + : edge?.withoutOrigin5xx >= minimum ? "edge-without-origin-response" + : edge?.errors5xx >= minimum ? "edge-error" : lith?.errors5xx >= minimum ? "lith-error" : "no-error-signal"; + const total = totals.get(host) || { edgeRequests: 0, edge5xx: 0, origin5xx: 0, lithRequests: 0, lith5xx: 0, hasEdge: false, hasLith: false }; + total.hasEdge ||= edge !== null; total.hasLith ||= lith !== null; + total.edgeRequests += edge?.requests || 0; total.edge5xx += edge?.errors5xx || 0; total.origin5xx += edge?.origin5xx || 0; + total.lithRequests += lith?.requests || 0; total.lith5xx += lith?.errors5xx || 0; totals.set(host, total); + if (bucket.assessment !== "no-error-signal") { + const key = `${host}:${bucket.assessment}`, finding = findings.get(key) || { host, assessment: bucket.assessment, hours: 0, count: 0 }; + finding.hours++; + finding.count += bucket.assessment === "origin-error-at-edge" ? edge.origin5xx + : bucket.assessment === "edge-without-origin-response" ? edge.withoutOrigin5xx + : bucket.assessment === "lith-error" ? lith.errors5xx : edge.errors5xx; + findings.set(key, finding); + } + } + for (const [host, total] of totals) for (const metric of ["edgeRequests", "edge5xx", "origin5xx", "lithRequests", "lith5xx"]) { + const lith = metric.startsWith("lith"); + if (lith ? total.hasLith : total.hasEdge) signals.metrics.push({ source: lith ? "lith-access" : "cloudflare", metric, target: host, count: total[metric] }); + } + for (const finding of findings.values()) signals.leads.push({ source: finding.assessment === "lith-error" ? "lith-access" : "cloudflare", kind: "edge-correlation", route: null, + count: finding.count, evidenceRefs: [], summary: `${finding.host}: ${finding.assessment} in ${finding.hours} UTC hour buckets. Aggregate timing overlap is a lead, not proof of cause; review coverage and reproduce.` }); + return { signals, correlation: { format: "aespatcher.edge.v1", start: scope.start, end: scope.end, bucketSeconds: 3600, + cloudflare: cloudflare.coverage, lith: { status: access.status, ...(access.status === "available" ? { truncated: access.truncated } : { reason: access.reason }) }, + buckets: [...buckets.values()].sort((a, b) => a.hour.localeCompare(b.hour) || a.host.localeCompare(b.host)) } }; +} + +export async function collectEdge(config, window, dependencies = {}) { + if (config.cloudflare?.enabled === false) return { signals: { format: "aespatcher.supplement.v1", coverage: [{ source: "cloudflare", status: "disabled", reason: "operator-disabled" }], metrics: [], leads: [], privateReports: [] }, correlation: null }; + const scope = edgeScope(config, window); + const [cloudflare, access] = await Promise.all([collectCloudflare(scope, dependencies), (dependencies.readAccess || readLithAccess)(config, scope)]); + return correlateEdge(scope, cloudflare, access, config.collection?.minimum || 3); +} diff --git a/toolchain/aespatchercron/lith-access.mjs b/toolchain/aespatchercron/lith-access.mjs new file mode 100644 index 0000000000..c698c27692 --- /dev/null +++ b/toolchain/aespatchercron/lith-access.mjs @@ -0,0 +1,53 @@ +import { readdir, lstat } from "node:fs/promises"; +import { createReadStream } from "node:fs"; +import { createGunzip } from "node:zlib"; + +// Runs on Lith. Raw requests, headers, paths and addresses never leave this function. +export async function collectAccessLogs(scope, directory = "/var/log/caddy") { + const start = Date.parse(scope.start), end = Date.parse(scope.end); + const names = (await readdir(directory)).filter(name => /^access(?:-[0-9T:.a-z-]+)?\.log(?:\.gz)?$/.test(name)); + const files = []; + for (const name of names) { + const info = await lstat(`${directory}/${name}`); + if (!info.isFile() || info.isSymbolicLink()) continue; + files.push({ name, modified: +info.mtime }); + } + files.sort((a, b) => b.modified - a.modified); + if (!files.length) throw new Error("No access log files"); + const buckets = new Map(), maxBytes = 256 * 1024 * 1024, deadline = Date.now() + 45000; + let bytes = 0, scanned = 0, earliest = Infinity, truncated = false, malformed = false; + for (const file of files.slice(0, 4)) { + const input = createReadStream(`${directory}/${file.name}`), stream = file.name.endsWith(".gz") ? input.pipe(createGunzip()) : input; + // Forward a rotated-file read failure to the decompressor as well. + if (stream !== input) input.on("error", error => stream.destroy(error)); + let rest = ""; + try { + for await (const chunk of stream) { + bytes += chunk.length; + if (bytes > maxBytes || Date.now() > deadline) { truncated = true; break; } + const lines = (rest + chunk.toString("utf8")).split("\n"); rest = lines.pop(); + if (rest.length > 1024 * 1024) throw new Error("Oversized access row"); + for (const line of lines) { + if (!line) continue; + let row; + try { row = JSON.parse(line); } catch { malformed = true; continue; } + const at = row.ts * 1000; + if (!Number.isFinite(at)) { malformed = true; continue; } + earliest = Math.min(earliest, at); + if (at < start || at >= end) continue; + const host = typeof row.request?.host === "string" ? row.request.host.toLowerCase().replace(/:\d+$/, "") : ""; + if (!scope.hosts.includes(host)) continue; + if (!Number.isInteger(row.status) || row.status < 100 || row.status > 599) { malformed = true; continue; } + const hour = new Date(Math.floor(at / 3600000) * 3600000).toISOString(), key = `${host}:${hour}`; + const bucket = buckets.get(key) || { host, hour, requests: 0, errors5xx: 0 }; + bucket.requests++; bucket.errors5xx += Number(row.status >= 500); scanned++; + buckets.set(key, bucket); + } + } + if (rest.trim()) malformed = true; + } finally { input.destroy(); stream.destroy(); } + if (truncated || earliest <= start) break; + } + // Retention, a read cap, or malformed rows make absence of origin errors inconclusive. + return { status: "available", rows: [...buckets.values()], scanned, truncated: truncated || malformed || earliest > start }; +} diff --git a/toolchain/aespatchercron/signals.mjs b/toolchain/aespatchercron/signals.mjs index 79193deb08..2f78d0cc44 100644 --- a/toolchain/aespatchercron/signals.mjs +++ b/toolchain/aespatchercron/signals.mjs @@ -1,4 +1,4 @@ -const sources = ["boots", "lith-journal", "lith-errors", "chat-clock", "chat-system", "posthog"]; +const sources = ["boots", "lith-journal", "lith-errors", "lith-access", "cloudflare", "chat-clock", "chat-system", "posthog"]; export const emptySignals = () => ({ format: "aespatcher.supplement.v1", coverage: [], leads: [], privateReports: [], metrics: [] }); export function validateSignals(data, routes) { diff --git a/toolchain/aespatchercron/test/edge.test.mjs b/toolchain/aespatchercron/test/edge.test.mjs new file mode 100644 index 0000000000..c60f27be6c --- /dev/null +++ b/toolchain/aespatchercron/test/edge.test.mjs @@ -0,0 +1,128 @@ +import { test } from "node:test"; +import assert from "node:assert/strict"; +import { mkdtemp, writeFile, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { gzipSync } from "node:zlib"; +import { edgeScope, edgeQuery, edgeRows, collectCloudflare, correlateEdge, collectEdge, validateAccess } from "../edge.mjs"; +import { collectAccessLogs } from "../lith-access.mjs"; +import { validateSignals } from "../signals.mjs"; +import { opportunities } from "../queue.mjs"; + +const window = { start: "2026-10-08T12:15:00.000Z", end: "2026-10-08T14:15:00.000Z" }; +const config = { cloudflare: { zones: ["aesthetic.computer"] } }; +const scope = edgeScope(config, window), zoneId = "a".repeat(32); +const auth = () => ({ email: "fixture@example.invalid", apiKey: "private-key" }); +const response = value => new Response(JSON.stringify(value)); +const group = (edge = 522, origin = 0, count = 8) => ({ count, dimensions: { + datetimeHour: "2026-10-08T13:00:00Z", clientRequestHTTPHost: "aesthetic.computer", edgeResponseStatus: edge, originResponseStatus: origin, +}, avg: { sampleInterval: 4 } }); +const result = rows => ({ data: { viewer: { zones: [{ httpRequestsAdaptiveGroups: rows }] } } }); +const coverage = [{ zone: "aesthetic.computer", status: "available", groups: 1, truncated: false, estimated: true }]; +const access = (errors = 3) => ({ status: "available", rows: [{ host: "aesthetic.computer", hour: "2026-10-08T13:00:00.000Z", requests: 30, errors5xx: errors }], scanned: 30, truncated: true }); + +test("edge scope rejects client zones and queries only allowed hosts in the exact window", () => { + assert.throws(() => edgeScope({ cloudflare: { zones: ["false.work"] } }, window), /scope/); + assert.throws(() => edgeScope({ cloudflare: { zones: ["unrelated.invalid"] } }, window), /scope/); + const query = edgeQuery(zoneId, ["aesthetic.computer"], scope); + assert.ok(query.includes(window.start) && query.includes(window.end)); + assert.match(query, /requestSource: "eyeball"/); + assert.doesNotMatch(query, /clientIP|clientRequestPath|rayName|userAgent|headers/); + assert.throws(() => edgeQuery(zoneId, ["false.work"], scope), /binding/); +}); + +test("Cloudflare estimates are not multiplied again and origin status zero is preserved", () => { + const out = edgeRows(result([group(), group(500, 500, 3), group(200, 503, 4)]), scope.hosts, scope); + assert.equal(out.rows[0].requests, 15); + assert.equal(out.rows[0].errors5xx, 11); + assert.equal(out.rows[0].origin5xx, 7); + assert.equal(out.rows[0].withoutOrigin5xx, 8); + assert.equal(out.rows[0].sampled, true); + const unknown = group(); unknown.dimensions.clientRequestHTTPHost = "private.example"; + assert.throws(() => edgeRows(result([unknown]), scope.hosts, scope), /aggregate/); + const future = group(); future.dimensions.datetimeHour = "2026-10-08T15:00:00Z"; + assert.throws(() => edgeRows(result([future]), scope.hosts, scope), /aggregate/); + assert.throws(() => edgeRows({ ...result([]), errors: [{ message: "partial result" }] }, scope.hosts, scope), /unavailable/); +}); + +test("zone binding, read-only requests, truncation and credential redaction hold", async () => { + const calls = []; + const out = await collectCloudflare({ ...scope, limit: 1 }, { auth, env: {}, fetchImpl: async (url, options) => { + calls.push({ url, options }); + assert.equal(options.redirect, "error"); + if (url.endsWith("/zones?name=aesthetic.computer")) return response({ success: true, result: [{ id: zoneId, name: "aesthetic.computer", status: "active" }] }); + assert.equal(url, "https://api.cloudflare.com/client/v4/graphql"); + assert.equal(options.method, "POST"); + assert.doesNotMatch(options.body, /private-key|fixture@example|mutation/); + return response(result([group(), group(200, 200, 20)])); + } }); + assert.equal(calls.length, 2); + assert.equal(out.coverage[0].truncated, true); + assert.doesNotMatch(JSON.stringify(out), /private-key|fixture@example/); + const mismatch = await collectCloudflare(scope, { auth, env: {}, fetchImpl: async () => response({ success: true, result: [{ id: zoneId, name: "false.work", status: "active" }] }) }); + assert.equal(mismatch.coverage[0].status, "unavailable"); + assert.deepEqual(mismatch.rows, []); +}); + +test("one inaccessible zone cannot erase another zone's evidence", async () => { + const two = edgeScope({ cloudflare: { zones: ["aesthetic.computer", "laklok.com"] } }, window); + const out = await collectCloudflare(two, { auth, env: {}, fetchImpl: async url => { + if (url.endsWith("name=laklok.com")) throw new Error("private-key raw error"); + if (url.includes("/zones?")) return response({ success: true, result: [{ id: zoneId, name: "aesthetic.computer", status: "active" }] }); + return response(result([group()])); + } }); + assert.deepEqual(out.coverage.map(row => row.status), ["available", "unavailable"]); + assert.equal(out.rows.length, 1); + assert.doesNotMatch(JSON.stringify(out), /private-key|raw error/); +}); + +test("correlation requires the same host and UTC hour and never auto-patches a timing overlap", () => { + const cf = { ...edgeRows(result([group(500, 500)]), scope.hosts, scope), coverage }; + const joined = correlateEdge(scope, cf, access()); + assert.equal(joined.correlation.buckets[0].assessment, "edge-and-lith-errors"); + assert.equal(joined.correlation.lith.truncated, true); + validateSignals(joined.signals, []); + assert.equal(opportunities({ runs: [], transitions: [], minimum: 3 }, joined.signals).length, 0); + const other = access(); other.rows[0].hour = "2026-10-08T12:00:00.000Z"; + assert.equal(correlateEdge(scope, cf, other).correlation.buckets.some(row => row.assessment === "edge-and-lith-errors"), false); + const anotherHost = access(); anotherHost.rows[0].host = "www.aesthetic.computer"; + assert.equal(correlateEdge(scope, cf, anotherHost).correlation.buckets.some(row => row.assessment === "edge-and-lith-errors"), false); +}); + +test("missing sources remain null rather than healthy zero counters", async () => { + const cf = { ...edgeRows(result([group()]), scope.hosts, scope), coverage }; + const missingLith = correlateEdge(scope, cf, { status: "unavailable", reason: "fixture-unavailable", rows: [] }); + assert.equal(missingLith.correlation.buckets[0].lith, null); + assert.equal(missingLith.signals.metrics.some(row => row.source === "lith-access"), false); + const missingCf = await collectEdge(config, window, { auth: () => ({}), env: {}, readAccess: async () => access(), fetchImpl: () => { throw Error("Unexpected request"); } }); + assert.equal(missingCf.correlation.buckets[0].edge, null); + assert.equal(missingCf.signals.metrics.some(row => row.source === "cloudflare"), false); + assert.equal(missingCf.signals.leads[0].source, "lith-access"); +}); + +test("Caddy rotation is read locally, filtered before export and bounded to exact timestamps", async t => { + const dir = await mkdtemp(join(tmpdir(), "aespatcher-access-")); t.after(() => rm(dir, { recursive: true, force: true })); + const row = (at, host, status) => JSON.stringify({ ts: Date.parse(at) / 1000, status, + request: { host, remote_ip: "192.0.2.1", uri: "/private?token=hidden", headers: { Authorization: ["secret"] } } }); + await writeFile(join(dir, "access-2026-10-08T12-30-00.000-size.log.gz"), gzipSync([ + row("2026-10-08T12:00:00Z", "aesthetic.computer", 500), row("2026-10-08T12:20:00Z", "aesthetic.computer", 200), "", + ].join("\n"))); + // Ensure the active file is newer, as on Lith. + await writeFile(join(dir, "access.log"), [row("2026-10-08T13:00:00Z", "aesthetic.computer:443", 503), + row("2026-10-08T13:30:00Z", "false.work", 500), row(window.end, "aesthetic.computer", 500), ""].join("\n")); + const out = validateAccess(await collectAccessLogs(scope, dir), scope); + assert.equal(out.scanned, 2); + assert.equal(out.rows.reduce((n, r) => n + r.errors5xx, 0), 1); + assert.equal(out.truncated, false); + assert.doesNotMatch(JSON.stringify(out), /false.work|192.0.2.1|private|hidden|Authorization|secret/); +}); + +test("retention gaps and malformed access rows make coverage incomplete", async t => { + const dir = await mkdtemp(join(tmpdir(), "aespatcher-access-")); t.after(() => rm(dir, { recursive: true, force: true })); + await writeFile(join(dir, "access.log"), JSON.stringify({ ts: Date.parse("2026-10-08T13:00:00Z") / 1000, status: 200, request: { host: "aesthetic.computer" } }) + "\nmalformed\n"); + const out = await collectAccessLogs(scope, dir); + assert.equal(out.truncated, true); + assert.equal(out.scanned, 1); + out.rows[0].ip = "private"; + assert.throws(() => validateAccess(out, scope), /aggregates/); +});