From fa1de7505d87c3945a7e148f29bb3f32df75a072 Mon Sep 17 00:00:00 2001 From: nafi <295082081+nafiisatu@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:56:08 +0100 Subject: [PATCH] Audit hash chain, IPFS pinning, dynamic limits, OTel Resolves #1581. Resolves #1583. Resolves #1584. Resolves #1585. --- backend/.env.example | 35 ++ backend/package.json | 5 + .../migration.sql | 92 +++ backend/prisma/schema.prisma | 63 ++ .../src/certificates/CertificateService.ts | 12 + backend/src/config/ipfs.ts | 114 ++++ backend/src/controllers/mediaController.ts | 94 +++ backend/src/db/auditChainStore.ts | 229 ++++++++ backend/src/index.ts | 23 + backend/src/instrumentation.ts | 277 +++++++++ backend/src/jobs/anchorMerkleRootJob.ts | 222 +++++++ backend/src/metrics/prometheusExporter.ts | 159 ++++++ backend/src/middleware/dynamicRateLimiter.ts | 129 +++++ backend/src/middleware/tracingMiddleware.ts | 100 ++++ backend/src/queues/compileQueue.ts | 7 +- backend/src/routes/audit.ts | 42 ++ backend/src/routes/index.ts | 2 + backend/src/routes/media.routes.ts | 16 + backend/src/routes/metrics.routes.ts | 3 +- backend/src/services/ipfsPinningService.ts | 540 ++++++++++++++++++ backend/src/services/reputationService.ts | 332 +++++++++++ .../src/services/tamperEvidentAuditService.ts | 296 ++++++++++ backend/src/utils/audit.ts | 15 + backend/src/utils/redis-pubsub.ts | 53 +- backend/src/utils/sentry.ts | 4 + backend/src/utils/slidingWindow.ts | 232 ++++++++ backend/src/workers/compileWorker.ts | 3 +- 27 files changed, 3082 insertions(+), 17 deletions(-) create mode 100644 backend/prisma/migrations/20261020000000_audit_chain_and_pinned_assets/migration.sql create mode 100644 backend/src/config/ipfs.ts create mode 100644 backend/src/controllers/mediaController.ts create mode 100644 backend/src/db/auditChainStore.ts create mode 100644 backend/src/instrumentation.ts create mode 100644 backend/src/jobs/anchorMerkleRootJob.ts create mode 100644 backend/src/metrics/prometheusExporter.ts create mode 100644 backend/src/middleware/dynamicRateLimiter.ts create mode 100644 backend/src/middleware/tracingMiddleware.ts create mode 100644 backend/src/routes/media.routes.ts create mode 100644 backend/src/services/ipfsPinningService.ts create mode 100644 backend/src/services/reputationService.ts create mode 100644 backend/src/services/tamperEvidentAuditService.ts create mode 100644 backend/src/utils/slidingWindow.ts diff --git a/backend/.env.example b/backend/.env.example index b7dec20dc..4e9810ade 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -265,3 +265,38 @@ ENCRYPTION_ROTATION_BATCH_SIZE=100 # Set to 'false' to hide the GET /api/v1/security/key-versions endpoint ENCRYPTION_EXPOSE_KEY_VERSIONS=true + +# ─── Tamper-evident audit chain (BE-009) ───────────────────────────────────── +# Start the in-process verification + Stellar anchoring schedulers +AUDIT_CHAIN_JOBS_ENABLED=false +AUDIT_ANCHOR_SECRET="" +AUDIT_ANCHOR_INTERVAL_MS=900000 +AUDIT_VERIFY_INTERVAL_MS=300000 +# Optional webhook (Slack/PagerDuty-compatible JSON) for critical tamper alarms +AUDIT_ALARM_WEBHOOK_URL="" + +# ─── IPFS / Filecoin pinning (BE-011) ──────────────────────────────────────── +PINATA_JWT="" +KUBO_API_URL="http://127.0.0.1:5001/api/v0/add" +WEB3STORAGE_TOKEN="" +IPFS_CDN_BASE_URL="https://cloudflare-ipfs.com" +IPFS_SECONDARY_GATEWAYS="https://gateway.pinata.cloud,https://w3s.link,https://ipfs.io" +IPFS_PIN_MAX_ATTEMPTS=3 +IPFS_PIN_BASE_DELAY_MS=150 +IPFS_PIN_MAX_DELAY_MS=2000 +IPFS_PIN_ATTEMPT_TIMEOUT_MS=2500 +IPFS_CERTIFICATE_PIN_SLA_MS=3000 + +# ─── Dynamic rate limiting (BE-012) ────────────────────────────────────────── +DYNAMIC_RATE_LIMIT_ENABLED=true +REPUTATION_CACHE_TTL_MS=60000 +# Stellar asset whose balance counts as an active stake +STAKING_ASSET_CODE="" +STAKING_ASSET_ISSUER="" +STAKING_MIN_BALANCE=1 + +# ─── OpenTelemetry tracing (BE-013) ────────────────────────────────────────── +OTEL_SDK_DISABLED=false +OTEL_SERVICE_NAME="web3-student-lab-backend" +# Jaeger OTLP/HTTP collector endpoint +OTEL_EXPORTER_JAEGER_ENDPOINT="http://localhost:4318/v1/traces" diff --git a/backend/package.json b/backend/package.json index 5991b8e31..3ab3364a6 100644 --- a/backend/package.json +++ b/backend/package.json @@ -32,6 +32,11 @@ "@aws-sdk/client-kms": "^3.1076.0", "@aws-sdk/client-s3": "^3.1117.0", "@aws-sdk/lib-storage": "^3.1076.0", + "@opentelemetry/api": "^1.9.1", + "@opentelemetry/exporter-trace-otlp-http": "^0.220.0", + "@opentelemetry/resources": "^2.10.0", + "@opentelemetry/sdk-node": "^0.220.0", + "@opentelemetry/semantic-conventions": "^1.43.0", "@prisma/adapter-pg": "^7.10.0", "@prisma/client": "^7.10.0", "@prisma/config": "^7.8.0", diff --git a/backend/prisma/migrations/20261020000000_audit_chain_and_pinned_assets/migration.sql b/backend/prisma/migrations/20261020000000_audit_chain_and_pinned_assets/migration.sql new file mode 100644 index 000000000..4b065ebe3 --- /dev/null +++ b/backend/prisma/migrations/20261020000000_audit_chain_and_pinned_assets/migration.sql @@ -0,0 +1,92 @@ +-- BE-011: immutable IPFS/Filecoin pins +CREATE TABLE "pinned_assets" ( + "id" TEXT NOT NULL, + "resourceType" TEXT NOT NULL, + "resourceId" TEXT NOT NULL, + "kind" TEXT NOT NULL, + "cid" TEXT NOT NULL, + "contentType" TEXT NOT NULL, + "sizeBytes" INTEGER NOT NULL, + "sha256" TEXT NOT NULL, + "provider" TEXT NOT NULL, + "filecoinProvider" TEXT, + "filecoinCid" TEXT, + "filecoinStatus" TEXT NOT NULL DEFAULT 'pending', + "cdnUrl" TEXT NOT NULL, + "gatewayVerified" BOOLEAN NOT NULL DEFAULT false, + "verifiedGateway" TEXT, + "verifiedAt" TIMESTAMP(3), + "pinLatencyMs" INTEGER NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "pinned_assets_pkey" PRIMARY KEY ("id") +); + +CREATE UNIQUE INDEX "pinned_assets_resourceType_resourceId_kind_key" ON "pinned_assets"("resourceType", "resourceId", "kind"); +CREATE INDEX "pinned_assets_cid_idx" ON "pinned_assets"("cid"); + +-- A pinned CID is content-addressed and must never be rewritten. +CREATE OR REPLACE FUNCTION pinned_assets_immutable_cid() RETURNS trigger AS $$ +BEGIN + IF NEW."cid" <> OLD."cid" OR NEW."sha256" <> OLD."sha256" THEN + RAISE EXCEPTION 'pinned_assets.cid is immutable'; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER "pinned_assets_immutable_cid" + BEFORE UPDATE ON "pinned_assets" + FOR EACH ROW EXECUTE FUNCTION pinned_assets_immutable_cid(); + +-- BE-009: tamper-evident audit hash chain +CREATE TABLE "audit_chain_entries" ( + "sequence" BIGINT NOT NULL, + "eventType" TEXT NOT NULL, + "actorId" TEXT, + "entity" TEXT, + "entityId" TEXT, + "payload" JSONB NOT NULL, + "prevHash" TEXT NOT NULL, + "entryHash" TEXT NOT NULL, + "createdAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "audit_chain_entries_pkey" PRIMARY KEY ("sequence") +); + +CREATE UNIQUE INDEX "audit_chain_entries_entryHash_key" ON "audit_chain_entries"("entryHash"); +CREATE INDEX "audit_chain_entries_createdAt_idx" ON "audit_chain_entries"("createdAt"); + +-- Defence in depth: the application never updates or deletes chain rows. +-- (A privileged attacker can drop this trigger; the hash chain and the +-- Stellar-anchored Merkle roots still expose any change they make.) +CREATE OR REPLACE FUNCTION audit_chain_entries_append_only() RETURNS trigger AS $$ +BEGIN + RAISE EXCEPTION 'audit_chain_entries is append-only'; +END; +$$ LANGUAGE plpgsql; + +CREATE TRIGGER "audit_chain_entries_append_only" + BEFORE UPDATE OR DELETE ON "audit_chain_entries" + FOR EACH ROW EXECUTE FUNCTION audit_chain_entries_append_only(); + +CREATE TABLE "audit_chain_anchors" ( + "id" TEXT NOT NULL, + "fromSequence" BIGINT NOT NULL, + "toSequence" BIGINT NOT NULL, + "entryCount" INTEGER NOT NULL, + "merkleRoot" TEXT NOT NULL, + "headHash" TEXT NOT NULL, + "network" TEXT NOT NULL, + "txHash" TEXT, + "status" TEXT NOT NULL DEFAULT 'pending', + "error" TEXT, + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "anchoredAt" TIMESTAMP(3), + + CONSTRAINT "audit_chain_anchors_pkey" PRIMARY KEY ("id") +); + +CREATE INDEX "audit_chain_anchors_toSequence_idx" ON "audit_chain_anchors"("toSequence"); +CREATE INDEX "audit_chain_anchors_status_idx" ON "audit_chain_anchors"("status"); diff --git a/backend/prisma/schema.prisma b/backend/prisma/schema.prisma index 4ae505623..0306afd74 100644 --- a/backend/prisma/schema.prisma +++ b/backend/prisma/schema.prisma @@ -612,3 +612,66 @@ model PaymentRecord { @@index([subscriptionId]) @@map("payment_records") } + +// Immutable IPFS/Filecoin pins for certificates and course media (BE-011) +model PinnedAsset { + id String @id @default(cuid()) + resourceType String + resourceId String + kind String + cid String + contentType String + sizeBytes Int + sha256 String + provider String + filecoinProvider String? + filecoinCid String? + filecoinStatus String @default("pending") // pending | stored | failed | disabled + cdnUrl String + gatewayVerified Boolean @default(false) + verifiedGateway String? + verifiedAt DateTime? + pinLatencyMs Int + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([resourceType, resourceId, kind]) + @@index([cid]) + @@map("pinned_assets") +} + +// Append-only SHA-256 hash chain of audit events (BE-009) +model AuditChainEntry { + sequence BigInt @id + eventType String + actorId String? + entity String? + entityId String? + payload Json + prevHash String + entryHash String @unique + createdAt DateTime + + @@index([createdAt]) + @@map("audit_chain_entries") +} + +// Merkle roots of audit chain ranges anchored on the Stellar ledger (BE-009) +model AuditChainAnchor { + id String @id @default(cuid()) + fromSequence BigInt + toSequence BigInt + entryCount Int + merkleRoot String + headHash String + network String + txHash String? + status String @default("pending") // pending | anchored | failed + error String? + createdAt DateTime @default(now()) + anchoredAt DateTime? + + @@index([toSequence]) + @@index([status]) + @@map("audit_chain_anchors") +} diff --git a/backend/src/certificates/CertificateService.ts b/backend/src/certificates/CertificateService.ts index 935ba78cb..f66b238ed 100644 --- a/backend/src/certificates/CertificateService.ts +++ b/backend/src/certificates/CertificateService.ts @@ -8,6 +8,7 @@ import prisma from '../db/index.js'; import { storageService } from '../services/storage/index.js'; +import { ipfsPinningService } from '../services/ipfsPinningService.js'; import { certificateBlockchainService } from '../blockchain/CertificateBlockchainService.js'; import { Certificate, @@ -208,6 +209,17 @@ export class CertificateService { txHash: mintResult.transactionHash, }); + // Auto-pin the certificate SVG + metadata to IPFS/Filecoin (BE-011). + // Runs alongside the email step so issuance latency is unaffected. + void ipfsPinningService + .autoPinCertificate({ + certificateId, + metadata: { ...(metadata as unknown as Record) }, + }) + .catch((pinError) => { + logger.error(`Failed to auto-pin certificate ${certificateId} to IPFS:`, pinError); + }); + // Send transactional email with PDF attachment and social sharing links try { const pdfBuffer = await certificateImageGenerator.generateCertificateImage({ diff --git a/backend/src/config/ipfs.ts b/backend/src/config/ipfs.ts new file mode 100644 index 000000000..11d96eca4 --- /dev/null +++ b/backend/src/config/ipfs.ts @@ -0,0 +1,114 @@ +/** + * IPFS / Filecoin pinning configuration — Issue BE-011 + * + * Environment variables: + * PINATA_JWT Pinata JWT (hot IPFS pinning, primary) + * PINATA_PIN_ENDPOINT default https://api.pinata.cloud/pinning/pinFileToIPFS + * KUBO_API_URL secondary IPFS pinning node (default http://127.0.0.1:5001/api/v0/add) + * WEB3STORAGE_TOKEN Web3.Storage token (Filecoin persistent storage deals) + * WEB3STORAGE_UPLOAD_ENDPOINT default https://api.web3.storage/upload + * IPFS_CDN_BASE_URL Cloudflare-fronted gateway serving active assets + * (default https://cloudflare-ipfs.com) + * IPFS_SECONDARY_GATEWAYS comma-separated fallback gateway base URLs + * IPFS_PIN_MAX_ATTEMPTS per-provider attempts before falling back (default 3) + * IPFS_PIN_BASE_DELAY_MS first backoff delay (default 150) + * IPFS_PIN_MAX_DELAY_MS backoff ceiling (default 2000) + * IPFS_PIN_ATTEMPT_TIMEOUT_MS per-request timeout (default 2500) + * IPFS_CERTIFICATE_PIN_SLA_MS certificate auto-pin deadline (default 3000) + */ + +export type PinningProviderType = 'pinata' | 'kubo' | 'web3storage'; + +export interface PinningProviderConfig { + name: string; + type: PinningProviderType; + endpoint: string; + /** Bearer token (Pinata / Web3.Storage) or basic-auth user (Kubo). */ + authToken?: string; + /** Basic-auth password (Kubo behind a proxy). */ + authSecret?: string; + /** + * `hot` — IPFS pinning used for low-latency retrieval (ordered fallback chain). + * `filecoin` — persistent storage with Filecoin deals (pinned in parallel). + */ + role: 'hot' | 'filecoin'; +} + +export interface RetryPolicy { + maxAttempts: number; + baseDelayMs: number; + maxDelayMs: number; + attemptTimeoutMs: number; +} + +export interface IpfsConfig { + providers: PinningProviderConfig[]; + /** CDN gateway base URL (Cloudflare) — first gateway tried and the public URL returned. */ + cdnBaseUrl: string; + /** Fallback gateway base URLs used when the CDN gateway is unreachable. */ + secondaryGateways: string[]; + retry: RetryPolicy; + /** Deadline for pinning a freshly issued certificate. */ + certificatePinSlaMs: number; +} + +function trimTrailingSlash(url: string): string { + return url.replace(/\/+$/, ''); +} + +function numberFromEnv(value: string | undefined, fallback: number): number { + const parsed = Number(value); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +export function loadIpfsConfig(env: NodeJS.ProcessEnv = process.env): IpfsConfig { + const providers: PinningProviderConfig[] = []; + + if (env.PINATA_JWT) { + providers.push({ + name: 'Pinata', + type: 'pinata', + endpoint: env.PINATA_PIN_ENDPOINT ?? 'https://api.pinata.cloud/pinning/pinFileToIPFS', + authToken: env.PINATA_JWT, + role: 'hot', + }); + } + + providers.push({ + name: 'Kubo', + type: 'kubo', + endpoint: env.KUBO_API_URL ?? 'http://127.0.0.1:5001/api/v0/add', + ...(env.KUBO_API_USER ? { authToken: env.KUBO_API_USER, authSecret: env.KUBO_API_PASSWORD ?? '' } : {}), + role: 'hot', + }); + + if (env.WEB3STORAGE_TOKEN) { + providers.push({ + name: 'Web3.Storage', + type: 'web3storage', + endpoint: env.WEB3STORAGE_UPLOAD_ENDPOINT ?? 'https://api.web3.storage/upload', + authToken: env.WEB3STORAGE_TOKEN, + role: 'filecoin', + }); + } + + const secondaryGateways = (env.IPFS_SECONDARY_GATEWAYS ?? 'https://gateway.pinata.cloud,https://w3s.link,https://ipfs.io') + .split(',') + .map((url) => trimTrailingSlash(url.trim())) + .filter(Boolean); + + return { + providers, + cdnBaseUrl: trimTrailingSlash(env.IPFS_CDN_BASE_URL ?? 'https://cloudflare-ipfs.com'), + secondaryGateways, + retry: { + maxAttempts: numberFromEnv(env.IPFS_PIN_MAX_ATTEMPTS, 3), + baseDelayMs: numberFromEnv(env.IPFS_PIN_BASE_DELAY_MS, 150), + maxDelayMs: numberFromEnv(env.IPFS_PIN_MAX_DELAY_MS, 2_000), + attemptTimeoutMs: numberFromEnv(env.IPFS_PIN_ATTEMPT_TIMEOUT_MS, 2_500), + }, + certificatePinSlaMs: numberFromEnv(env.IPFS_CERTIFICATE_PIN_SLA_MS, 3_000), + }; +} + +export const ipfsConfig = loadIpfsConfig(); diff --git a/backend/src/controllers/mediaController.ts b/backend/src/controllers/mediaController.ts new file mode 100644 index 000000000..43933a867 --- /dev/null +++ b/backend/src/controllers/mediaController.ts @@ -0,0 +1,94 @@ +import { Request, Response } from 'express'; +import { asyncHandler } from '../middleware/errorHandler.js'; +import { ipfsPinningService, isCidV1 } from '../services/ipfsPinningService.js'; +import logger from '../utils/logger.js'; +import { ApiResponse } from '../utils/response.js'; + +/** Upper bound for base64 uploads through the JSON API (course media should use storage uploads). */ +const MAX_PIN_BYTES = 10 * 1024 * 1024; +const RESOURCE_TYPES = new Set(['certificate', 'course', 'lesson', 'media']); + +export const mediaController = { + // Pin an asset (base64 payload) to IPFS + Filecoin and record its CID + pinAsset: asyncHandler(async (req: Request, res: Response) => { + const { data, name, contentType, resourceType, resourceId, kind } = req.body ?? {}; + + if (typeof data !== 'string' || !data) { + return res.status(400).json(ApiResponse.error('`data` (base64 content) is required')); + } + if (typeof name !== 'string' || typeof contentType !== 'string' || typeof kind !== 'string') { + return res.status(400).json(ApiResponse.error('`name`, `contentType` and `kind` are required')); + } + if (!RESOURCE_TYPES.has(resourceType) || typeof resourceId !== 'string' || !resourceId) { + return res.status(400).json(ApiResponse.error('Invalid `resourceType` or `resourceId`')); + } + + const content = Buffer.from(data, 'base64'); + if (content.length === 0 || content.length > MAX_PIN_BYTES) { + return res.status(413).json(ApiResponse.error(`Content must be between 1 byte and ${MAX_PIN_BYTES} bytes`)); + } + + try { + const record = await ipfsPinningService.pinAsset({ content, name, contentType, resourceType, resourceId, kind }); + return res.status(201).json(ApiResponse.success('Asset pinned', record)); + } catch (error) { + logger.error('Media pinning failed', { + resourceType, + resourceId, + error: error instanceof Error ? error.message : String(error), + }); + return res.status(502).json(ApiResponse.error('All IPFS pinning providers failed')); + } + }), + + // Resolve a CID to its stored records and gateway URLs (optionally redirect to the CDN) + getByCid: asyncHandler(async (req: Request, res: Response) => { + const cid = String(req.params.cid); + if (!isCidV1(cid)) { + return res.status(400).json(ApiResponse.error('Malformed CID: a CIDv1 (base32) is required')); + } + + const records = await ipfsPinningService.findByCid(cid); + if (records.length === 0) { + return res.status(404).json(ApiResponse.error('CID is not tracked by this service')); + } + + if (req.query.redirect === 'true') { + return res.redirect(302, ipfsPinningService.cdnUrl(cid)); + } + + return res.status(200).json( + ApiResponse.success('Pinned asset retrieved', { + cid, + cdnUrl: ipfsPinningService.cdnUrl(cid), + gateways: ipfsPinningService.gatewayUrls(cid), + records, + }), + ); + }), + + // Check gateway accessibility (CDN first, then secondary gateways) and persist the result + verifyCid: asyncHandler(async (req: Request, res: Response) => { + const cid = String(req.params.cid); + if (!isCidV1(cid)) { + return res.status(400).json(ApiResponse.error('Malformed CID: a CIDv1 (base32) is required')); + } + + const [record] = await ipfsPinningService.findByCid(cid); + const check = record + ? await ipfsPinningService.reverify(record) + : await ipfsPinningService.verifyGatewayAccessibility(cid); + + return res.status(check.accessible ? 200 : 503).json(ApiResponse.success('Gateway accessibility checked', check)); + }), + + // List the immutable CIDs pinned for a certificate + getCertificateAssets: asyncHandler(async (req: Request, res: Response) => { + const certificateId = String(req.params.certificateId); + const records = await ipfsPinningService.findByResource('certificate', certificateId); + if (records.length === 0) { + return res.status(404).json(ApiResponse.error('No pinned assets for this certificate')); + } + return res.status(200).json(ApiResponse.success('Certificate assets retrieved', records)); + }), +}; diff --git a/backend/src/db/auditChainStore.ts b/backend/src/db/auditChainStore.ts new file mode 100644 index 000000000..87815b81b --- /dev/null +++ b/backend/src/db/auditChainStore.ts @@ -0,0 +1,229 @@ +/** + * Audit Chain Store — Issue BE-009 + * + * Persistence for the tamper-evident audit hash chain (`audit_chain_entries`) + * and the Merkle roots anchored on Stellar (`audit_chain_anchors`). + * + * Appends are serialised with a PostgreSQL transaction-scoped advisory lock so + * that concurrent writers (multiple API instances) always link to the true + * chain head: read head → build entry → insert all happen under the lock. + * Sequences are assigned by the application (head + 1), never by a database + * sequence, so a deleted row always leaves a detectable gap. + * + * An in-memory implementation is provided for tests and local development. + */ + +import prisma from './index.js'; + +export interface AuditChainRecord { + sequence: bigint; + eventType: string; + actorId: string | null; + entity: string | null; + entityId: string | null; + payload: unknown; + prevHash: string; + entryHash: string; + createdAt: Date; +} + +export type AnchorStatus = 'pending' | 'anchored' | 'failed'; + +export interface AuditAnchorRecord { + id: string; + fromSequence: bigint; + toSequence: bigint; + entryCount: number; + merkleRoot: string; + headHash: string; + network: string; + txHash: string | null; + status: AnchorStatus; + error: string | null; + createdAt: Date; + anchoredAt: Date | null; +} + +export type NewAnchor = Pick< + AuditAnchorRecord, + 'fromSequence' | 'toSequence' | 'entryCount' | 'merkleRoot' | 'headHash' | 'network' +>; + +export interface AuditChainStore { + /** + * Append one entry atomically. `build` receives the current head (or null + * for an empty chain) and returns the fully-hashed record to insert. + */ + append(build: (head: AuditChainRecord | null) => AuditChainRecord): Promise; + getHead(): Promise; + getEntry(sequence: bigint): Promise; + /** Entries with `sequence > afterSequence`, ascending, at most `limit`. */ + readBatch(afterSequence: bigint, limit: number): Promise; + /** Entry hashes in `[fromSequence, toSequence]`, ascending by sequence. */ + readHashes(fromSequence: bigint, toSequence: bigint): Promise>; + + createAnchor(anchor: NewAnchor): Promise; + updateAnchor( + id: string, + patch: Partial>, + ): Promise; + getAnchor(id: string): Promise; + /** Latest successfully anchored range. */ + getLatestAnchor(): Promise; + listAnchors(limit: number): Promise; +} + +const CHAIN_LOCK_KEY = 'audit_chain_entries'; + +// Prisma's generated delegates are reached through `any` so this module does +// not depend on the regenerated client shape at type-check time. +const db = prisma as any; + +function toAnchor(row: any): AuditAnchorRecord { + return { ...row, status: row.status as AnchorStatus }; +} + +export class PrismaAuditChainStore implements AuditChainStore { + async append(build: (head: AuditChainRecord | null) => AuditChainRecord): Promise { + return db.$transaction(async (tx: any) => { + await tx.$executeRawUnsafe('SELECT pg_advisory_xact_lock(hashtext($1))', CHAIN_LOCK_KEY); + const head: AuditChainRecord | null = await tx.auditChainEntry.findFirst({ orderBy: { sequence: 'desc' } }); + const entry = build(head); + return tx.auditChainEntry.create({ data: entry }); + }); + } + + getHead(): Promise { + return db.auditChainEntry.findFirst({ orderBy: { sequence: 'desc' } }); + } + + getEntry(sequence: bigint): Promise { + return db.auditChainEntry.findUnique({ where: { sequence } }); + } + + readBatch(afterSequence: bigint, limit: number): Promise { + return db.auditChainEntry.findMany({ + where: { sequence: { gt: afterSequence } }, + orderBy: { sequence: 'asc' }, + take: limit, + }); + } + + readHashes(fromSequence: bigint, toSequence: bigint): Promise> { + return db.auditChainEntry.findMany({ + where: { sequence: { gte: fromSequence, lte: toSequence } }, + orderBy: { sequence: 'asc' }, + select: { sequence: true, entryHash: true }, + }); + } + + async createAnchor(anchor: NewAnchor): Promise { + return toAnchor(await db.auditChainAnchor.create({ data: { ...anchor, status: 'pending' } })); + } + + async updateAnchor( + id: string, + patch: Partial>, + ): Promise { + return toAnchor(await db.auditChainAnchor.update({ where: { id }, data: patch })); + } + + async getAnchor(id: string): Promise { + const row = await db.auditChainAnchor.findUnique({ where: { id } }); + return row ? toAnchor(row) : null; + } + + async getLatestAnchor(): Promise { + const row = await db.auditChainAnchor.findFirst({ + where: { status: 'anchored' }, + orderBy: { toSequence: 'desc' }, + }); + return row ? toAnchor(row) : null; + } + + async listAnchors(limit: number): Promise { + const rows = await db.auditChainAnchor.findMany({ orderBy: { createdAt: 'desc' }, take: limit }); + return rows.map(toAnchor); + } +} + +export class InMemoryAuditChainStore implements AuditChainStore { + readonly entries: AuditChainRecord[] = []; + readonly anchors: AuditAnchorRecord[] = []; + private queue: Promise = Promise.resolve(); + + append(build: (head: AuditChainRecord | null) => AuditChainRecord): Promise { + // Serialise appends the same way the advisory lock does in PostgreSQL. + const run = this.queue.then(() => { + const entry = build(this.entries[this.entries.length - 1] ?? null); + this.entries.push({ ...entry }); + return entry; + }); + this.queue = run.catch(() => undefined); + return run; + } + + async getHead(): Promise { + return this.entries[this.entries.length - 1] ?? null; + } + + async getEntry(sequence: bigint): Promise { + return this.entries.find((e) => e.sequence === sequence) ?? null; + } + + async readBatch(afterSequence: bigint, limit: number): Promise { + return this.entries + .filter((e) => e.sequence > afterSequence) + .sort((a, b) => (a.sequence < b.sequence ? -1 : 1)) + .slice(0, limit); + } + + async readHashes(fromSequence: bigint, toSequence: bigint): Promise> { + return this.entries + .filter((e) => e.sequence >= fromSequence && e.sequence <= toSequence) + .sort((a, b) => (a.sequence < b.sequence ? -1 : 1)) + .map(({ sequence, entryHash }) => ({ sequence, entryHash })); + } + + async createAnchor(anchor: NewAnchor): Promise { + const record: AuditAnchorRecord = { + ...anchor, + id: `anchor-${this.anchors.length + 1}`, + txHash: null, + status: 'pending', + error: null, + createdAt: new Date(), + anchoredAt: null, + }; + this.anchors.push(record); + return record; + } + + async updateAnchor( + id: string, + patch: Partial>, + ): Promise { + const anchor = this.anchors.find((a) => a.id === id); + if (!anchor) throw new Error(`Unknown anchor ${id}`); + Object.assign(anchor, patch); + return anchor; + } + + async getAnchor(id: string): Promise { + return this.anchors.find((a) => a.id === id) ?? null; + } + + async getLatestAnchor(): Promise { + return ( + this.anchors + .filter((a) => a.status === 'anchored') + .sort((a, b) => (a.toSequence < b.toSequence ? 1 : -1))[0] ?? null + ); + } + + async listAnchors(limit: number): Promise { + return [...this.anchors].reverse().slice(0, limit); + } +} + +export const auditChainStore: AuditChainStore = new PrismaAuditChainStore(); diff --git a/backend/src/index.ts b/backend/src/index.ts index c5a981764..fd9f83154 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -1,13 +1,17 @@ +import { startTracing } from './instrumentation.js'; import dotenv from 'dotenv'; import express, { Request, Response } from 'express'; import { createCorsMiddleware } from './config/cors.config.js'; import swaggerDocsRouter from './config/swagger.serve.js'; import { jsonBodySizeLimit } from './middleware/bodySizeLimit.js'; import { dbRoutingMiddleware } from './middleware/dbRouting.js'; +import { dynamicRateLimiter } from './middleware/dynamicRateLimiter.js'; +import { tracingMiddleware } from './middleware/tracingMiddleware.js'; import apiRouter from './routes/api.js'; import { livenessHandler, readinessHandler } from './routes/health.routes.js'; import routes from './routes/index.js'; import { runStartupMigrations } from './db/startupMigrations.js'; +import { startAuditChainJobs } from './jobs/anchorMerkleRootJob.js'; import logger from './utils/logger.js'; import { getSentryErrorHandler, getSentryRequestHandler, initializeSentry } from './utils/sentry.js'; @@ -16,9 +20,15 @@ dotenv.config(); export const app = express(); const port = process.env.PORT || 8080; +// Start OpenTelemetry (W3C trace-context, Jaeger export) before serving traffic +void startTracing(); + // Initialize Sentry Telemetry and Distributed Tracing initializeSentry(app); +// OpenTelemetry SERVER span per request; downstream work becomes child spans +app.use(tracingMiddleware()); + // Mount Sentry Request Handler middleware app.use(getSentryRequestHandler()); @@ -64,6 +74,15 @@ app.post('/api/security/csp-report', express.json(), (req: Request, res: Respons res.status(204).end(); }); +// Reputation-aware sliding-window rate limiting for every API route. +// Set DYNAMIC_RATE_LIMIT_ENABLED=false to disable. +app.use( + '/api', + dynamicRateLimiter({ + skip: () => process.env.NODE_ENV === 'test' || process.env.DYNAMIC_RATE_LIMIT_ENABLED === 'false', + }) +); + // Mount OpenAPI 3.1 interactive docs (Swagger UI) + JSON/YAML spec export. app.use('/api/docs', swaggerDocsRouter); @@ -82,6 +101,10 @@ if (process.env.NODE_ENV !== 'test') { app.listen(port, () => { console.log(`Server is running on port ${port}`); }); + // Routine audit hash-chain verification + Stellar Merkle-root anchoring + if (process.env.AUDIT_CHAIN_JOBS_ENABLED === 'true') { + startAuditChainJobs(); + } }) .catch((error) => { logger.error('refusing to serve traffic until migrations succeed', error); diff --git a/backend/src/instrumentation.ts b/backend/src/instrumentation.ts new file mode 100644 index 000000000..b5eb964ef --- /dev/null +++ b/backend/src/instrumentation.ts @@ -0,0 +1,277 @@ +/** + * OpenTelemetry Instrumentation — Issue BE-013 + * + * Boots `@opentelemetry/sdk-node` and exports the helpers used to propagate + * W3C trace-context (`traceparent` / `tracestate`) across process boundaries: + * + * - HTTP: `tracingMiddleware` extracts the inbound context and opens a + * SERVER span for every request. + * - BullMQ: producers attach `withTraceContext(data)` to job payloads and + * workers wrap their processor in `traceJob(...)`, so a job span is a child + * of the HTTP request that enqueued it. + * - Redis Pub/Sub: publishers embed `injectTraceContext()` in the message + * envelope and subscribers run handlers in `runInSpan(..., { parent })`. + * + * Spans are exported over OTLP/HTTP to Jaeger (Jaeger ≥1.35 ingests OTLP + * natively on :4318). Configure with: + * OTEL_SERVICE_NAME (default: web3-student-lab-backend) + * OTEL_EXPORTER_JAEGER_ENDPOINT (default: http://localhost:4318/v1/traces) + * OTEL_TRACES_SAMPLER / _ARG (standard OTel sampler env vars) + * OTEL_SDK_DISABLED=true (turn tracing off entirely) + * + * When the SDK is not started the OpenTelemetry API is a no-op, so every + * helper below is safe to call unconditionally. + */ + +import { + context, + propagation, + ROOT_CONTEXT, + SpanKind, + SpanStatusCode, + trace, + type Attributes, + type Context, + type Span, +} from '@opentelemetry/api'; +import { queueJobDuration } from './metrics/prometheusExporter.js'; +import logger from './utils/logger.js'; + +export const SERVICE_NAME = process.env.OTEL_SERVICE_NAME || 'web3-student-lab-backend'; + +export const tracer = trace.getTracer(SERVICE_NAME); + +/** Serialised W3C trace-context headers carried inside job / message payloads. */ +export type TraceCarrier = Record; + +interface ShutdownableSdk { + shutdown(): Promise; +} + +let sdk: ShutdownableSdk | null = null; +let starting: Promise | null = null; + +/** + * Whether the OpenTelemetry SDK should own the global tracer provider. Sentry + * reads this to skip its own OpenTelemetry setup so the two never compete. + */ +export function isTracingEnabled(): boolean { + return process.env.OTEL_SDK_DISABLED !== 'true' && process.env.NODE_ENV !== 'test'; +} + +/** + * Start the OpenTelemetry Node SDK. Idempotent; resolves to `true` once the + * SDK is running and `false` when tracing is disabled or failed to load. + */ +export function startTracing(): Promise { + if (starting) return starting; + + starting = (async () => { + if (!isTracingEnabled()) { + return false; + } + + try { + const [{ NodeSDK }, { OTLPTraceExporter }, { resourceFromAttributes }, semconv] = await Promise.all([ + import('@opentelemetry/sdk-node'), + import('@opentelemetry/exporter-trace-otlp-http'), + import('@opentelemetry/resources'), + import('@opentelemetry/semantic-conventions'), + ]); + + const instance = new NodeSDK({ + resource: resourceFromAttributes({ + [semconv.ATTR_SERVICE_NAME]: SERVICE_NAME, + [semconv.ATTR_SERVICE_VERSION]: process.env.npm_package_version || '1.0.0', + 'deployment.environment.name': process.env.NODE_ENV || 'development', + }), + traceExporter: new OTLPTraceExporter({ + url: + process.env.OTEL_EXPORTER_JAEGER_ENDPOINT || + process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT || + 'http://localhost:4318/v1/traces', + }), + }); + + instance.start(); + sdk = instance; + + process.once('SIGTERM', () => { + void shutdownTracing(); + }); + + logger.info('OpenTelemetry tracing started', { service: SERVICE_NAME }); + return true; + } catch (error) { + logger.warn('OpenTelemetry tracing could not be started; continuing without it', { + error: error instanceof Error ? error.message : String(error), + }); + return false; + } + })(); + + return starting; +} + +/** Flush pending spans and stop the SDK. */ +export async function shutdownTracing(): Promise { + if (!sdk) return; + const instance = sdk; + sdk = null; + starting = null; + try { + await instance.shutdown(); + } catch (error) { + logger.warn('OpenTelemetry shutdown failed', { + error: error instanceof Error ? error.message : String(error), + }); + } +} + +// ─── Context propagation ───────────────────────────────────────────────────── + +/** Serialise the given (default: active) context into W3C headers. */ +export function injectTraceContext(ctx: Context = context.active()): TraceCarrier { + const carrier: TraceCarrier = {}; + propagation.inject(ctx, carrier); + return carrier; +} + +/** Rebuild a context from W3C headers; falls back to the root context. */ +export function extractTraceContext(carrier?: TraceCarrier | null): Context { + if (!carrier) return ROOT_CONTEXT; + return propagation.extract(ROOT_CONTEXT, carrier); +} + +export interface RunInSpanOptions { + kind?: SpanKind; + attributes?: Attributes; + /** Explicit parent context; defaults to the active context. */ + parent?: Context; +} + +function failSpan(span: Span, error: unknown): void { + span.recordException(error instanceof Error ? error : new Error(String(error))); + span.setStatus({ + code: SpanStatusCode.ERROR, + message: error instanceof Error ? error.message : String(error), + }); +} + +/** + * Run `fn` inside a new active span. Works for sync and async callbacks: the + * span ends when the returned promise settles (or immediately for sync work) + * and errors are recorded on the span before being re-thrown. + */ +export function runInSpan(name: string, options: RunInSpanOptions, fn: (span: Span) => T): T { + const parent = options.parent ?? context.active(); + const span = tracer.startSpan( + name, + { kind: options.kind ?? SpanKind.INTERNAL, attributes: options.attributes ?? {} }, + parent, + ); + + return context.with(trace.setSpan(parent, span), () => { + let result: T; + try { + result = fn(span); + } catch (error) { + failSpan(span, error); + span.end(); + throw error; + } + + if (result instanceof Promise) { + return result.then( + (value) => { + span.end(); + return value; + }, + (error: unknown) => { + failSpan(span, error); + span.end(); + throw error; + }, + ) as T; + } + + span.end(); + return result; + }); +} + +// ─── BullMQ ────────────────────────────────────────────────────────────────── + +/** Job payload field that carries the producer's trace context. */ +export const JOB_TRACE_FIELD = '_traceContext'; + +export type TracedJobData = T & { [JOB_TRACE_FIELD]?: TraceCarrier }; + +/** Attach the active trace context to a BullMQ job payload. */ +export function withTraceContext(data: T): TracedJobData { + return { ...data, [JOB_TRACE_FIELD]: injectTraceContext() }; +} + +/** Wrap a `queue.add(...)` call in a PRODUCER span and inject its context. */ +export function traceEnqueue( + queueName: string, + data: T, + add: (tracedData: TracedJobData) => Promise, +): Promise { + return runInSpan( + `${queueName} publish`, + { + kind: SpanKind.PRODUCER, + attributes: { 'messaging.system': 'bullmq', 'messaging.destination.name': queueName }, + }, + () => add(withTraceContext(data)), + ); +} + +interface TraceableJob { + id?: string | undefined; + name: string; + attemptsMade: number; + data: unknown; +} + +/** + * Execute a BullMQ processor as a CONSUMER span whose parent is the context + * the producer attached to the job, linking the job back to its HTTP request. + */ +export async function traceJob(queueName: string, job: TraceableJob, processor: () => Promise): Promise { + const carrier = + job.data && typeof job.data === 'object' + ? ((job.data as Record)[JOB_TRACE_FIELD] as TraceCarrier | undefined) + : undefined; + + const startedAt = process.hrtime.bigint(); + let outcome = 'completed'; + try { + return await runInSpan( + `${queueName} process`, + { + kind: SpanKind.CONSUMER, + parent: extractTraceContext(carrier), + attributes: { + 'messaging.system': 'bullmq', + 'messaging.destination.name': queueName, + 'messaging.operation.name': job.name, + 'messaging.message.id': job.id ?? '', + 'bullmq.job.attempts_made': job.attemptsMade, + }, + }, + processor, + ); + } catch (error) { + outcome = 'failed'; + throw error; + } finally { + queueJobDuration.observe( + { queue: queueName, outcome }, + Number(process.hrtime.bigint() - startedAt) / 1e9, + ); + } +} + +export { SpanKind, SpanStatusCode }; diff --git a/backend/src/jobs/anchorMerkleRootJob.ts b/backend/src/jobs/anchorMerkleRootJob.ts new file mode 100644 index 000000000..f5328eff9 --- /dev/null +++ b/backend/src/jobs/anchorMerkleRootJob.ts @@ -0,0 +1,222 @@ +/** + * Anchor Merkle Root Job — Issue BE-009 + * + * Periodically: + * 1. verifies the audit hash chain end-to-end (refusing to anchor a chain + * that already fails verification — that raises the critical alarm); + * 2. computes the Merkle root of every entry appended since the last anchor; + * 3. publishes it on Stellar as the transaction's 32-byte `MEMO_HASH` + * (plus a `manageData` entry `w3sl_audit_root` holding the same bytes); + * 4. records the anchor so later verifications can compare against it. + * + * `verifyPublishedAnchor()` fetches the transaction back from Horizon and + * checks the on-chain memo against a fresh local recomputation. + * + * Environment: + * AUDIT_ANCHOR_SECRET signing key (falls back to STELLAR_ISSUER_SECRET) + * AUDIT_ANCHOR_INTERVAL_MS schedule interval (default 15 minutes) + * AUDIT_VERIFY_INTERVAL_MS chain verification interval (default 5 minutes) + * STELLAR_HORIZON_URL / STELLAR_NETWORK ('public' | 'testnet') + */ + +import { BASE_FEE, Horizon, Keypair, Memo, Networks, Operation, TransactionBuilder } from '@stellar/stellar-sdk'; +import { auditChainStore, type AuditAnchorRecord, type AuditChainStore } from '../db/auditChainStore.js'; +import { + tamperEvidentAuditService, + type TamperEvidentAuditService, +} from '../services/tamperEvidentAuditService.js'; +import logger from '../utils/logger.js'; + +export const ANCHOR_DATA_KEY = 'w3sl_audit_root'; + +export interface AnchorSubmitter { + readonly network: string; + /** Publish the 32-byte root on-chain; resolves to the transaction hash. */ + submit(merkleRootHex: string): Promise; + /** Fetch the root published by `txHash` (hex), or null if absent. */ + fetchPublishedRoot(txHash: string): Promise; +} + +export class StellarAnchorSubmitter implements AnchorSubmitter { + readonly network: string; + private readonly server: Horizon.Server; + private readonly passphrase: string; + + constructor( + private readonly secret = process.env.AUDIT_ANCHOR_SECRET || process.env.STELLAR_ISSUER_SECRET || '', + horizonUrl = process.env.STELLAR_HORIZON_URL || 'https://horizon-testnet.stellar.org', + ) { + const isPublic = (process.env.STELLAR_NETWORK || 'testnet').toLowerCase() === 'public'; + this.network = isPublic ? 'public' : 'testnet'; + this.passphrase = isPublic ? Networks.PUBLIC : Networks.TESTNET; + this.server = new Horizon.Server(horizonUrl); + } + + async submit(merkleRootHex: string): Promise { + if (!this.secret) { + throw new Error('AUDIT_ANCHOR_SECRET (or STELLAR_ISSUER_SECRET) is not configured'); + } + const keypair = Keypair.fromSecret(this.secret); + const account = await this.server.loadAccount(keypair.publicKey()); + const root = Buffer.from(merkleRootHex, 'hex'); + + const tx = new TransactionBuilder(account, { fee: BASE_FEE, networkPassphrase: this.passphrase }) + .addOperation(Operation.manageData({ name: ANCHOR_DATA_KEY, value: root })) + .addMemo(Memo.hash(root)) + .setTimeout(60) + .build(); + tx.sign(keypair); + + const response = await this.server.submitTransaction(tx); + return response.hash; + } + + async fetchPublishedRoot(txHash: string): Promise { + const tx = await this.server.transactions().transaction(txHash).call(); + if (tx.memo_type !== 'hash' || !tx.memo) return null; + return Buffer.from(tx.memo, 'base64').toString('hex'); + } +} + +export interface AnchorJobDeps { + service?: TamperEvidentAuditService; + store?: AuditChainStore; + submitter?: AnchorSubmitter; +} + +export type AnchorJobResult = + | { status: 'skipped'; reason: string } + | { status: 'anchored'; anchor: AuditAnchorRecord } + | { status: 'failed'; reason: string; anchor?: AuditAnchorRecord }; + +let defaultSubmitter: AnchorSubmitter | null = null; +function getDefaultSubmitter(): AnchorSubmitter { + defaultSubmitter ??= new StellarAnchorSubmitter(); + return defaultSubmitter; +} + +/** Anchor every entry appended since the previous anchor. */ +export async function runAnchorMerkleRootJob(deps: AnchorJobDeps = {}): Promise { + const service = deps.service ?? tamperEvidentAuditService; + const store = deps.store ?? auditChainStore; + const submitter = deps.submitter ?? getDefaultSubmitter(); + + const verification = await service.verifyChain(); + if (!verification.valid) { + return { status: 'failed', reason: 'Audit chain failed verification; refusing to anchor a tampered chain' }; + } + + const head = await store.getHead(); + const previous = await store.getLatestAnchor(); + const fromSequence = previous ? previous.toSequence + 1n : 1n; + + if (!head || head.sequence < fromSequence) { + return { status: 'skipped', reason: 'No new audit entries since the last anchor' }; + } + + const { root, count } = await service.computeRangeRoot(fromSequence, head.sequence); + const anchor = await store.createAnchor({ + fromSequence, + toSequence: head.sequence, + entryCount: count, + merkleRoot: root, + headHash: head.entryHash, + network: submitter.network, + }); + + try { + const txHash = await submitter.submit(root); + const anchored = await store.updateAnchor(anchor.id, { txHash, status: 'anchored', anchoredAt: new Date() }); + logger.info('Audit chain Merkle root anchored on Stellar', { + fromSequence: fromSequence.toString(), + toSequence: head.sequence.toString(), + merkleRoot: root, + txHash, + }); + return { status: 'anchored', anchor: anchored }; + } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + const failed = await store.updateAnchor(anchor.id, { status: 'failed', error: reason.slice(0, 500) }); + logger.error('Failed to anchor audit chain Merkle root', { reason, anchorId: anchor.id }); + return { status: 'failed', reason, anchor: failed }; + } +} + +export interface AnchorVerification { + anchorId: string; + txHash: string | null; + localRoot: string; + storedRoot: string; + publishedRoot: string | null; + matches: boolean; +} + +/** Confirm the root published on Stellar equals a fresh local computation. */ +export async function verifyPublishedAnchor( + anchor: AuditAnchorRecord, + deps: Omit = {}, +): Promise { + const service = deps.service ?? tamperEvidentAuditService; + const submitter = deps.submitter ?? getDefaultSubmitter(); + + const { root: localRoot } = await service.computeRangeRoot(anchor.fromSequence, anchor.toSequence); + const publishedRoot = anchor.txHash ? await submitter.fetchPublishedRoot(anchor.txHash) : null; + + return { + anchorId: anchor.id, + txHash: anchor.txHash, + localRoot, + storedRoot: anchor.merkleRoot, + publishedRoot, + matches: publishedRoot !== null && publishedRoot === localRoot && localRoot === anchor.merkleRoot, + }; +} + +/** + * Start the in-process schedulers: routine chain verification and periodic + * anchoring. Returns a stop function. Timers are unref'd so they never keep + * the process alive on shutdown. + */ +export function startAuditChainJobs(deps: AnchorJobDeps = {}): () => void { + const service = deps.service ?? tamperEvidentAuditService; + const anchorEveryMs = Number(process.env.AUDIT_ANCHOR_INTERVAL_MS || 15 * 60_000); + const verifyEveryMs = Number(process.env.AUDIT_VERIFY_INTERVAL_MS || 5 * 60_000); + + let anchoring = false; + const anchorTimer = setInterval(() => { + if (anchoring) return; + anchoring = true; + runAnchorMerkleRootJob(deps) + .catch((error) => logger.error('Audit anchor job crashed', { error: String(error) })) + .finally(() => { + anchoring = false; + }); + }, anchorEveryMs); + + let verifying = false; + const verifyTimer = setInterval(() => { + if (verifying) return; + verifying = true; + service + .verifyChain() + .then((result) => + logger.info('Audit chain verification completed', { + valid: result.valid, + entriesChecked: result.entriesChecked, + durationMs: result.durationMs, + }), + ) + .catch((error) => logger.error('Audit chain verification crashed', { error: String(error) })) + .finally(() => { + verifying = false; + }); + }, verifyEveryMs); + + anchorTimer.unref(); + verifyTimer.unref(); + + return () => { + clearInterval(anchorTimer); + clearInterval(verifyTimer); + }; +} diff --git a/backend/src/metrics/prometheusExporter.ts b/backend/src/metrics/prometheusExporter.ts new file mode 100644 index 000000000..df717b827 --- /dev/null +++ b/backend/src/metrics/prometheusExporter.ts @@ -0,0 +1,159 @@ +/** + * Prometheus Exporter — Issue BE-013 + * + * Lightweight in-process registry for the operational metrics produced by the + * OpenTelemetry tracing pipeline (HTTP server latency, tracing overhead, queue + * job duration and Pub/Sub message throughput). Rendered in the standard + * Prometheus text exposition format (version 0.0.4) and appended to the + * existing `/api/v1/metrics/prometheus` scrape payload. + * + * Kept dependency-free on purpose: recording a sample is a couple of array + * writes, so the per-request cost stays far below the 1.5ms tracing budget. + */ + +/** Metric name prefix, shared with MetricsExporter. */ +const PREFIX = 'w3sl'; + +type Labels = Record; + +/** Default latency buckets in seconds (1ms … 10s). */ +export const DEFAULT_LATENCY_BUCKETS = [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10]; + +/** Fine-grained buckets for the tracing overhead histogram (50µs … 5ms). */ +export const OVERHEAD_BUCKETS = [0.00005, 0.0001, 0.00025, 0.0005, 0.001, 0.0015, 0.0025, 0.005]; + +function escapeLabelValue(value: string): string { + return value.replace(/\\/g, '\\\\').replace(/"/g, '\\"').replace(/\n/g, '\\n'); +} + +function renderLabels(labels: Labels): string { + const entries = Object.entries(labels); + if (entries.length === 0) return ''; + return `{${entries.map(([k, v]) => `${k}="${escapeLabelValue(v)}"`).join(',')}}`; +} + +function labelKey(labels: Labels): string { + return JSON.stringify(Object.entries(labels).sort(([a], [b]) => a.localeCompare(b))); +} + +function formatValue(value: number): string { + if (value === Infinity) return '+Inf'; + if (value === -Infinity) return '-Inf'; + return String(value); +} + +interface HistogramSeries { + labels: Labels; + bucketCounts: number[]; + sum: number; + count: number; +} + +export class Histogram { + private readonly series = new Map(); + + constructor( + readonly name: string, + readonly help: string, + private readonly buckets: number[] = DEFAULT_LATENCY_BUCKETS, + ) {} + + observe(labels: Labels, value: number): void { + const key = labelKey(labels); + let entry = this.series.get(key); + if (!entry) { + entry = { labels, bucketCounts: new Array(this.buckets.length).fill(0), sum: 0, count: 0 }; + this.series.set(key, entry); + } + for (let i = 0; i < this.buckets.length; i += 1) { + if (value <= this.buckets[i]!) entry.bucketCounts[i]! += 1; + } + entry.sum += value; + entry.count += 1; + } + + reset(): void { + this.series.clear(); + } + + render(): string { + const lines = [`# HELP ${this.name} ${this.help}`, `# TYPE ${this.name} histogram`]; + for (const entry of this.series.values()) { + this.buckets.forEach((le, i) => { + lines.push( + `${this.name}_bucket${renderLabels({ ...entry.labels, le: formatValue(le) })} ${entry.bucketCounts[i]}`, + ); + }); + lines.push(`${this.name}_bucket${renderLabels({ ...entry.labels, le: '+Inf' })} ${entry.count}`); + lines.push(`${this.name}_sum${renderLabels(entry.labels)} ${entry.sum}`); + lines.push(`${this.name}_count${renderLabels(entry.labels)} ${entry.count}`); + } + return lines.join('\n'); + } +} + +export class Counter { + private readonly series = new Map(); + + constructor( + readonly name: string, + readonly help: string, + ) {} + + inc(labels: Labels = {}, amount = 1): void { + const key = labelKey(labels); + const entry = this.series.get(key); + if (entry) { + entry.value += amount; + } else { + this.series.set(key, { labels, value: amount }); + } + } + + reset(): void { + this.series.clear(); + } + + render(): string { + const lines = [`# HELP ${this.name} ${this.help}`, `# TYPE ${this.name} counter`]; + for (const entry of this.series.values()) { + lines.push(`${this.name}${renderLabels(entry.labels)} ${entry.value}`); + } + return lines.join('\n'); + } +} + +// ─── Registered metric families ─────────────────────────────────────────────── + +export const httpServerDuration = new Histogram( + `${PREFIX}_http_server_request_duration_seconds`, + 'Duration of inbound HTTP requests, by method, normalised route and status class.', +); + +export const tracingOverhead = new Histogram( + `${PREFIX}_tracing_overhead_seconds`, + 'Time spent inside the tracing middleware itself per request (budget: 1.5ms).', + OVERHEAD_BUCKETS, +); + +export const queueJobDuration = new Histogram( + `${PREFIX}_queue_job_duration_seconds`, + 'Duration of traced BullMQ job executions, by queue and outcome.', +); + +export const pubsubMessages = new Counter( + `${PREFIX}_pubsub_messages_total`, + 'Redis Pub/Sub messages carrying trace context, by channel and direction.', +); + +const families: Array = [httpServerDuration, tracingOverhead, queueJobDuration, pubsubMessages]; + +/** Render every tracing-related metric family as Prometheus text exposition. */ +export function renderTracingPrometheus(): string { + return `${families.map((family) => family.render()).join('\n')}\n`; +} + +/** Clear all recorded samples (tests / admin reset). */ +export function resetTracingMetrics(): void { + for (const family of families) family.reset(); +} diff --git a/backend/src/middleware/dynamicRateLimiter.ts b/backend/src/middleware/dynamicRateLimiter.ts new file mode 100644 index 000000000..356195022 --- /dev/null +++ b/backend/src/middleware/dynamicRateLimiter.ts @@ -0,0 +1,129 @@ +/** + * Dynamic Rate Limiter — Issue BE-012 + * + * Reputation-aware sliding-window-log limiter. Each request is: + * 1. identified (DID / student / IP) and scored by `reputationService`; + * 2. checked against the tier's burst (1s) and sustained (60s) windows in a + * single atomic Redis call (`utils/slidingWindow.ts`), so there are no + * fixed-window boundary spikes; + * 3. annotated with standard rate-limit headers. + * + * Headers on every response: + * X-RateLimit-Limit / X-RateLimit-Remaining / X-RateLimit-Reset (epoch s) + * X-RateLimit-Tier, RateLimit-Policy (e.g. `10;w=1, 60;w=60`) + * and on HTTP 429 additionally `Retry-After` (seconds). + * + * Rejections are reported back to the reputation service, so a caller that + * keeps hammering is demoted to the `restricted` tier, while verified students + * and staking DIDs are elevated automatically. + * + * Usage: + * app.use('/api', dynamicRateLimiter()); + */ + +import { NextFunction, Request, RequestHandler, Response } from 'express'; +import { + reputationService as defaultReputationService, + type CallerIdentity, + type ReputationProfile, + type ReputationService, +} from '../services/reputationService.js'; +import logger from '../utils/logger.js'; +import { redisConnection } from '../utils/redis.js'; +import { + RedisSlidingWindowStore, + ResilientSlidingWindowStore, + type SlidingWindowResult, + type SlidingWindowStore, +} from '../utils/slidingWindow.js'; + +export interface DynamicRateLimiterOptions { + store?: SlidingWindowStore; + reputation?: ReputationService; + /** Redis key prefix for the sliding-window logs. */ + prefix?: string; + /** Skip limiting for some requests (health checks, internal traffic). */ + skip?: (req: Request) => boolean; +} + +const DEFAULT_PREFIX = 'rl:dyn:'; + +function setRateLimitHeaders(res: Response, result: SlidingWindowResult, tier: string, now: number): void { + const { binding } = result; + res.setHeader('X-RateLimit-Limit', String(binding.limit)); + res.setHeader('X-RateLimit-Remaining', String(binding.remaining)); + res.setHeader('X-RateLimit-Reset', String(Math.ceil((now + binding.resetMs) / 1000))); + res.setHeader('X-RateLimit-Tier', tier); + res.setHeader( + 'RateLimit-Policy', + result.windows.map((w) => `${w.limit};w=${Math.ceil(w.windowMs / 1000)}`).join(', '), + ); +} + +export function dynamicRateLimiter(options: DynamicRateLimiterOptions = {}): RequestHandler { + const store = options.store ?? new ResilientSlidingWindowStore(new RedisSlidingWindowStore(redisConnection)); + const reputation = options.reputation ?? defaultReputationService; + const prefix = options.prefix ?? DEFAULT_PREFIX; + + return async (req: Request, res: Response, next: NextFunction): Promise => { + if (options.skip?.(req)) { + next(); + return; + } + + let identity: CallerIdentity; + let profile: ReputationProfile | null; + try { + identity = await reputation.identify(req); + profile = await reputation.getProfile(identity); + } catch (error) { + // Reputation lookup failure must never block traffic: fall back to strict IP limits. + logger.warn('Reputation lookup failed; applying anonymous limits', { + error: error instanceof Error ? error.message : String(error), + }); + identity = reputation.resolveIdentity(req, undefined); + profile = null; + } + + const tier = profile?.tier ?? 'anonymous'; + const now = Date.now(); + let result: SlidingWindowResult; + try { + result = await store.hit(`${prefix}${identity.key}`, reputation.limitsFor(tier), now); + } catch (error) { + logger.error('Dynamic rate limiter unavailable; allowing request', { + error: error instanceof Error ? error.message : String(error), + }); + next(); + return; + } + + setRateLimitHeaders(res, result, tier, now); + + if (result.allowed) { + next(); + return; + } + + const retryAfterSeconds = Math.max(1, Math.ceil(result.retryAfterMs / 1000)); + res.setHeader('Retry-After', String(retryAfterSeconds)); + + void reputation.recordViolation(identity).catch(() => undefined); + logger.warn('Dynamic rate limit exceeded', { + identity: identity.kind, + tier, + window: result.binding.name, + path: req.path, + }); + + res.status(429).json({ + status: 'error', + message: 'Too many requests. Please slow down.', + tier, + window: result.binding.name, + retry_after: retryAfterSeconds, + }); + }; +} + +export default dynamicRateLimiter; diff --git a/backend/src/middleware/tracingMiddleware.ts b/backend/src/middleware/tracingMiddleware.ts new file mode 100644 index 000000000..8225c8af0 --- /dev/null +++ b/backend/src/middleware/tracingMiddleware.ts @@ -0,0 +1,100 @@ +/** + * Tracing Middleware — Issue BE-013 + * + * Opens an OpenTelemetry SERVER span for every inbound HTTP request, continuing + * any W3C `traceparent` / `tracestate` sent by the caller, and runs the rest of + * the Express pipeline inside that span's context. Anything started while the + * request is in flight — Prisma queries, RPC calls, BullMQ enqueues, Redis + * publishes — becomes a child of this span. + * + * The trace id is returned to the client in the `traceresponse` header and + * exposed as `res.locals.traceId` for log correlation. + * + * The middleware measures its own cost and reports it through the + * `w3sl_tracing_overhead_seconds` histogram (budget: < 1.5ms per request). + */ + +import { NextFunction, Request, RequestHandler, Response } from 'express'; +import { context, isSpanContextValid, propagation, ROOT_CONTEXT, SpanKind, SpanStatusCode, trace } from '@opentelemetry/api'; +import { tracer } from '../instrumentation.js'; +import { normalizeRouteLabel } from '../metrics/MetricsExporter.js'; +import { httpServerDuration, tracingOverhead } from '../metrics/prometheusExporter.js'; + +export interface TracingMiddlewareOptions { + /** Exact paths that are not traced (health probes, scrapes). */ + ignorePaths?: string[]; +} + +const DEFAULT_IGNORED_PATHS = ['/health', '/health/live', '/health/ready']; + +function elapsedSeconds(start: bigint): number { + return Number(process.hrtime.bigint() - start) / 1e9; +} + +function statusClass(statusCode: number): string { + return `${Math.floor(statusCode / 100)}xx`; +} + +export function tracingMiddleware(options: TracingMiddlewareOptions = {}): RequestHandler { + const ignored = new Set(options.ignorePaths ?? DEFAULT_IGNORED_PATHS); + + return (req: Request, res: Response, next: NextFunction): void => { + if (ignored.has(req.path)) { + next(); + return; + } + + const requestStart = process.hrtime.bigint(); + + const parent = propagation.extract(ROOT_CONTEXT, req.headers); + const span = tracer.startSpan( + req.method, + { + kind: SpanKind.SERVER, + attributes: { + 'http.request.method': req.method, + 'url.path': req.path, + 'url.scheme': req.protocol, + 'user_agent.original': req.get('user-agent') ?? '', + }, + }, + parent, + ); + + const spanContext = span.spanContext(); + if (isSpanContextValid(spanContext)) { + const flags = spanContext.traceFlags.toString(16).padStart(2, '0'); + res.setHeader('traceresponse', `00-${spanContext.traceId}-${spanContext.spanId}-${flags}`); + res.locals.traceId = spanContext.traceId; + } + + let overhead = elapsedSeconds(requestStart); + + res.once('finish', () => { + const finishStart = process.hrtime.bigint(); + + const rawRoute = req.route?.path ? `${req.baseUrl}${String(req.route.path)}` : `${req.baseUrl}${req.path}`; + const route = normalizeRouteLabel(rawRoute); + + span.updateName(`${req.method} ${route}`); + span.setAttribute('http.route', route); + span.setAttribute('http.response.status_code', res.statusCode); + if (res.statusCode >= 500) { + span.setStatus({ code: SpanStatusCode.ERROR }); + } + span.end(); + + httpServerDuration.observe( + { method: req.method, route, status_class: statusClass(res.statusCode) }, + elapsedSeconds(requestStart), + ); + + overhead += elapsedSeconds(finishStart); + tracingOverhead.observe({}, overhead); + }); + + context.with(trace.setSpan(parent, span), next); + }; +} + +export default tracingMiddleware; diff --git a/backend/src/queues/compileQueue.ts b/backend/src/queues/compileQueue.ts index 8b1373790..b224636d0 100644 --- a/backend/src/queues/compileQueue.ts +++ b/backend/src/queues/compileQueue.ts @@ -11,6 +11,7 @@ */ import { Queue, type JobsOptions } from 'bullmq'; +import { traceEnqueue, type TraceCarrier } from '../instrumentation.js'; import logger from '../utils/logger.js'; // --------------------------------------------------------------------------- @@ -48,6 +49,8 @@ export interface CompileJobData { flags?: string[]; /** Correlation ID for distributed tracing */ traceId?: string; + /** W3C trace-context of the enqueuing request (set by traceEnqueue) */ + _traceContext?: TraceCarrier; } export interface CompileJobResult { @@ -147,7 +150,9 @@ export async function enqueueCompileJob( ...opts, }; - const job = await compileQueue.add('compile', jobData, jobOpts); + const job = await traceEnqueue(COMPILE_QUEUE_NAME, jobData, (tracedData) => + compileQueue.add('compile', tracedData, jobOpts), + ); const waitingCount = await compileQueue.getWaitingCount(); diff --git a/backend/src/routes/audit.ts b/backend/src/routes/audit.ts index 5ccc78e61..a2c5349aa 100644 --- a/backend/src/routes/audit.ts +++ b/backend/src/routes/audit.ts @@ -1,5 +1,10 @@ import { Router, Request, Response } from 'express'; import { AuditService } from '../services/audit.service.js'; +import { auditChainStore } from '../db/auditChainStore.js'; +import { verifyPublishedAnchor } from '../jobs/anchorMerkleRootJob.js'; +import { requireAdmin } from '../middleware/admin.js'; +import { authenticateToken } from '../middleware/auth.js'; +import { tamperEvidentAuditService } from '../services/tamperEvidentAuditService.js'; const router = Router(); const auditService = new AuditService(); @@ -31,4 +36,41 @@ router.get('/verify', async (_req: Request, res: Response) => { } }); +// ─── Tamper-evident hash chain (BE-009) ────────────────────────────────────── + +const serializeBigInts = (value: unknown) => + JSON.parse(JSON.stringify(value, (_key, v) => (typeof v === 'bigint' ? v.toString() : v))); + +router.get('/chain/verify', authenticateToken, requireAdmin, async (_req: Request, res: Response) => { + try { + const result = await tamperEvidentAuditService.verifyChain(); + return res.status(result.valid ? 200 : 409).json(result); + } catch (error: any) { + return res.status(500).json({ error: error.message }); + } +}); + +router.get('/chain/anchors', authenticateToken, requireAdmin, async (req: Request, res: Response) => { + try { + const limit = Math.min(Number(req.query.limit) || 20, 100); + const anchors = await auditChainStore.listAnchors(limit); + return res.status(200).json({ anchors: serializeBigInts(anchors) }); + } catch (error: any) { + return res.status(500).json({ error: error.message }); + } +}); + +router.get('/chain/anchors/:id/verify', authenticateToken, requireAdmin, async (req: Request, res: Response) => { + try { + const anchor = await auditChainStore.getAnchor(String(req.params.id)); + if (!anchor) { + return res.status(404).json({ error: 'Anchor not found' }); + } + const verification = await verifyPublishedAnchor(anchor); + return res.status(verification.matches ? 200 : 409).json(serializeBigInts(verification)); + } catch (error: any) { + return res.status(500).json({ error: error.message }); + } +}); + export default router; diff --git a/backend/src/routes/index.ts b/backend/src/routes/index.ts index 288e9445d..62254d116 100644 --- a/backend/src/routes/index.ts +++ b/backend/src/routes/index.ts @@ -18,6 +18,7 @@ import generatorRouter from './generator/generator.routes.js'; import healthRouter from './health.routes.js'; import i18nRouter from './i18n.routes.js'; import learningRoutes from './learning/learning.routes.js'; +import mediaRouter from './media.routes.js'; import oauthRouter from './oauth.routes.js'; import osctRouter from './osct/osct.routes.js'; import playgroundValidateRouter from './playground.routes.js'; @@ -77,6 +78,7 @@ router.use('/feedback', validateWorkspaceMembership, feedbackRouter); router.use('/learning', validateWorkspaceMembership, learningRoutes); router.use('/canvas', validateWorkspaceMembership, canvasRouter); +router.use('/media', mediaRouter); router.use('/dashboard', dashboardRoutes); router.use('/dashboard/activity-log', activityLogRouter); router.use('/auth', authRoutes); diff --git a/backend/src/routes/media.routes.ts b/backend/src/routes/media.routes.ts new file mode 100644 index 000000000..ed30e9bbc --- /dev/null +++ b/backend/src/routes/media.routes.ts @@ -0,0 +1,16 @@ +import { Router } from 'express'; +import { mediaController } from '../controllers/mediaController.js'; +import { requireAdmin } from '../middleware/admin.js'; +import { authenticateToken } from '../middleware/auth.js'; + +const router: ReturnType = Router(); + +// Public lookups +router.get('/certificates/:certificateId', mediaController.getCertificateAssets); +router.get('/:cid', mediaController.getByCid); + +// Protected operations +router.post('/pin', authenticateToken, requireAdmin, mediaController.pinAsset); +router.post('/:cid/verify', authenticateToken, mediaController.verifyCid); + +export default router; diff --git a/backend/src/routes/metrics.routes.ts b/backend/src/routes/metrics.routes.ts index 024209fd4..b1af99a8c 100644 --- a/backend/src/routes/metrics.routes.ts +++ b/backend/src/routes/metrics.routes.ts @@ -24,6 +24,7 @@ import { buildMetricsSnapshot, renderPrometheus, } from '../metrics/MetricsExporter.js'; +import { renderTracingPrometheus } from '../metrics/prometheusExporter.js'; import { requireMetricsAuth } from '../middleware/metricsAuth.js'; import { slidingWindowRateLimiter } from '../middleware/rateLimiter.js'; import { authenticateToken } from '../middleware/auth.js'; @@ -67,7 +68,7 @@ router.use( */ router.get('/prometheus', (_req: Request, res: Response) => { res.setHeader('Content-Type', PROMETHEUS_CONTENT_TYPE); - res.status(200).send(renderPrometheus()); + res.status(200).send(`${renderPrometheus()}${renderTracingPrometheus()}`); }); /** diff --git a/backend/src/services/ipfsPinningService.ts b/backend/src/services/ipfsPinningService.ts new file mode 100644 index 000000000..7e7057367 --- /dev/null +++ b/backend/src/services/ipfsPinningService.ts @@ -0,0 +1,540 @@ +/** + * IPFS / Filecoin Pinning Pipeline — Issue BE-011 + * + * Dual-pinning pipeline for certificate metadata, certificate SVGs and course + * media: + * + * 1. Hot pin — the asset is pinned to IPFS through the ordered `hot` + * providers (Pinata → Kubo). Each provider is retried with exponential + * backoff + jitter before falling back to the next one. + * 2. CID validation — the returned CID must be a well-formed CIDv1 + * (base32 multibase, version 1, sha2-256 multihash); anything else is + * treated as a provider failure. + * 3. Persistence — the immutable CID is stored in PostgreSQL + * (`pinned_assets`, guarded by a trigger that forbids rewriting a CID). + * 4. Filecoin — the same bytes are uploaded in the background to the + * `filecoin` provider (Web3.Storage) which brokers persistent storage + * deals; status is tracked per asset. + * 5. Gateway verification — the CID is fetched through the Cloudflare CDN + * gateway (warming its cache for active assets) and, on failure, through + * the secondary gateways, again with exponential backoff. + * + * `autoPinCertificate()` runs steps 1–3 for a freshly issued certificate under + * a 3-second deadline; steps 4–5 continue asynchronously. + */ + +import { createHash } from 'crypto'; +import { ipfsConfig, type IpfsConfig, type PinningProviderConfig } from '../config/ipfs.js'; +import prisma from '../db/index.js'; +import logger from '../utils/logger.js'; + +// ─── CIDv1 validation ──────────────────────────────────────────────────────── + +const BASE32_ALPHABET = 'abcdefghijklmnopqrstuvwxyz234567'; +const SHA2_256 = 0x12; + +function base32Decode(input: string): Uint8Array | null { + const out: number[] = []; + let buffer = 0; + let bits = 0; + for (const char of input) { + const value = BASE32_ALPHABET.indexOf(char); + if (value === -1) return null; + buffer = (buffer << 5) | value; + bits += 5; + if (bits >= 8) { + bits -= 8; + out.push((buffer >> bits) & 0xff); + } + } + return Uint8Array.from(out); +} + +function readVarint(bytes: Uint8Array, offset: number): [number, number] | null { + let value = 0; + let shift = 0; + for (let i = offset; i < bytes.length && i < offset + 9; i += 1) { + const byte = bytes[i]!; + value += (byte & 0x7f) * 2 ** shift; + if ((byte & 0x80) === 0) return [value, i + 1]; + shift += 7; + } + return null; +} + +/** + * Strict CIDv1 check: base32 multibase prefix `b`, version varint = 1, a codec + * varint, and a complete sha2-256 multihash (code 0x12, 32-byte digest). + */ +export function isCidV1(cid: string): boolean { + if (typeof cid !== 'string' || !/^b[a-z2-7]{50,}$/.test(cid)) return false; + const bytes = base32Decode(cid.slice(1)); + if (!bytes) return false; + + const version = readVarint(bytes, 0); + if (!version || version[0] !== 1) return false; + const codec = readVarint(bytes, version[1]); + if (!codec) return false; + const hashCode = readVarint(bytes, codec[1]); + if (!hashCode || hashCode[0] !== SHA2_256) return false; + const digestLength = readVarint(bytes, hashCode[1]); + if (!digestLength || digestLength[0] !== 32) return false; + + return digestLength[1] + digestLength[0] === bytes.length; +} + +// ─── Retry helper ──────────────────────────────────────────────────────────── + +const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); + +export function backoffDelay(attempt: number, baseDelayMs: number, maxDelayMs: number): number { + const exponential = Math.min(maxDelayMs, baseDelayMs * 2 ** attempt); + // Full jitter keeps concurrent retries from synchronising. + return Math.round(exponential / 2 + Math.random() * (exponential / 2)); +} + +async function withRetry( + label: string, + fn: (attempt: number) => Promise, + options: { maxAttempts: number; baseDelayMs: number; maxDelayMs: number; deadline?: number }, +): Promise { + let lastError: unknown; + for (let attempt = 0; attempt < options.maxAttempts; attempt += 1) { + try { + return await fn(attempt); + } catch (error) { + lastError = error; + if (attempt === options.maxAttempts - 1) break; + const delay = backoffDelay(attempt, options.baseDelayMs, options.maxDelayMs); + if (options.deadline && Date.now() + delay >= options.deadline) break; + logger.debug(`${label} attempt ${attempt + 1} failed; retrying in ${delay}ms`, { + error: error instanceof Error ? error.message : String(error), + }); + await sleep(delay); + } + } + throw lastError instanceof Error ? lastError : new Error(String(lastError)); +} + +// ─── Types ─────────────────────────────────────────────────────────────────── + +export interface PinAssetRequest { + content: Buffer; + name: string; + contentType: string; + /** e.g. `certificate`, `course`. */ + resourceType: string; + resourceId: string; + /** e.g. `certificate-metadata`, `certificate-svg`, `course-video`. */ + kind: string; +} + +export interface PinnedAssetRecord { + id: string; + resourceType: string; + resourceId: string; + kind: string; + cid: string; + contentType: string; + sizeBytes: number; + sha256: string; + provider: string; + filecoinProvider: string | null; + filecoinCid: string | null; + filecoinStatus: string; + cdnUrl: string; + gatewayVerified: boolean; + verifiedGateway: string | null; + verifiedAt: Date | null; + pinLatencyMs: number; + createdAt: Date; + updatedAt: Date; +} + +export interface GatewayCheckResult { + cid: string; + accessible: boolean; + gateway: string | null; + attempts: Array<{ gateway: string; ok: boolean; status?: number; error?: string }>; +} + +export interface CertificatePinRequest { + certificateId: string; + metadata: Record; + /** Optional pre-rendered SVG; one is rendered from the metadata otherwise. */ + svg?: string; +} + +export interface CertificatePinResult { + certificateId: string; + svg: PinnedAssetRecord; + metadata: PinnedAssetRecord; + elapsedMs: number; + withinSla: boolean; +} + +export interface IpfsPinningServiceOptions { + config?: IpfsConfig; + fetchFn?: typeof fetch; + /** Persistence override (tests). Defaults to Prisma `pinnedAsset`. */ + repository?: PinnedAssetRepository; +} + +export interface PinnedAssetRepository { + upsert(record: Omit): Promise; + update(id: string, patch: Partial): Promise; + findByCid(cid: string): Promise; + findByResource(resourceType: string, resourceId: string): Promise; +} + +const db = prisma as any; + +export const prismaPinnedAssetRepository: PinnedAssetRepository = { + async upsert(record) { + // The CID of an existing (resourceType, resourceId, kind) is never + // rewritten — only the mutable status fields are refreshed. + return db.pinnedAsset.upsert({ + where: { + resourceType_resourceId_kind: { + resourceType: record.resourceType, + resourceId: record.resourceId, + kind: record.kind, + }, + }, + create: record, + update: { cdnUrl: record.cdnUrl }, + }); + }, + update(id, patch) { + return db.pinnedAsset.update({ where: { id }, data: patch }); + }, + findByCid(cid) { + return db.pinnedAsset.findMany({ where: { cid }, orderBy: { createdAt: 'asc' } }); + }, + findByResource(resourceType, resourceId) { + return db.pinnedAsset.findMany({ where: { resourceType, resourceId }, orderBy: { createdAt: 'asc' } }); + }, +}; + +// ─── Service ───────────────────────────────────────────────────────────────── + +const XML_ESCAPES: Record = { '<': '<', '>': '>', '&': '&', "'": ''', '"': '"' }; + +function escapeXml(value: string): string { + return value.replace(/[<>&'"]/g, (c) => XML_ESCAPES[c] ?? c); +} + +/** Minimal, deterministic certificate SVG rendered from NFT metadata. */ +export function renderCertificateSvg(metadata: Record): string { + const name = escapeXml(String(metadata.name ?? 'Certificate of Completion')); + const description = escapeXml(String(metadata.description ?? '').slice(0, 140)); + const attributes = Array.isArray(metadata.attributes) ? (metadata.attributes as Array>) : []; + const lines = attributes + .slice(0, 6) + .map( + (attr, i) => + `${escapeXml( + `${String(attr.trait_type ?? '')}: ${String(attr.value ?? '')}`, + )}`, + ) + .join(''); + + return ( + `` + + `` + + `` + + `${name}` + + `${description}` + + lines + + `Web3 Student Lab · Verifiable on IPFS` + + `` + ); +} + +export class IpfsPinningService { + private readonly config: IpfsConfig; + private readonly fetchFn: typeof fetch; + private readonly repository: PinnedAssetRepository; + + constructor(options: IpfsPinningServiceOptions = {}) { + this.config = options.config ?? ipfsConfig; + this.fetchFn = options.fetchFn ?? fetch; + this.repository = options.repository ?? prismaPinnedAssetRepository; + } + + /** Public, CDN-cached URL for a CID. */ + cdnUrl(cid: string): string { + return `${this.config.cdnBaseUrl}/ipfs/${cid}`; + } + + gatewayUrls(cid: string): string[] { + return [this.config.cdnBaseUrl, ...this.config.secondaryGateways].map((base) => `${base}/ipfs/${cid}`); + } + + /** + * Pin an asset (hot IPFS pin + DB record) and schedule Filecoin storage and + * gateway verification in the background. + */ + async pinAsset(request: PinAssetRequest, deadline?: number): Promise { + const started = Date.now(); + const sha256 = createHash('sha256').update(request.content).digest('hex'); + const { cid, provider } = await this.pinHot(request, deadline); + + const filecoin = this.config.providers.find((p) => p.role === 'filecoin'); + const record = await this.repository.upsert({ + resourceType: request.resourceType, + resourceId: request.resourceId, + kind: request.kind, + cid, + contentType: request.contentType, + sizeBytes: request.content.length, + sha256, + provider, + filecoinProvider: filecoin?.name ?? null, + filecoinCid: null, + filecoinStatus: filecoin ? 'pending' : 'disabled', + cdnUrl: this.cdnUrl(cid), + gatewayVerified: false, + verifiedGateway: null, + verifiedAt: null, + pinLatencyMs: Date.now() - started, + }); + + if (record.cid !== cid) { + logger.warn('Re-pinned asset produced a different CID; keeping the original immutable CID', { + resourceType: request.resourceType, + resourceId: request.resourceId, + kind: request.kind, + storedCid: record.cid, + newCid: cid, + }); + } + + void this.completeInBackground(record, request, filecoin); + return record; + } + + /** + * Auto-pin a newly issued certificate: SVG first (so metadata can reference + * it as `image`), then the metadata JSON — all within the configured SLA. + */ + async autoPinCertificate(request: CertificatePinRequest): Promise { + const started = Date.now(); + const deadline = started + this.config.certificatePinSlaMs; + + const svgContent = request.svg ?? renderCertificateSvg(request.metadata); + const svg = await this.pinAsset( + { + content: Buffer.from(svgContent, 'utf8'), + name: `certificate-${request.certificateId}.svg`, + contentType: 'image/svg+xml', + resourceType: 'certificate', + resourceId: request.certificateId, + kind: 'certificate-svg', + }, + deadline, + ); + + const metadata = await this.pinAsset( + { + content: Buffer.from(JSON.stringify({ ...request.metadata, image: `ipfs://${svg.cid}` }), 'utf8'), + name: `certificate-${request.certificateId}.json`, + contentType: 'application/json', + resourceType: 'certificate', + resourceId: request.certificateId, + kind: 'certificate-metadata', + }, + deadline, + ); + + const elapsedMs = Date.now() - started; + const withinSla = elapsedMs <= this.config.certificatePinSlaMs; + const logMeta = { + certificateId: request.certificateId, + metadataCid: metadata.cid, + svgCid: svg.cid, + elapsedMs, + slaMs: this.config.certificatePinSlaMs, + }; + if (withinSla) { + logger.info('Certificate pinned to IPFS', logMeta); + } else { + logger.warn('Certificate pinned to IPFS after the SLA deadline', logMeta); + } + + return { certificateId: request.certificateId, svg, metadata, elapsedMs, withinSla }; + } + + /** + * Check that a CID is retrievable: the CDN gateway first, then each + * secondary gateway, retrying each with exponential backoff. + */ + async verifyGatewayAccessibility(cid: string, maxAttemptsPerGateway = 2): Promise { + if (!isCidV1(cid)) throw new Error(`Not a valid CIDv1: ${cid}`); + + const attempts: GatewayCheckResult['attempts'] = []; + for (const url of this.gatewayUrls(cid)) { + try { + const status = await withRetry( + `gateway ${url}`, + async () => { + const response = await this.fetchFn(url, { + method: 'HEAD', + signal: AbortSignal.timeout(this.config.retry.attemptTimeoutMs * 2), + }); + if (!response.ok) throw Object.assign(new Error(`HTTP ${response.status}`), { status: response.status }); + return response.status; + }, + { ...this.config.retry, maxAttempts: maxAttemptsPerGateway }, + ); + attempts.push({ gateway: url, ok: true, status }); + return { cid, accessible: true, gateway: url, attempts }; + } catch (error) { + attempts.push({ + gateway: url, + ok: false, + ...(typeof (error as { status?: number }).status === 'number' + ? { status: (error as { status: number }).status } + : {}), + error: error instanceof Error ? error.message : String(error), + }); + } + } + return { cid, accessible: false, gateway: null, attempts }; + } + + findByCid(cid: string): Promise { + return this.repository.findByCid(cid); + } + + findByResource(resourceType: string, resourceId: string): Promise { + return this.repository.findByResource(resourceType, resourceId); + } + + /** Re-run gateway verification for a stored asset and persist the result. */ + async reverify(record: PinnedAssetRecord): Promise { + const check = await this.verifyGatewayAccessibility(record.cid); + await this.repository.update(record.id, { + gatewayVerified: check.accessible, + verifiedGateway: check.gateway, + verifiedAt: check.accessible ? new Date() : record.verifiedAt, + }); + return check; + } + + // ─── internals ───────────────────────────────────────────────────────────── + + private async pinHot(request: PinAssetRequest, deadline?: number): Promise<{ cid: string; provider: string }> { + const providers = this.config.providers.filter((p) => p.role === 'hot'); + const errors: Error[] = []; + + for (const provider of providers) { + if (deadline && Date.now() >= deadline) break; + try { + const cid = await withRetry( + `pin ${provider.name}`, + async () => { + const result = await this.upload(provider, request); + if (!isCidV1(result)) throw new Error(`${provider.name} returned a non-CIDv1 identifier "${result}"`); + return result; + }, + { ...this.config.retry, ...(deadline ? { deadline } : {}) }, + ); + return { cid, provider: provider.name }; + } catch (error) { + const err = error instanceof Error ? error : new Error(String(error)); + errors.push(err); + logger.warn(`IPFS provider ${provider.name} failed; falling back`, { error: err.message, name: request.name }); + } + } + + throw new AggregateError(errors, `All IPFS pinning providers failed for ${request.name}`); + } + + private async completeInBackground( + record: PinnedAssetRecord, + request: PinAssetRequest, + filecoin: PinningProviderConfig | undefined, + ): Promise { + if (filecoin) { + try { + const filecoinCid = await withRetry(`filecoin ${filecoin.name}`, () => this.upload(filecoin, request), { + ...this.config.retry, + maxAttempts: this.config.retry.maxAttempts + 2, + }); + if (filecoinCid !== record.cid) { + logger.warn('Filecoin provider returned a different CID (different DAG layout)', { + cid: record.cid, + filecoinCid, + }); + } + await this.repository.update(record.id, { filecoinCid, filecoinStatus: 'stored' }); + } catch (error) { + logger.error('Filecoin storage deal upload failed', { + cid: record.cid, + error: error instanceof Error ? error.message : String(error), + }); + await this.repository.update(record.id, { filecoinStatus: 'failed' }).catch(() => undefined); + } + } + + try { + await this.reverify(record); + } catch (error) { + logger.warn('Gateway verification failed', { + cid: record.cid, + error: error instanceof Error ? error.message : String(error), + }); + } + } + + private async upload(provider: PinningProviderConfig, request: PinAssetRequest): Promise { + const signal = AbortSignal.timeout(this.config.retry.attemptTimeoutMs); + const blob = new Blob([Uint8Array.from(request.content)], { type: request.contentType }); + const headers: Record = {}; + let url = provider.endpoint; + let body: FormData | Blob; + + if (provider.type === 'pinata') { + headers.Authorization = `Bearer ${provider.authToken ?? ''}`; + const form = new FormData(); + form.append('file', blob, request.name); + form.append('pinataOptions', JSON.stringify({ cidVersion: 1 })); + form.append( + 'pinataMetadata', + JSON.stringify({ name: request.name, keyvalues: { kind: request.kind, resourceId: request.resourceId } }), + ); + body = form; + } else if (provider.type === 'kubo') { + if (provider.authToken) { + headers.Authorization = `Basic ${Buffer.from(`${provider.authToken}:${provider.authSecret ?? ''}`).toString('base64')}`; + } + const separator = url.includes('?') ? '&' : '?'; + url = `${url}${separator}cid-version=1&hash=sha2-256&pin=true`; + const form = new FormData(); + form.append('file', blob, request.name); + body = form; + } else { + headers.Authorization = `Bearer ${provider.authToken ?? ''}`; + headers['X-Name'] = encodeURIComponent(request.name); + headers['Content-Type'] = request.contentType; + body = blob; + } + + const response = await this.fetchFn(url, { method: 'POST', headers, body, signal }); + if (!response.ok) { + throw new Error(`HTTP ${response.status} from ${provider.name}`); + } + + const text = await response.text(); + const firstLine = text.split('\n').find((line) => line.trim().length > 0) ?? text; + const parsed = JSON.parse(firstLine) as Record; + const cid = provider.type === 'pinata' ? parsed.IpfsHash : provider.type === 'kubo' ? parsed.Hash : parsed.cid; + if (typeof cid !== 'string' || cid.length === 0) { + throw new Error(`${provider.name} response did not contain a CID`); + } + return cid; + } +} + +export const ipfsPinningService = new IpfsPinningService(); diff --git a/backend/src/services/reputationService.ts b/backend/src/services/reputationService.ts new file mode 100644 index 000000000..83efb1b6d --- /dev/null +++ b/backend/src/services/reputationService.ts @@ -0,0 +1,332 @@ +/** + * Reputation Service — Issue BE-012 + * + * Scores every caller (an anonymous IP or an authenticated student / DID) and + * maps the score to a rate-limit tier. Positive signals raise the tier + * automatically; recent rate-limit violations lower it: + * + * signal points + * ─────────────────────────────────────── ────── + * authenticated session +40 + * linked decentralized identifier (DID) +20 + * verified student (≥1 active certificate) +20 + * active on-chain stake (Stellar asset) +40 + * each recent rate-limit violation -10 + * + * score < 0 → restricted score < 40 → anonymous + * score < 60 → authenticated score < 100 → verified + * score ≥ 100 → staker (authenticated DID with an active stake) + * + * Profiles are cached for `REPUTATION_CACHE_TTL_MS` (default 60s) so the + * database / Horizon are hit at most once per caller per minute. + */ + +import type { Request } from 'express'; +import { Horizon } from '@stellar/stellar-sdk'; +import { getStudentById } from '../auth/auth.service.js'; +import { verifyAccessToken } from '../auth/token.service.js'; +import type { User } from '../auth/types.js'; +import prisma from '../db/index.js'; +import type { WindowRule } from '../utils/slidingWindow.js'; +import logger from '../utils/logger.js'; +import { redisConnection } from '../utils/redis.js'; + +export type ReputationTier = 'restricted' | 'anonymous' | 'authenticated' | 'verified' | 'staker'; + +export interface CallerIdentity { + /** Stable rate-limit key: `did:…`, `user:…` or `ip:…`. */ + key: string; + kind: 'did' | 'user' | 'ip'; + userId?: string; + did?: string | null; + walletAddress?: string | null; + ip: string; +} + +export interface ReputationProfile { + identity: CallerIdentity; + score: number; + tier: ReputationTier; + signals: { + authenticated: boolean; + hasDid: boolean; + verifiedStudent: boolean; + activeStake: boolean; + recentViolations: number; + }; + computedAt: number; +} + +/** Burst (1s) + sustained (60s) windows per tier. Unauthenticated IPs get strict bursts. */ +export const TIER_LIMITS: Record = { + restricted: [ + { name: 'burst', limit: 3, windowMs: 1_000 }, + { name: 'sustained', limit: 20, windowMs: 60_000 }, + ], + anonymous: [ + { name: 'burst', limit: 10, windowMs: 1_000 }, + { name: 'sustained', limit: 60, windowMs: 60_000 }, + ], + authenticated: [ + { name: 'burst', limit: 20, windowMs: 1_000 }, + { name: 'sustained', limit: 300, windowMs: 60_000 }, + ], + verified: [ + { name: 'burst', limit: 40, windowMs: 1_000 }, + { name: 'sustained', limit: 600, windowMs: 60_000 }, + ], + staker: [ + { name: 'burst', limit: 100, windowMs: 1_000 }, + { name: 'sustained', limit: 1_500, windowMs: 60_000 }, + ], +}; + +export const SCORE_WEIGHTS = { + authenticated: 40, + did: 20, + verifiedStudent: 20, + activeStake: 40, + violation: -10, +} as const; + +const VIOLATION_TTL_SECONDS = 3_600; +const MAX_COUNTED_VIOLATIONS = 10; + +export function tierForScore(score: number): ReputationTier { + if (score < 0) return 'restricted'; + if (score < 40) return 'anonymous'; + if (score < 60) return 'authenticated'; + if (score < 100) return 'verified'; + return 'staker'; +} + +/** Checks whether a Stellar account currently holds an active stake. */ +export interface StakingProvider { + hasActiveStake(walletAddress: string): Promise; +} + +/** + * Default staking check: the wallet holds at least `STAKING_MIN_BALANCE` of the + * staking asset (`STAKING_ASSET_CODE` issued by `STAKING_ASSET_ISSUER`). + * Without that configuration nobody is treated as staking. + */ +export class HorizonStakingProvider implements StakingProvider { + private readonly server: Horizon.Server; + private readonly assetCode = process.env.STAKING_ASSET_CODE || ''; + private readonly assetIssuer = process.env.STAKING_ASSET_ISSUER || ''; + private readonly minBalance = Number(process.env.STAKING_MIN_BALANCE || '1'); + + constructor(horizonUrl = process.env.STELLAR_HORIZON_URL || 'https://horizon-testnet.stellar.org') { + this.server = new Horizon.Server(horizonUrl); + } + + async hasActiveStake(walletAddress: string): Promise { + if (!this.assetCode || !this.assetIssuer) return false; + try { + const account = await this.server.loadAccount(walletAddress); + return account.balances.some( + (balance) => + 'asset_code' in balance && + 'asset_issuer' in balance && + balance.asset_code === this.assetCode && + balance.asset_issuer === this.assetIssuer && + Number(balance.balance) >= this.minBalance, + ); + } catch (error) { + logger.debug('Staking lookup failed', { + error: error instanceof Error ? error.message : String(error), + }); + return false; + } + } +} + +/** Persists recent violations so reputation decays across instances. */ +export interface ViolationStore { + increment(identityKey: string): Promise; + count(identityKey: string): Promise; +} + +class RedisViolationStore implements ViolationStore { + private readonly memory = new Map(); + + private key(identityKey: string): string { + return `rep:violations:${identityKey}`; + } + + async increment(identityKey: string): Promise { + try { + const count = Number(await redisConnection.incr(this.key(identityKey))); + if (count === 1) await redisConnection.expire(this.key(identityKey), VIOLATION_TTL_SECONDS); + return count; + } catch { + const now = Date.now(); + const entry = this.memory.get(identityKey); + const count = entry && entry.expiresAt > now ? entry.count + 1 : 1; + this.memory.set(identityKey, { + count, + expiresAt: entry && entry.expiresAt > now ? entry.expiresAt : now + VIOLATION_TTL_SECONDS * 1000, + }); + return count; + } + } + + async count(identityKey: string): Promise { + try { + return Number((await redisConnection.get(this.key(identityKey))) ?? 0); + } catch { + const entry = this.memory.get(identityKey); + return entry && entry.expiresAt > Date.now() ? entry.count : 0; + } + } +} + +export interface ReputationServiceOptions { + stakingProvider?: StakingProvider; + violationStore?: ViolationStore; + /** Returns true when the student has at least one active certificate. */ + isVerifiedStudent?: (userId: string) => Promise; + cacheTtlMs?: number; + maxCacheEntries?: number; +} + +async function hasActiveCertificate(userId: string): Promise { + try { + const count = await prisma.certificate.count({ + where: { studentId: userId, status: { in: ['ACTIVE', 'MINTED'] } }, + }); + return count > 0; + } catch { + return false; + } +} + +export class ReputationService { + private readonly stakingProvider: StakingProvider; + private readonly violations: ViolationStore; + private readonly isVerifiedStudent: (userId: string) => Promise; + private readonly cacheTtlMs: number; + private readonly maxCacheEntries: number; + private readonly cache = new Map(); + private readonly users = new Map(); + + constructor(options: ReputationServiceOptions = {}) { + this.stakingProvider = options.stakingProvider ?? new HorizonStakingProvider(); + this.violations = options.violationStore ?? new RedisViolationStore(); + this.isVerifiedStudent = options.isVerifiedStudent ?? hasActiveCertificate; + this.cacheTtlMs = options.cacheTtlMs ?? Number(process.env.REPUTATION_CACHE_TTL_MS || '60000'); + this.maxCacheEntries = options.maxCacheEntries ?? 50_000; + } + + /** + * Identify the caller. Uses `req.user` when an auth middleware already ran; + * otherwise a valid Bearer access token is decoded (and the student loaded, + * cached) so globally-mounted limiters still recognise logged-in students. + * Authorization itself is still enforced by the route's own middleware. + */ + async identify(req: Request): Promise { + if (req.user) return this.resolveIdentity(req, req.user); + + const header = req.headers.authorization; + const token = header?.startsWith('Bearer ') ? header.slice(7) : ''; + if (!token) return this.resolveIdentity(req); + + try { + const { userId } = verifyAccessToken(token); + const user = await this.loadUser(userId); + return this.resolveIdentity(req, user ?? undefined); + } catch { + return this.resolveIdentity(req); + } + } + + resolveIdentity(req: Request, user: User | undefined = req.user): CallerIdentity { + const ip = req.ip || req.socket?.remoteAddress || 'unknown'; + if (user?.did) { + return { key: `did:${user.did}`, kind: 'did', userId: user.id, did: user.did, walletAddress: user.walletAddress, ip }; + } + if (user?.id) { + return { key: `user:${user.id}`, kind: 'user', userId: user.id, did: null, walletAddress: user.walletAddress, ip }; + } + return { key: `ip:${ip}`, kind: 'ip', ip }; + } + + async getProfile(identity: CallerIdentity): Promise { + const cached = this.cache.get(identity.key); + if (cached && Date.now() - cached.computedAt < this.cacheTtlMs) return cached; + + const profile = await this.computeProfile(identity); + this.remember(profile); + return profile; + } + + async computeProfile(identity: CallerIdentity): Promise { + const authenticated = identity.kind !== 'ip'; + const hasDid = Boolean(identity.did); + + const [verifiedStudent, activeStake, recentViolations] = await Promise.all([ + authenticated && identity.userId ? this.isVerifiedStudent(identity.userId) : Promise.resolve(false), + authenticated && identity.walletAddress + ? this.stakingProvider.hasActiveStake(identity.walletAddress) + : Promise.resolve(false), + this.violations.count(identity.key), + ]); + + let score = 0; + if (authenticated) score += SCORE_WEIGHTS.authenticated; + if (hasDid) score += SCORE_WEIGHTS.did; + if (verifiedStudent) score += SCORE_WEIGHTS.verifiedStudent; + if (activeStake) score += SCORE_WEIGHTS.activeStake; + score += Math.min(recentViolations, MAX_COUNTED_VIOLATIONS) * SCORE_WEIGHTS.violation; + + // The top tier is reserved for authenticated DIDs with an active stake. + let tier = tierForScore(score); + if (tier === 'staker' && !(hasDid && activeStake)) tier = 'verified'; + + return { + identity, + score, + tier, + signals: { authenticated, hasDid, verifiedStudent, activeStake, recentViolations }, + computedAt: Date.now(), + }; + } + + /** Record a rejected request; the caller's cached profile is re-scored next time. */ + async recordViolation(identity: CallerIdentity): Promise { + await this.violations.increment(identity.key); + this.cache.delete(identity.key); + } + + /** Drop a cached profile (e.g. after a student earns a certificate or stakes). */ + invalidate(identityKey: string): void { + this.cache.delete(identityKey); + } + + limitsFor(tier: ReputationTier): WindowRule[] { + return TIER_LIMITS[tier]; + } + + private async loadUser(userId: string): Promise { + const cached = this.users.get(userId); + if (cached && Date.now() - cached.at < this.cacheTtlMs) return cached.user; + + const user = await getStudentById(userId); + if (this.users.size >= this.maxCacheEntries) { + const oldestKey = this.users.keys().next().value; + if (oldestKey !== undefined) this.users.delete(oldestKey); + } + this.users.set(userId, { user, at: Date.now() }); + return user; + } + + private remember(profile: ReputationProfile): void { + if (this.cache.size >= this.maxCacheEntries) { + const oldestKey = this.cache.keys().next().value; + if (oldestKey !== undefined) this.cache.delete(oldestKey); + } + this.cache.set(profile.identity.key, profile); + } +} + +export const reputationService = new ReputationService(); diff --git a/backend/src/services/tamperEvidentAuditService.ts b/backend/src/services/tamperEvidentAuditService.ts new file mode 100644 index 000000000..5ac1987ab --- /dev/null +++ b/backend/src/services/tamperEvidentAuditService.ts @@ -0,0 +1,296 @@ +/** + * Tamper-Evident Audit Service — Issue BE-009 + * + * Append-only audit log in which every entry commits to its predecessor: + * + * entryHash = SHA-256( prevHash ‖ canonicalJSON(sequence, eventType, actorId, + * entity, entityId, payload, createdAt) ) + * + * The first entry links to the all-zero genesis hash. Any change to a stored + * row — editing a field, deleting a row, re-ordering, or truncating the tail — + * breaks recomputation during `verifyChain()`, which then raises a CRITICAL + * alarm (structured log, Sentry, `tamper` event and optional webhook). + * + * Merkle roots over ranges of entry hashes are anchored on Stellar by + * `jobs/anchorMerkleRootJob.ts`; `verifyChain()` also checks that the latest + * anchored range still exists and still hashes to the published root, which + * catches an attacker who rewrites the whole chain consistently. + * + * Verification streams the chain in batches and does one SHA-256 per entry, + * comfortably under 1s per 10,000 entries. + */ + +import { createHash } from 'crypto'; +import { EventEmitter } from 'events'; +import { auditChainStore, type AuditAnchorRecord, type AuditChainRecord, type AuditChainStore } from '../db/auditChainStore.js'; +import logger from '../utils/logger.js'; +import { captureMessage } from '../utils/sentry.js'; + +export const GENESIS_HASH = '0'.repeat(64); + +export interface AuditEventInput { + eventType: string; + actorId?: string | null; + entity?: string | null; + entityId?: string | null; + payload?: unknown; +} + +export type TamperKind = 'HASH_MISMATCH' | 'CHAIN_BREAK' | 'SEQUENCE_GAP' | 'ANCHOR_MISMATCH' | 'TRUNCATED'; + +export interface TamperFinding { + kind: TamperKind; + sequence: string; + message: string; +} + +export interface ChainVerificationResult { + valid: boolean; + entriesChecked: number; + headSequence: string | null; + headHash: string | null; + durationMs: number; + findings: TamperFinding[]; +} + +export interface VerifyOptions { + batchSize?: number; + /** Stop after this many findings (default 100). */ + maxFindings?: number; + /** Raise the critical alarm on failure (default true). */ + alarm?: boolean; +} + +// ─── Hashing primitives ────────────────────────────────────────────────────── + +/** Deterministic JSON: object keys sorted recursively, `undefined` dropped. */ +export function canonicalJson(value: unknown): string { + if (value === null || typeof value !== 'object') { + return JSON.stringify(value ?? null); + } + if (value instanceof Date) return JSON.stringify(value.toISOString()); + if (Array.isArray(value)) return `[${value.map((v) => canonicalJson(v)).join(',')}]`; + + const entries = Object.entries(value as Record) + .filter(([, v]) => v !== undefined) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)); + return `{${entries.map(([k, v]) => `${JSON.stringify(k)}:${canonicalJson(v)}`).join(',')}}`; +} + +export function sha256Hex(data: string | Buffer): string { + return createHash('sha256').update(data).digest('hex'); +} + +export function computeEntryHash(entry: Omit): string { + const body = canonicalJson({ + sequence: entry.sequence.toString(), + eventType: entry.eventType, + actorId: entry.actorId, + entity: entry.entity, + entityId: entry.entityId, + payload: entry.payload, + createdAt: entry.createdAt.toISOString(), + }); + return sha256Hex(`${entry.prevHash}${body}`); +} + +/** + * Binary Merkle root over hex SHA-256 leaves. Parents are + * SHA-256(left ‖ right) over the raw 32-byte digests; an odd node at any level + * is paired with itself. Returns the genesis hash for an empty list. + */ +export function computeMerkleRoot(leafHashes: string[]): string { + if (leafHashes.length === 0) return GENESIS_HASH; + + let level = leafHashes.map((h) => Buffer.from(h, 'hex')); + while (level.length > 1) { + const next: Buffer[] = []; + for (let i = 0; i < level.length; i += 2) { + const left = level[i]!; + const right = level[i + 1] ?? left; + next.push(createHash('sha256').update(left).update(right).digest()); + } + level = next; + } + return level[0]!.toString('hex'); +} + +// ─── Service ───────────────────────────────────────────────────────────────── + +export class TamperEvidentAuditService extends EventEmitter { + /** Highest sequence this process has appended — a witness against tail truncation. */ + private highestAppended = 0n; + + constructor(private readonly store: AuditChainStore = auditChainStore) { + super(); + } + + /** Append an event to the chain. */ + async record(event: AuditEventInput): Promise { + const entry = await this.store.append((head) => { + const base = { + sequence: head ? head.sequence + 1n : 1n, + eventType: event.eventType, + actorId: event.actorId ?? null, + entity: event.entity ?? null, + entityId: event.entityId ?? null, + // Normalise through JSON so the hashed value equals what JSONB returns. + payload: JSON.parse(JSON.stringify(event.payload ?? {})), + prevHash: head ? head.entryHash : GENESIS_HASH, + createdAt: new Date(), + }; + return { ...base, entryHash: computeEntryHash(base) }; + }); + if (entry.sequence > this.highestAppended) this.highestAppended = entry.sequence; + return entry; + } + + /** + * Walk the whole chain and confirm continuity. Detects modified rows + * (hash mismatch), deleted / inserted rows (sequence gap, broken link) and + * tail truncation or wholesale rewrites (against the latest Stellar anchor). + */ + async verifyChain(options: VerifyOptions = {}): Promise { + const batchSize = options.batchSize ?? 5_000; + const maxFindings = options.maxFindings ?? 100; + const started = performance.now(); + const findings: TamperFinding[] = []; + + let expectedSequence = 1n; + let prevHash = GENESIS_HASH; + let last: AuditChainRecord | null = null; + let checked = 0; + + for (;;) { + const batch = await this.store.readBatch(last ? last.sequence : 0n, batchSize); + if (batch.length === 0) break; + + for (const entry of batch) { + checked += 1; + if (entry.sequence !== expectedSequence) { + findings.push({ + kind: 'SEQUENCE_GAP', + sequence: entry.sequence.toString(), + message: `Expected sequence ${expectedSequence}, found ${entry.sequence} — rows deleted or inserted`, + }); + } + if (entry.prevHash !== prevHash) { + findings.push({ + kind: 'CHAIN_BREAK', + sequence: entry.sequence.toString(), + message: 'prevHash does not match the preceding entry hash', + }); + } + if (computeEntryHash(entry) !== entry.entryHash) { + findings.push({ + kind: 'HASH_MISMATCH', + sequence: entry.sequence.toString(), + message: 'Entry content does not match its stored hash — row modified', + }); + } + + expectedSequence = entry.sequence + 1n; + prevHash = entry.entryHash; + last = entry; + if (findings.length >= maxFindings) break; + } + if (findings.length >= maxFindings || batch.length < batchSize) break; + } + + const headSequence = last ? last.sequence : 0n; + if (findings.length < maxFindings && headSequence < this.highestAppended) { + findings.push({ + kind: 'TRUNCATED', + sequence: this.highestAppended.toString(), + message: `Chain ends at ${headSequence} but this instance appended up to ${this.highestAppended}`, + }); + } + if (findings.length < maxFindings) { + findings.push(...(await this.checkLatestAnchor(last))); + } + + const result: ChainVerificationResult = { + valid: findings.length === 0, + entriesChecked: checked, + headSequence: last ? last.sequence.toString() : null, + headHash: last ? last.entryHash : null, + durationMs: Math.round((performance.now() - started) * 100) / 100, + findings, + }; + + if (!result.valid && options.alarm !== false) { + await this.raiseAlarm(result); + } + return result; + } + + /** Merkle root over the entry hashes of `[fromSequence, toSequence]`. */ + async computeRangeRoot(fromSequence: bigint, toSequence: bigint): Promise<{ root: string; count: number }> { + const rows = await this.store.readHashes(fromSequence, toSequence); + return { root: computeMerkleRoot(rows.map((r) => r.entryHash)), count: rows.length }; + } + + /** Recompute an anchor's range locally and compare with the stored root. */ + async verifyAnchorLocally(anchor: AuditAnchorRecord): Promise { + const { root, count } = await this.computeRangeRoot(anchor.fromSequence, anchor.toSequence); + return root === anchor.merkleRoot && count === anchor.entryCount; + } + + private async checkLatestAnchor(last: AuditChainRecord | null): Promise { + const anchor = await this.store.getLatestAnchor(); + if (!anchor) return []; + + if (!last || last.sequence < anchor.toSequence) { + return [ + { + kind: 'TRUNCATED', + sequence: anchor.toSequence.toString(), + message: `Chain ends at ${last?.sequence ?? 0} but entries up to ${anchor.toSequence} were anchored on Stellar`, + }, + ]; + } + + if (!(await this.verifyAnchorLocally(anchor))) { + return [ + { + kind: 'ANCHOR_MISMATCH', + sequence: anchor.toSequence.toString(), + message: `Entries ${anchor.fromSequence}-${anchor.toSequence} no longer match the Merkle root anchored in tx ${anchor.txHash}`, + }, + ]; + } + return []; + } + + private async raiseAlarm(result: ChainVerificationResult): Promise { + const summary = { + severity: 'CRITICAL', + alarm: 'AUDIT_CHAIN_TAMPERED', + findings: result.findings.slice(0, 10), + totalFindings: result.findings.length, + headSequence: result.headSequence, + }; + + logger.error('CRITICAL: audit hash chain verification failed — possible tampering', summary); + captureMessage('Audit hash chain tampering detected', 'error', summary); + this.emit('tamper', result); + + const webhook = process.env.AUDIT_ALARM_WEBHOOK_URL; + if (webhook) { + try { + await fetch(webhook, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ text: 'CRITICAL: audit hash chain tampering detected', ...summary }), + signal: AbortSignal.timeout(5_000), + }); + } catch (error) { + logger.error('Failed to deliver audit tamper alarm webhook', { + error: error instanceof Error ? error.message : String(error), + }); + } + } + } +} + +export const tamperEvidentAuditService = new TamperEvidentAuditService(); diff --git a/backend/src/utils/audit.ts b/backend/src/utils/audit.ts index 560dcbbc6..5230b6706 100644 --- a/backend/src/utils/audit.ts +++ b/backend/src/utils/audit.ts @@ -1,6 +1,7 @@ import { Request } from 'express'; import { createHash } from 'crypto'; import prisma from '../db/index.js'; +import { tamperEvidentAuditService } from '../services/tamperEvidentAuditService.js'; import logger, { auditLogger, getCorrelationId } from './logger.js'; export interface AuditLogData { @@ -99,6 +100,20 @@ export async function logAudit(data: AuditLogData): Promise { }, }); + // Mirror into the tamper-evident hash chain (BE-009). A failure here is + // logged loudly but never blocks the action being audited. + try { + await tamperEvidentAuditService.record({ + eventType: data.action, + actorId: data.userId ?? null, + entity: data.entity ?? null, + entityId: data.entityId ?? null, + payload: { ...payload, timestamp, legacyHash: hash }, + }); + } catch (chainError) { + logger.error('Failed to append audit event to hash chain', { error: chainError, correlationId }); + } + logger.info( `Audit Log: ${data.action} by ${data.userEmail ?? data.userId ?? 'unknown'}`, { diff --git a/backend/src/utils/redis-pubsub.ts b/backend/src/utils/redis-pubsub.ts index 38a30d209..856af5c68 100644 --- a/backend/src/utils/redis-pubsub.ts +++ b/backend/src/utils/redis-pubsub.ts @@ -13,6 +13,8 @@ * wsSync.subscribe((message) => { /* broadcast to local WS clients *\/ }); */ +import { extractTraceContext, injectTraceContext, runInSpan, SpanKind, type TraceCarrier } from '../instrumentation.js'; +import { pubsubMessages } from '../metrics/prometheusExporter.js'; import { redisConnection } from './redis.js'; export interface WsMessage { @@ -50,12 +52,24 @@ export function createWsSync( async function publish(message: WsMessage): Promise { try { - const payload = JSON.stringify({ - ...message, - _channel: channelName, - _timestamp: Date.now(), - }); - await redis.publish(channelName, payload); + await runInSpan( + `${channelName} publish`, + { + kind: SpanKind.PRODUCER, + attributes: { 'messaging.system': 'redis', 'messaging.destination.name': channelName }, + }, + async () => { + const payload = JSON.stringify({ + ...message, + _channel: channelName, + _timestamp: Date.now(), + // W3C trace-context so subscribers continue the publisher's trace + _trace: injectTraceContext(), + }); + await redis.publish(channelName, payload); + }, + ); + pubsubMessages.inc({ channel: channelName, direction: 'published' }); } catch (err) { console.error(`[ws-sync] Publish error on "${channelName}":`, err); } @@ -74,14 +88,25 @@ export function createWsSync( const parsed = JSON.parse(message) as WsMessage; // Skip our own messages (loop prevention) if (parsed._sender === process.env.NODE_ID) return; - // Notify all local handlers - for (const sub of subscribers) { - try { - sub(parsed); - } catch (err) { - console.error(`[ws-sync] Handler error on "${channelName}":`, err); - } - } + pubsubMessages.inc({ channel: channelName, direction: 'received' }); + // Notify all local handlers inside a CONSUMER span linked to the publisher + runInSpan( + `${channelName} receive`, + { + kind: SpanKind.CONSUMER, + parent: extractTraceContext(parsed._trace as TraceCarrier | undefined), + attributes: { 'messaging.system': 'redis', 'messaging.destination.name': channelName }, + }, + () => { + for (const sub of subscribers) { + try { + sub(parsed); + } catch (err) { + console.error(`[ws-sync] Handler error on "${channelName}":`, err); + } + } + }, + ); } catch { // Ignore malformed messages } diff --git a/backend/src/utils/sentry.ts b/backend/src/utils/sentry.ts index 01d2df539..dded48d73 100644 --- a/backend/src/utils/sentry.ts +++ b/backend/src/utils/sentry.ts @@ -1,5 +1,6 @@ import type { ErrorRequestHandler, RequestHandler, Application } from 'express'; import * as Sentry from '@sentry/node'; +import { isTracingEnabled } from '../instrumentation.js'; import { redactSensitiveData } from './logSanitizer.js'; let sentryEnabled = false; @@ -31,6 +32,9 @@ export function initializeSentry(app?: Application): void { tracesSampleRate: process.env.NODE_ENV === 'production' ? 0.1 : 1.0, attachStacktrace: true, normalizeDepth: 5, + // The OpenTelemetry SDK (src/instrumentation.ts) owns the global tracer + // provider and exports to Jaeger; Sentry keeps error tracking. + skipOpenTelemetrySetup: isTracingEnabled(), beforeSend(event) { if (process.env.NODE_ENV === 'test') { return null; diff --git a/backend/src/utils/slidingWindow.ts b/backend/src/utils/slidingWindow.ts new file mode 100644 index 000000000..fb948933f --- /dev/null +++ b/backend/src/utils/slidingWindow.ts @@ -0,0 +1,232 @@ +/** + * Sliding-Window Log — Issue BE-012 + * + * Exact sliding-window-log rate limiting on a Redis sorted set: every accepted + * request is a member scored by its timestamp, entries older than the window + * are trimmed on each check, and a request is admitted only while fewer than + * `limit` entries remain. Because the window slides continuously there is no + * boundary at which the counter resets, so callers cannot double up a burst + * across two fixed windows. + * + * Several windows (e.g. a 1s burst window and a 60s sustained window) are + * evaluated atomically in a single Lua call: a request is recorded in all of + * them or in none, so rejected requests never consume quota. + * + * An in-memory implementation with identical semantics is used when Redis is + * unavailable (and in tests). + */ + +import { randomUUID } from 'crypto'; + +export interface WindowRule { + /** Short name for headers / logs, e.g. `burst` or `sustained`. */ + name: string; + /** Maximum requests admitted inside the window. */ + limit: number; + /** Window length in milliseconds. */ + windowMs: number; +} + +export interface WindowState { + name: string; + limit: number; + windowMs: number; + /** Requests counted in the window, including this one if admitted. */ + count: number; + remaining: number; + /** Milliseconds until the oldest counted request leaves the window. */ + resetMs: number; +} + +export interface SlidingWindowResult { + allowed: boolean; + /** The most restrictive window (lowest remaining / the one that blocked). */ + binding: WindowState; + windows: WindowState[]; + /** Milliseconds until a retry can succeed; 0 when allowed. */ + retryAfterMs: number; +} + +/** Minimal Redis surface needed (ioredis-compatible). */ +export interface SlidingWindowRedis { + eval(script: string, numKeys: number, ...args: Array): Promise; +} + +export interface SlidingWindowStore { + hit(key: string, rules: WindowRule[], now?: number): Promise; +} + +/** + * KEYS[i] = sorted set for rule i + * ARGV[1] = now (ms), ARGV[2] = unique member id + * ARGV[3+2(i-1)] = windowMs_i, ARGV[4+2(i-1)] = limit_i + * Returns a flat array: allowed, then per rule {count, oldestScore}. + */ +const SLIDING_WINDOW_LOG_SCRIPT = ` +local now = tonumber(ARGV[1]) +local member = ARGV[2] +local n = #KEYS +local counts = {} +local oldest = {} +local allowed = 1 + +for i = 1, n do + local window = tonumber(ARGV[1 + 2 * i]) + local limit = tonumber(ARGV[2 + 2 * i]) + redis.call('ZREMRANGEBYSCORE', KEYS[i], '-inf', now - window) + local count = redis.call('ZCARD', KEYS[i]) + counts[i] = count + local first = redis.call('ZRANGE', KEYS[i], 0, 0, 'WITHSCORES') + oldest[i] = first[2] or now + if count >= limit then + allowed = 0 + end +end + +if allowed == 1 then + for i = 1, n do + local window = tonumber(ARGV[1 + 2 * i]) + redis.call('ZADD', KEYS[i], now, member) + redis.call('PEXPIRE', KEYS[i], window) + counts[i] = counts[i] + 1 + end +end + +local out = { allowed } +for i = 1, n do + table.insert(out, counts[i]) + table.insert(out, tostring(oldest[i])) +end +return out +`; + +function buildResult( + rules: WindowRule[], + allowed: boolean, + counts: number[], + oldest: number[], + now: number, +): SlidingWindowResult { + const windows: WindowState[] = rules.map((rule, i) => { + const count = counts[i] ?? 0; + const oldestAt = oldest[i] ?? now; + return { + name: rule.name, + limit: rule.limit, + windowMs: rule.windowMs, + count, + remaining: Math.max(0, rule.limit - count), + resetMs: Math.max(0, oldestAt + rule.windowMs - now), + }; + }); + + const blocking = windows.filter((w) => w.count >= w.limit); + let retryAfterMs = 0; + let binding: WindowState; + + if (!allowed && blocking.length > 0) { + // Retry once every exhausted window has freed a slot. + binding = blocking.reduce((a, b) => (b.resetMs > a.resetMs ? b : a)); + retryAfterMs = Math.max(1, binding.resetMs); + } else { + binding = windows.reduce((a, b) => (b.remaining / b.limit < a.remaining / a.limit ? b : a)); + } + + return { allowed, binding, windows, retryAfterMs }; +} + +function assertRules(rules: WindowRule[]): void { + if (rules.length === 0) throw new Error('slidingWindow: at least one window rule is required'); + for (const rule of rules) { + if (!(rule.limit > 0) || !(rule.windowMs > 0)) { + throw new Error(`slidingWindow: invalid rule "${rule.name}"`); + } + } +} + +export class RedisSlidingWindowStore implements SlidingWindowStore { + constructor(private readonly redis: SlidingWindowRedis) {} + + async hit(key: string, rules: WindowRule[], now = Date.now()): Promise { + assertRules(rules); + // Hash tag keeps every window of one identity on the same cluster slot. + const keys = rules.map((rule) => `{${key}}:${rule.name}`); + const args: Array = [now, `${now}-${randomUUID()}`]; + for (const rule of rules) args.push(rule.windowMs, rule.limit); + + const raw = (await this.redis.eval(SLIDING_WINDOW_LOG_SCRIPT, keys.length, ...keys, ...args)) as Array< + number | string + >; + + const allowed = Number(raw[0]) === 1; + const counts: number[] = []; + const oldest: number[] = []; + for (let i = 0; i < rules.length; i += 1) { + counts.push(Number(raw[1 + i * 2])); + oldest.push(Number(raw[2 + i * 2])); + } + return buildResult(rules, allowed, counts, oldest, now); + } +} + +export class MemorySlidingWindowStore implements SlidingWindowStore { + private readonly logs = new Map(); + + async hit(key: string, rules: WindowRule[], now = Date.now()): Promise { + assertRules(rules); + const counts: number[] = []; + const oldest: number[] = []; + let allowed = true; + + const logs = rules.map((rule) => { + const logKey = `${key}:${rule.name}`; + const log = (this.logs.get(logKey) ?? []).filter((ts) => ts > now - rule.windowMs); + this.logs.set(logKey, log); + counts.push(log.length); + oldest.push(log[0] ?? now); + if (log.length >= rule.limit) allowed = false; + return log; + }); + + if (allowed) { + logs.forEach((log, i) => { + log.push(now); + counts[i] = log.length; + }); + } + + this.evictIdle(now, rules); + return buildResult(rules, allowed, counts, oldest, now); + } + + /** Drop empty logs so the map does not grow without bound. */ + private evictIdle(now: number, rules: WindowRule[]): void { + if (this.logs.size < 10_000) return; + const maxWindow = Math.max(...rules.map((r) => r.windowMs)); + for (const [key, log] of this.logs) { + if (log.length === 0 || (log[log.length - 1] ?? 0) <= now - maxWindow) this.logs.delete(key); + } + } + + clear(): void { + this.logs.clear(); + } +} + +/** + * Redis-backed store that degrades to an in-process log if Redis errors, so + * an outage weakens (per-instance) limiting instead of disabling it. + */ +export class ResilientSlidingWindowStore implements SlidingWindowStore { + private readonly fallback = new MemorySlidingWindowStore(); + + constructor(private readonly primary: SlidingWindowStore) {} + + async hit(key: string, rules: WindowRule[], now = Date.now()): Promise { + try { + return await this.primary.hit(key, rules, now); + } catch { + return this.fallback.hit(key, rules, now); + } + } +} diff --git a/backend/src/workers/compileWorker.ts b/backend/src/workers/compileWorker.ts index a5d4358b0..97fa76403 100644 --- a/backend/src/workers/compileWorker.ts +++ b/backend/src/workers/compileWorker.ts @@ -23,6 +23,7 @@ import { Worker, type Job } from 'bullmq'; import { exec } from 'child_process'; import { promisify } from 'util'; import { broadcastEvent } from '../websocket/gateway.js'; +import { traceJob } from '../instrumentation.js'; import logger from '../utils/logger.js'; import { COMPILE_QUEUE_NAME, @@ -244,7 +245,7 @@ async function processCompileJob(job: Job): Pr export const compileWorker = new Worker( COMPILE_QUEUE_NAME, - processCompileJob, + (job) => traceJob(COMPILE_QUEUE_NAME, job, () => processCompileJob(job)), { connection: redisConnection, // Concurrency throttle: limit simultaneous builds per process