From d7e093663afacbc5a868e07aa4a8a82b1342d594 Mon Sep 17 00:00:00 2001 From: Tom Scanlan Date: Fri, 5 Jun 2026 10:21:47 -0400 Subject: [PATCH] feat(contrail): private-network deployment + concurrent-init hardening (#44) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(contrail-base): networkOverrides config for private-network deployments Add optional ContrailConfig.networkOverrides with three subfields: plcUrl, slingshotUrl, additionalAllowedHosts. Plumb config through resolvePDS, getPDSViaDidDoc (inlined; module-level didResolver removed), getPDS, resolvePDSCached, getClient, and all five identity.ts functions (fetchAndSave, resolveIdentity, resolveIdentities, resolveActor, refreshStaleIdentities). Pass config through backfill.ts call sites (getClient + getPDS) and the persistent.ts:160 refreshStaleIdentities call. additionalAllowedHosts is exact + case-insensitive + port-agnostic; the default SSRF validator still runs for non-listed hosts. All changes are additive — omitting networkOverrides preserves current public-internet defaults. validatePdsUrl remains internal (no new exports beyond the networkOverrides field itself). * test(contrail-base): network overrides + SSRF regression coverage Add three test files exercising the networkOverrides plumbing: - client.test.ts: ContrailConfig.networkOverrides type, validatePdsUrl regression baseline (public HTTPS accepted; non-HTTPS + private CIDRs + localhost + link-local rejected), additionalAllowedHosts allowlist (allowed host accepted; non-listed hosts still rejected; port-agnostic match), getClient + getPDS + resolvePDS plumb-through with mocked fetch. - identity-config.test.ts: resolveIdentity, resolveIdentities, resolveActor, refreshStaleIdentities all route to override slingshot URL. - network-overrides.test.ts: integration test against node:http stub PLC and Slingshot servers. Asserts resolvePDS, getClient, and refreshStaleIdentities actually hit the configured override URLs end to end, exercising the full chain. The SSRF regression cases give the existing validator implicit baseline explicit coverage in this PR. * feat(contrail-base): apply networkOverrides.resolver to spaces credential verifier * feat(contrail-appview): apply networkOverrides to label-endpoint resolution * fix(contrail-appview): use IF NOT EXISTS on schema migrations to remove init race Concurrent `initSchema` calls previously raced on `ALTER TABLE ADD COLUMN` because the SQL had no `IF NOT EXISTS` clause, and the two consumer-site `try { ... } catch { /* Column already exists — ignore */ }` blocks silently swallowed *every* DDL error — masking missing tables, syntax errors, and type mismatches alongside the intended duplicate-column case. L3 replaces both swallows with a dialect-aware `addColumnIfNotExists` helper: - Postgres: emits `ALTER TABLE ... ADD COLUMN IF NOT EXISTS` (atomic in PG). - SQLite: pre-checks `PRAGMA table_info`, then narrowly absorbs only the literal "duplicate column name" error from the still-possible PRAGMA->ALTER race. Every other error surfaces. The `MIGRATIONS` array is now structured `{ table, column, columnDef, target }` data routed through the same helper. `target: "spaces"` routes to `spacesDb` when one is configured (split-DB deployments), `target: "feeds"` is gated on feeds being configured so the absent `feed_backfills` table no longer silently fails. Tests at `packages/contrail/tests/schema-idempotency.test.ts` cover: - sequential and concurrent (3-way) `initSchema` calls - repeated init does not leave duplicate count columns - helper is a no-op when column already exists - helper surfaces "no such table" errors (the L3-removed swallow case) - `initSchema` propagates errors thrown from extension schema modules OpenMeet's consumer-side Postgres advisory lock around `contrail.init()` exists only to work around this race; with L3 in place that lock can be deleted in a follow-up. * test: L3 concurrent CREATE race on Postgres * chore: add changeset for private-network deployment support * fix(contrail-appview): thread networkOverrides config through all resolver/identity call sites The networkOverrides config param was threaded into the resolver/identity functions but dropped at ~8 appview call sites (live-ingest refresh cycle, on-demand refresh, and router getProfile/getFeed/collection/profiles/notify), so private-network deploys silently fell back to the public resolver and the un-widened SSRF guard on those paths. Pass the in-scope config at each site. Add a regression test driving an override config through the live-ingest refresh path (runIngestCycle) and a router actor-resolution path (getProfile), asserting the override slingshot is hit instead of the default. * refactor(contrail-base): dedup SSRF validator into shared validateExternalUrl labels/resolve.ts validateEndpointUrl duplicated contrail-base validatePdsUrl, and this PR had edited the additionalAllowedHosts allowlist into both copies — a future SSRF-rule fix applied to one copy would leave the other exploitable. Export a single validateExternalUrl(url, allowedHosts?) from contrail-base and consume it from both the PDS client and labeler-endpoint resolution. validateEndpointUrl stays exported as a thin alias for backward compat. Behavior is identical (covered by existing SSRF tests). Also document the PDS resolve cache's process-wide-config assumption (keyed by DID only; safe now that all callers thread the same config). * docs(contrail-base): document validateExternalUrl threat-model scope Note inline that the SSRF guard is best-effort literal-match only: no DNS resolution, partial IPv6 / encoded-IP coverage, egress network policy expected for fully untrusted resolver inputs. No behavior change. --- .changeset/private-network-overrides.md | 19 ++ .../contrail-appview/src/core/backfill.ts | 8 +- .../contrail-appview/src/core/db/schema.ts | 236 ++++++++++++++--- .../contrail-appview/src/core/jetstream.ts | 2 +- .../src/core/labels/resolve.ts | 69 +++-- .../src/core/labels/subscribe.ts | 34 ++- .../contrail-appview/src/core/persistent.ts | 2 +- packages/contrail-appview/src/core/refresh.ts | 2 +- .../src/core/router/collection.ts | 2 +- .../contrail-appview/src/core/router/feed.ts | 2 +- .../contrail-appview/src/core/router/index.ts | 4 +- .../src/core/router/notify.ts | 2 +- .../src/core/router/profiles.ts | 4 +- .../src/core/spaces/router.ts | 2 +- .../src/core/spaces/schema.ts | 15 +- packages/contrail-base/src/client.ts | 111 +++++--- packages/contrail-base/src/identity.ts | 27 +- packages/contrail-base/src/spaces/auth.ts | 18 +- packages/contrail-base/src/types.ts | 31 +++ .../contrail-community/src/integration.ts | 2 +- packages/contrail/tests/client.test.ts | 239 ++++++++++++++++++ .../contrail/tests/identity-config.test.ts | 87 +++++++ .../contrail/tests/labels-resolve.test.ts | 81 ++++++ .../tests/network-overrides-appview.test.ts | 96 +++++++ .../contrail/tests/network-overrides.test.ts | 189 ++++++++++++++ .../tests/postgres-concurrent-init.test.ts | 92 +++++++ .../contrail/tests/schema-idempotency.test.ts | 139 ++++++++++ packages/contrail/tests/spaces-auth.test.ts | 65 +++++ 28 files changed, 1450 insertions(+), 130 deletions(-) create mode 100644 .changeset/private-network-overrides.md create mode 100644 packages/contrail/tests/client.test.ts create mode 100644 packages/contrail/tests/identity-config.test.ts create mode 100644 packages/contrail/tests/labels-resolve.test.ts create mode 100644 packages/contrail/tests/network-overrides-appview.test.ts create mode 100644 packages/contrail/tests/network-overrides.test.ts create mode 100644 packages/contrail/tests/postgres-concurrent-init.test.ts create mode 100644 packages/contrail/tests/schema-idempotency.test.ts create mode 100644 packages/contrail/tests/spaces-auth.test.ts diff --git a/.changeset/private-network-overrides.md b/.changeset/private-network-overrides.md new file mode 100644 index 0000000..874794e --- /dev/null +++ b/.changeset/private-network-overrides.md @@ -0,0 +1,19 @@ +--- +"@atmo-dev/contrail-base": minor +"@atmo-dev/contrail-appview": minor +"@atmo-dev/contrail-community": minor +--- + +Private-network deployment support via a new optional `ContrailConfig.networkOverrides` block. + +`networkOverrides` carries three optional subfields, all defaulting to the current public-internet behavior (omit the block entirely and nothing changes): + +- **`resolver`** — a custom `DidDocumentResolver` used during DID-doc PDS fallback, labeler-endpoint resolution, and spaces service-auth JWT verification. Lets a deployment point at a private PLC mirror or inject a custom fetch (mTLS, retry, instrumentation). Trusted; not SSRF-checked. +- **`slingshotUrl`** — override the slingshot identity-resolver endpoint. Trusted; not SSRF-checked. +- **`additionalAllowedHosts`** — hostnames that bypass the default SSRF guard when validating a resolved PDS or labeler endpoint. Match is exact, case-insensitive, port-agnostic (e.g. `["pds.dev.svc.cluster.local"]`). This is the only knob that widens the validator; there is no "disable SSRF" flag. + +The overrides are threaded through PDS/identity resolution (`resolvePDS`, `getPDS`, `getClient`, `resolveIdentity*`, `refreshStaleIdentities`), labeler endpoint resolution and ingest (`resolveLabelerEndpoint`, `getLabelerState`, label subscribe cycles), and service-auth verification (`buildVerifier` in both the appview router and the community integration). The in-scope `config` is now also passed at every appview call site that resolves identities or PDS endpoints — the live-ingest refresh cycle (`runIngestCycle` → `refreshStaleIdentities`), the on-demand `refresh` path, and the router actor/identity/PDS resolution paths (`getProfile`, `getFeed`, collection queries, profile hydration, notify) — so private-network deploys honor the override on those paths instead of silently falling back to the public resolver and un-widened SSRF guard. + +The SSRF guard is now a single shared validator: `validateExternalUrl(url, additionalAllowedHosts?)` is exported from `contrail-base` and consumed by both the PDS client and labeler-endpoint resolution. `validateEndpointUrl` remains exported as a thin alias for backward compatibility. This removes the previous duplicate validator (`validatePdsUrl` + `validateEndpointUrl`) where an allowlist or SSRF-rule edit could be applied to only one copy. + +Also hardens schema initialization for concurrent/Postgres deployments: a dialect-aware `addColumnIfNotExists` (Postgres `ADD COLUMN IF NOT EXISTS`; SQLite pre-check), narrow absorption of the Postgres concurrent-`CREATE` race (42P07 / 23505 on pg_type/pg_class/pg_namespace indexes), and per-statement (rather than batched) DDL during `initSchema` / `initSpacesSchema` / spaces schema. Genuine DDL errors (syntax, type mismatch, missing column/table) still propagate. diff --git a/packages/contrail-appview/src/core/backfill.ts b/packages/contrail-appview/src/core/backfill.ts index a0474a7..f7fdd8b 100644 --- a/packages/contrail-appview/src/core/backfill.ts +++ b/packages/contrail-appview/src/core/backfill.ts @@ -186,7 +186,7 @@ export async function backfillUser( if (!client) { try { client = await withRetry( - () => getClient(did as Did, db), + () => getClient(did as Did, db, config), `getClient(${did})`, Math.min(retries, 1), timeout @@ -356,7 +356,7 @@ export async function backfillPending( for (let i = 0; i < dids.length; i += 200) { await Promise.allSettled( dids.slice(i, i + 200).map((did) => - getPDS(did as Did, db).catch(() => {}) + getPDS(did as Did, db, config).catch(() => {}) ) ); } @@ -386,7 +386,7 @@ export async function backfillPending( let client: Client | undefined; try { client = await withRetry( - () => getClient(did as Did, db), + () => getClient(did as Did, db, config), `getClient(${did})`, 0, FAST_TIMEOUT @@ -436,7 +436,7 @@ export async function backfillPending( let client: Client | undefined; try { client = await withRetry( - () => getClient(did as Did, db), + () => getClient(did as Did, db, config), `getClient(${did})`, 2 ); diff --git a/packages/contrail-appview/src/core/db/schema.ts b/packages/contrail-appview/src/core/db/schema.ts index c1778f0..df93e72 100644 --- a/packages/contrail-appview/src/core/db/schema.ts +++ b/packages/contrail-appview/src/core/db/schema.ts @@ -1,6 +1,6 @@ import type { ContrailConfig, Database, ResolvedContrailConfig, ResolvedMaps } from "../types"; import type { SqlDialect } from "../dialect"; -import { buildFtsSchema, getDialect } from "../dialect"; +import { buildFtsSchema, getDialect, postgresDialect } from "../dialect"; import { getRelationField, countColumnName, @@ -197,6 +197,135 @@ export function buildCountColumns(config: ContrailConfig, opts: BuilderOpts = {} return stmts; } +/** + * Idempotently add a column to a table, surfacing real DDL errors. + * + * Postgres supports `ALTER TABLE ... ADD COLUMN IF NOT EXISTS` natively, so we + * issue that and let any non-duplicate error propagate. SQLite (including + * `node:sqlite`) does NOT support `IF NOT EXISTS` on `ADD COLUMN`, so we + * pre-check `PRAGMA table_info` and short-circuit if the column is already + * there. Because the PRAGMA-check + ALTER pair is not atomic, a concurrent + * second `initSchema` call can still hit a "duplicate column name" race; we + * narrowly absorb exactly that error message and re-throw everything else. + * + * Net effect: only the duplicate-column case is absorbed. Missing tables, + * syntax errors, type mismatches, and any other DDL failure will throw. + * + * Exported for direct testing of the idempotency contract; callers in + * `initSchema` use this internally. + */ +export async function addColumnIfNotExists( + db: Database, + table: string, + column: string, + columnDef: string, +): Promise { + const dialect = getDialect(db); + if (dialect === postgresDialect) { + await db.prepare( + `ALTER TABLE ${table} ADD COLUMN IF NOT EXISTS ${column} ${columnDef}`, + ).run(); + return; + } + // SQLite path: check existence first, then ALTER without IF NOT EXISTS. + // PRAGMA table_info() does not accept parameter binding, so we rely on the + // caller to pass a sanitized identifier (all current callers do — table + // names come from `recordsTableName`/`spacesRecordsTableName` which + // sanitize, and column names come from `countColumnName` / + // `groupedCountColumnName` which also sanitize). + const info = await db + .prepare(`PRAGMA table_info(${table})`) + .all<{ name: string }>(); + if (info.results.some((c) => c.name === column)) return; + try { + await db.prepare( + `ALTER TABLE ${table} ADD COLUMN ${column} ${columnDef}`, + ).run(); + } catch (err) { + // Narrow swallow: only the "duplicate column" race between the PRAGMA + // read and the ALTER is acceptable. Everything else surfaces. + if (!isDuplicateColumnError(err)) throw err; + } +} + +function isDuplicateColumnError(err: unknown): boolean { + if (!err || typeof err !== "object") return false; + const msg = (err as { message?: unknown }).message; + if (typeof msg !== "string") return false; + // node:sqlite / better-sqlite3: "duplicate column name: " + return /duplicate column name/i.test(msg); +} + +/** + * Postgres `CREATE TABLE IF NOT EXISTS` and `CREATE INDEX IF NOT EXISTS` are + * NOT atomic against concurrent creators: two transactions can both pass the + * existence check before either has inserted into `pg_class` / `pg_type`. The + * loser raises 23505 on `pg_type_typname_nsp_index` (the unique index on + * `(typname, typnamespace)`) or `pg_class_relname_nsp_index`. Pre-existing + * tables also surface as 42P07 (`duplicate_table`). + * + * SQLite serializes DDL globally, so this race never manifests there. + * + * The caller is expected to issue idempotent DDL (IF NOT EXISTS); this helper + * only absorbs the narrow concurrent-create race. + */ +function isConcurrentCreateError(err: unknown): boolean { + if (!err || typeof err !== "object") return false; + const code = (err as { code?: unknown }).code; + if (code === "42P07" || code === "42P06") return true; + if (code === "23505") { + const constraint = (err as { constraint?: unknown }).constraint; + return ( + constraint === "pg_type_typname_nsp_index" || + constraint === "pg_class_relname_nsp_index" || + constraint === "pg_namespace_nspname_index" + ); + } + return false; +} + +/** + * Run a single DDL statement, absorbing only the concurrent-create race that + * `CREATE TABLE IF NOT EXISTS` / `CREATE INDEX IF NOT EXISTS` can hit on + * Postgres when multiple processes init the same schema in parallel. Genuine + * DDL errors (syntax, type mismatch, missing column) surface unchanged. + */ +async function runIdempotentDdl(db: Database, stmt: string): Promise { + try { + await db.prepare(stmt).run(); + } catch (err) { + if (!isConcurrentCreateError(err)) throw err; + } +} + +/** + * Apply the ALTER+INDEX statements emitted by `buildCountColumns` + * idempotently and without swallowing non-duplicate errors. + * + * `buildCountColumns` mixes two statement shapes: `ALTER TABLE ... ADD COLUMN + * ...` (not idempotent on SQLite without a pre-check; supports IF NOT EXISTS + * on Postgres) and `CREATE INDEX IF NOT EXISTS ...` (idempotent on both + * dialects). We route ALTERs through `addColumnIfNotExists` and run indexes + * directly. + */ +export async function applyCountColumns( + db: Database, + config: ContrailConfig, + opts: BuilderOpts = {}, +): Promise { + for (const stmt of buildCountColumns(config, opts)) { + const match = stmt.match( + /^ALTER TABLE\s+(\S+)\s+ADD COLUMN\s+(\S+)\s+(.+)$/i, + ); + if (match) { + const [, table, column, columnDef] = match; + await addColumnIfNotExists(db, table, column, columnDef); + } else { + await db.prepare(stmt).run(); + } + } +} + function buildFeedTables(config: ContrailConfig, dialect: SqlDialect): string[] { if (!config.feeds || Object.keys(config.feeds).length === 0) return []; const stmts = [ @@ -250,22 +379,56 @@ export function buildFtsTables( return stmts; } -const MIGRATIONS = [ - "ALTER TABLE backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0", - "ALTER TABLE backfills ADD COLUMN last_error TEXT", - "ALTER TABLE spaces_invites ADD COLUMN kind TEXT NOT NULL DEFAULT 'join'", - "ALTER TABLE feed_backfills ADD COLUMN retries INTEGER NOT NULL DEFAULT 0", - "ALTER TABLE feed_backfills ADD COLUMN last_error TEXT", - "ALTER TABLE feed_backfills ADD COLUMN started_at BIGINT", +/** + * Schema migrations expressed as structured ADD-COLUMN ops. Each entry is + * applied via `addColumnIfNotExists` so the operation is idempotent on both + * dialects without swallowing genuine DDL errors. + * + * `target: "spaces"` is routed to the spaces DB (which may differ from the + * main DB in split-DB deployments) and only applied when spaces is enabled. + * `target: "feeds"` is only applied when feeds are configured (the + * `feed_backfills` table doesn't exist otherwise). All other migrations + * target the main DB unconditionally. + */ +interface MigrationOp { + table: string; + column: string; + columnDef: string; + target?: "spaces" | "feeds"; +} + +const MIGRATIONS: MigrationOp[] = [ + { table: "backfills", column: "retries", columnDef: "INTEGER NOT NULL DEFAULT 0" }, + { table: "backfills", column: "last_error", columnDef: "TEXT" }, + { + table: "spaces_invites", + column: "kind", + columnDef: "TEXT NOT NULL DEFAULT 'join'", + target: "spaces", + }, + { table: "feed_backfills", column: "retries", columnDef: "INTEGER NOT NULL DEFAULT 0", target: "feeds" }, + { table: "feed_backfills", column: "last_error", columnDef: "TEXT", target: "feeds" }, + { table: "feed_backfills", column: "started_at", columnDef: "BIGINT", target: "feeds" }, ]; -async function runMigrations(db: Database): Promise { - for (const sql of MIGRATIONS) { - try { - await db.prepare(sql).run(); - } catch { - // Column already exists — ignore +async function runMigrations( + db: Database, + spacesDb: Database | undefined, + hasSpaces: boolean, + hasFeeds: boolean, +): Promise { + for (const op of MIGRATIONS) { + if (op.target === "spaces") { + if (!hasSpaces) continue; + await addColumnIfNotExists(spacesDb ?? db, op.table, op.column, op.columnDef); + continue; + } + if (op.target === "feeds") { + if (!hasFeeds) continue; + await addColumnIfNotExists(db, op.table, op.column, op.columnDef); + continue; } + await addColumnIfNotExists(db, op.table, op.column, op.columnDef); } } @@ -290,15 +453,20 @@ async function applySpacesSchema( const base = buildSpacesBaseSchema(dialect); const perCollection = buildCollectionTables(config, dialect, { forSpaces: true }); const indexes = buildDynamicIndexes(config, dialect, { forSpaces: true }); - await target.batch([...base, ...perCollection, ...indexes].map((s) => target.prepare(s))); + // Per-statement (not batched) so concurrent applySpacesSchema on Postgres + // races only on the individual CREATE statements; see initSchema for + // rationale. + for (const stmt of [...base, ...perCollection, ...indexes]) { + await runIdempotentDdl(target, stmt); + } const ftsStmts = buildFtsTables(config, dialect, { forSpaces: true }); for (const stmt of ftsStmts) { try { await target.prepare(stmt).run(); } catch { /* FTS5 unavailable */ } } - for (const stmt of buildCountColumns(config, { forSpaces: true })) { - try { await target.prepare(stmt).run(); } catch { /* already exists */ } - } + // Idempotent count-column ALTERs + their indexes. Non-duplicate-column + // errors propagate. + await applyCountColumns(target, config, { forSpaces: true }); } export async function initSchema( @@ -320,7 +488,13 @@ export async function initSchema( const all = [...baseStatements, ...collectionStatements, ...indexStatements, ...feedStatements]; - await db.batch(all.map((s) => db.prepare(s))); + // Per-statement run (not a batched transaction) so concurrent initSchema + // callers on Postgres race only on individual CREATEs; the loser's + // duplicate-relation error is absorbed by runIdempotentDdl. Each statement + // is already idempotent (IF NOT EXISTS). + for (const stmt of all) { + await runIdempotentDdl(db, stmt); + } if (config.spaces?.authority || config.spaces?.recordHost) { await applySpacesSchema(spacesSharesMainDb ? db : spacesDb!, config, dialect); @@ -339,7 +513,9 @@ export async function initSchema( // Labels tables live on the main DB — they're keyed by at-URI / DID and // are read alongside public records during hydration. const labelsStmts = buildLabelsSchema(dialect); - await db.batch(labelsStmts.map((s) => db.prepare(s))); + for (const stmt of labelsStmts) { + await runIdempotentDdl(db, stmt); + } } // FTS5 may not be available (e.g. node:sqlite) — skip gracefully @@ -350,14 +526,14 @@ export async function initSchema( // FTS5 not supported in this environment } } - await runMigrations(db); - - // Add count columns (ALTER TABLE — may already exist) - for (const stmt of buildCountColumns(config)) { - try { - await db.prepare(stmt).run(); - } catch { - // Column/index already exists — ignore - } - } + const hasSpaces = !!(config.spaces?.authority || config.spaces?.recordHost); + const hasFeeds = !!(config.feeds && Object.keys(config.feeds).length > 0); + // Spaces-targeted migrations route to spacesDb when one is configured; + // otherwise they hit the main db (which is where the spaces tables live + // when no separate spacesDb is supplied). + await runMigrations(db, spacesSharesMainDb ? undefined : spacesDb, hasSpaces, hasFeeds); + + // Idempotent count-column ALTERs + their indexes. Routed through + // `applyCountColumns` so non-duplicate-column errors propagate. + await applyCountColumns(db, config); } diff --git a/packages/contrail-appview/src/core/jetstream.ts b/packages/contrail-appview/src/core/jetstream.ts index abc0287..486d398 100644 --- a/packages/contrail-appview/src/core/jetstream.ts +++ b/packages/contrail-appview/src/core/jetstream.ts @@ -307,7 +307,7 @@ export async function runIngestCycle( const uniqueDids = [...new Set(events.map((e) => e.did))]; if (uniqueDids.length > 0) { try { - await refreshStaleIdentities(db, uniqueDids); + await refreshStaleIdentities(db, uniqueDids, config); } catch (err) { log.warn(`Identity refresh failed: ${err}`); } diff --git a/packages/contrail-appview/src/core/labels/resolve.ts b/packages/contrail-appview/src/core/labels/resolve.ts index 694bfec..af673d9 100644 --- a/packages/contrail-appview/src/core/labels/resolve.ts +++ b/packages/contrail-appview/src/core/labels/resolve.ts @@ -1,31 +1,37 @@ import { CompositeDidDocumentResolver, + type DidDocumentResolver, PlcDidDocumentResolver, WebDidDocumentResolver, } from "@atcute/identity-resolver"; import type { Did } from "@atcute/lexicons"; import type { Database } from "../types"; +import { validateExternalUrl } from "../client"; -/** Reject endpoint URLs that point to private/internal addresses or non-HTTPS. - * Mirrors the validator in core/client.ts — labeler endpoints should be - * publicly reachable for the same reasons PDS endpoints should. */ -function validateEndpointUrl(url: string): boolean { - try { - const parsed = new URL(url); - if (parsed.protocol !== "https:") return false; - const host = parsed.hostname; - if (host === "localhost" || host === "127.0.0.1" || host === "[::1]") return false; - if (host.startsWith("10.")) return false; - if (host.startsWith("192.168.")) return false; - if (host.startsWith("169.254.")) return false; - if (/^172\.(1[6-9]|2\d|3[01])\./.test(host)) return false; - return true; - } catch { - return false; - } +/** Optional network-override knobs accepted by labeler-endpoint resolution. + * Mirrors the `ContrailConfig.networkOverrides` shape — kept narrow here so + * callers can pass `config.networkOverrides` directly without re-shaping. + * Omitting the object preserves the previous public-internet behavior. */ +export interface LabelerResolveOverrides { + /** DID document resolver used when looking up the labeler service entry. + * When unset, falls back to a default composite (PLC + Web) pointing at the + * upstream PLC directory. Trusted; not SSRF-checked. + * Mirrors the resolver-injection pattern in `core/client.ts`. */ + resolver?: DidDocumentResolver; + /** Hostnames (DNS names or IP literals) to allow past the default SSRF + * guard when validating a resolved labeler endpoint. Match is exact, + * case-insensitive, port-agnostic. */ + additionalAllowedHosts?: string[]; } -const didResolver = new CompositeDidDocumentResolver({ +/** Reject endpoint URLs that point to private/internal addresses or non-HTTPS. + * Thin alias for the single shared SSRF guard {@link validateExternalUrl} in + * `contrail-base` — labeler endpoints are validated by the exact same rules as + * PDS endpoints, so the allowlist logic must live in one place. Kept exported + * under this name for existing callers/tests. */ +export const validateEndpointUrl = validateExternalUrl; + +const DEFAULT_DID_RESOLVER: DidDocumentResolver = new CompositeDidDocumentResolver({ methods: { plc: new PlcDidDocumentResolver(), web: new WebDidDocumentResolver(), @@ -33,16 +39,26 @@ const didResolver = new CompositeDidDocumentResolver({ }); /** Look up the labeler service endpoint from a DID. - * Reads the DID doc's `service[id="#atproto_labeler"].serviceEndpoint`. */ -export async function resolveLabelerEndpoint(did: string): Promise { + * Reads the DID doc's `service[id="#atproto_labeler"].serviceEndpoint`. + * + * `networkOverrides` (optional): customize the DID resolver used during the + * lookup, and/or which hostnames bypass the default SSRF guard. Omitting it + * preserves the original public-internet behavior. */ +export async function resolveLabelerEndpoint( + did: string, + networkOverrides?: LabelerResolveOverrides, +): Promise { if (!did.startsWith("did:plc:") && !did.startsWith("did:web:")) return null; + const resolver = networkOverrides?.resolver ?? DEFAULT_DID_RESOLVER; try { - const doc = await didResolver.resolve(did as Did<"plc"> | Did<"web">); + const doc = await resolver.resolve(did as Did<"plc"> | Did<"web">); const endpoint = doc.service ?.find((s) => s.id === "#atproto_labeler") ?.serviceEndpoint?.toString(); if (!endpoint) return null; - if (!validateEndpointUrl(endpoint)) return null; + if (!validateEndpointUrl(endpoint, networkOverrides?.additionalAllowedHosts ?? [])) { + return null; + } return endpoint; } catch { return null; @@ -63,11 +79,16 @@ const ENDPOINT_TTL_MS = 6 * 60 * 60 * 1000; // 6h, matches the recommended clien /** Get cached `(endpoint, cursor)` for a labeler. Resolves endpoint on * cache miss or staleness; persists endpoint + resolved_at back to the DB - * so subsequent ingest cycles avoid the network round-trip. */ + * so subsequent ingest cycles avoid the network round-trip. + * + * `networkOverrides` (optional): forwarded to `resolveLabelerEndpoint` for + * the cache-miss/stale path. Has no effect when `endpointOverride` is set + * or when a fresh cached endpoint is used. */ export async function getLabelerState( db: Database, did: string, endpointOverride: string | undefined, + networkOverrides?: LabelerResolveOverrides, ): Promise { const row = await db .prepare( @@ -81,7 +102,7 @@ export async function getLabelerState( !row?.resolved_at || Date.now() - row.resolved_at > ENDPOINT_TTL_MS; if (!endpoint || (!endpointOverride && stale)) { - endpoint = await resolveLabelerEndpoint(did); + endpoint = await resolveLabelerEndpoint(did, networkOverrides); if (!endpoint) return null; const now = Date.now(); await db diff --git a/packages/contrail-appview/src/core/labels/subscribe.ts b/packages/contrail-appview/src/core/labels/subscribe.ts index c80ed2b..1351d01 100644 --- a/packages/contrail-appview/src/core/labels/subscribe.ts +++ b/packages/contrail-appview/src/core/labels/subscribe.ts @@ -36,7 +36,15 @@ export async function runLabelIngestCycle( } const remaining = Math.max(2_000, deadline - Date.now()); try { - await pumpOneLabeler(db, source, log, remaining, /* persistent */ false); + await pumpOneLabeler( + db, + source, + log, + remaining, + /* persistent */ false, + {}, + config.networkOverrides, + ); } catch (err) { log.warn(`[labels] cycle for ${source.did} failed: ${err}`); } @@ -62,7 +70,7 @@ export async function runPersistentLabels( const signal = options.signal; const tasks = config.labels.sources.map((source) => - runOneLabelerForever(db, source, log, signal, options), + runOneLabelerForever(db, source, log, signal, options, config.networkOverrides), ); await Promise.all(tasks); } @@ -73,15 +81,24 @@ async function runOneLabelerForever( log: Logger, signal: AbortSignal | undefined, options: PersistentLabelsOptions, + networkOverrides: ContrailConfig["networkOverrides"], ): Promise { let attempts = 0; while (!signal?.aborted) { try { - await pumpOneLabeler(db, source, log, /* timeoutMs */ Infinity, true, { - signal, - batchSize: options.batchSize ?? DEFAULT_BATCH_SIZE, - flushIntervalMs: options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS, - }); + await pumpOneLabeler( + db, + source, + log, + /* timeoutMs */ Infinity, + true, + { + signal, + batchSize: options.batchSize ?? DEFAULT_BATCH_SIZE, + flushIntervalMs: options.flushIntervalMs ?? DEFAULT_FLUSH_INTERVAL_MS, + }, + networkOverrides, + ); attempts = 0; } catch (err) { if (signal?.aborted) break; @@ -114,8 +131,9 @@ async function pumpOneLabeler( timeoutMs: number, persistent: boolean, pumpOpts: PumpOptions = {}, + networkOverrides?: ContrailConfig["networkOverrides"], ): Promise { - const state = await getLabelerState(db, source.did, source.endpoint); + const state = await getLabelerState(db, source.did, source.endpoint, networkOverrides); if (!state) { log.warn(`[labels] could not resolve labeler endpoint for ${source.did}; skipping`); return; diff --git a/packages/contrail-appview/src/core/persistent.ts b/packages/contrail-appview/src/core/persistent.ts index 882c9b8..94ca666 100644 --- a/packages/contrail-appview/src/core/persistent.ts +++ b/packages/contrail-appview/src/core/persistent.ts @@ -157,7 +157,7 @@ async function streamAndFlush( const uniqueDids = [...new Set(batch.map((e) => e.did))]; if (uniqueDids.length > 0) { try { - await refreshStaleIdentities(db, uniqueDids); + await refreshStaleIdentities(db, uniqueDids, config); } catch (err) { log.warn(`Identity refresh failed: ${err}`); } diff --git a/packages/contrail-appview/src/core/refresh.ts b/packages/contrail-appview/src/core/refresh.ts index 7b2fd21..acbae5c 100644 --- a/packages/contrail-appview/src/core/refresh.ts +++ b/packages/contrail-appview/src/core/refresh.ts @@ -134,7 +134,7 @@ export async function refresh( let client: Client; try { client = await withTimeout( - () => getClient(did as Did, db), + () => getClient(did as Did, db, config), requestTimeout ); } catch { diff --git a/packages/contrail-appview/src/core/router/collection.ts b/packages/contrail-appview/src/core/router/collection.ts index 9482631..875a4b7 100644 --- a/packages/contrail-appview/src/core/router/collection.ts +++ b/packages/contrail-appview/src/core/router/collection.ts @@ -301,7 +301,7 @@ export async function runPipeline( let did: string | undefined; if (actor) { - const resolved = await resolveActor(db, actor); + const resolved = await resolveActor(db, actor, config); if (!resolved) throw new Error("Could not resolve actor"); did = resolved; // backfillUser expects the record NSID (for PDS calls), not the short name. diff --git a/packages/contrail-appview/src/core/router/feed.ts b/packages/contrail-appview/src/core/router/feed.ts index 0d6a21d..e817e40 100644 --- a/packages/contrail-appview/src/core/router/feed.ts +++ b/packages/contrail-appview/src/core/router/feed.ts @@ -243,7 +243,7 @@ export function registerFeedRoutes( return c.json({ error: "Unknown feed" }, 404); } - const did = await resolveActor(db, actor); + const did = await resolveActor(db, actor, config); if (!did) return c.json({ error: "Could not resolve actor" }, 400); await maybeBackfillFeed(c, db, config, did, feedName, feedConfig); diff --git a/packages/contrail-appview/src/core/router/index.ts b/packages/contrail-appview/src/core/router/index.ts index 10e4990..10291f4 100644 --- a/packages/contrail-appview/src/core/router/index.ts +++ b/packages/contrail-appview/src/core/router/index.ts @@ -88,7 +88,7 @@ export function createApp( const actor = c.req.query("actor"); if (!actor) return c.json({ error: "actor parameter required" }, 400); - const did = await resolveActor(db, actor); + const did = await resolveActor(db, actor, config); if (!did) return c.json({ error: "Could not resolve actor" }, 400); // Ensure profile records are backfilled @@ -138,7 +138,7 @@ export function createApp( : config.spaces?.authority ? { adapter: options.spaces?.adapter ?? new HostedAdapter(spacesDb, config), - verifier: buildVerifier(config.spaces.authority), + verifier: buildVerifier(config.spaces.authority, config.networkOverrides), manifestVerifier: config.spaces.authority.signing ? createManifestVerifier({ resolveKey: async (iss) => diff --git a/packages/contrail-appview/src/core/router/notify.ts b/packages/contrail-appview/src/core/router/notify.ts index 5ff630b..bedd74e 100644 --- a/packages/contrail-appview/src/core/router/notify.ts +++ b/packages/contrail-appview/src/core/router/notify.ts @@ -78,7 +78,7 @@ export async function processNotifyUris( ); for (const { uri, parsed } of validUris) { - const pds = await getPDS(parsed.did as Did, db); + const pds = await getPDS(parsed.did as Did, db, config); if (!pds) { errors.push(`could not resolve PDS for ${parsed.did}`); continue; diff --git a/packages/contrail-appview/src/core/router/profiles.ts b/packages/contrail-appview/src/core/router/profiles.ts index 5552357..8a95c82 100644 --- a/packages/contrail-appview/src/core/router/profiles.ts +++ b/packages/contrail-appview/src/core/router/profiles.ts @@ -85,7 +85,7 @@ export async function resolveProfiles( } // Resolve identities for all DIDs - const identities = await resolveIdentities(db, dids); + const identities = await resolveIdentities(db, dids, config); // Fetch missing profile records from PDS on demand const missingDids = dids.filter((d) => !result[d]); @@ -134,7 +134,7 @@ async function fetchMissingProfiles( const rkey = configRkey ?? "self"; const table = recordsTableName(shortName ?? collection); try { - const pds = await getPDS(did as Did, db); + const pds = await getPDS(did as Did, db, config); if (!pds) return; const url = new URL("/xrpc/com.atproto.repo.getRecord", pds); diff --git a/packages/contrail-appview/src/core/spaces/router.ts b/packages/contrail-appview/src/core/spaces/router.ts index a0557f5..9d4ce90 100644 --- a/packages/contrail-appview/src/core/spaces/router.ts +++ b/packages/contrail-appview/src/core/spaces/router.ts @@ -54,7 +54,7 @@ export function registerSpacesRoutes( if (!authorityConfig) return; const adapter = options.adapter ?? ctx?.adapter ?? new HostedAdapter(db, config); - const verifier = ctx?.verifier ?? buildVerifier(authorityConfig); + const verifier = ctx?.verifier ?? buildVerifier(authorityConfig, config.networkOverrides); const auth = options.authMiddleware ?? createServiceAuthMiddleware(verifier); const localRecordHost = spacesConfig.recordHost ? adapter : null; diff --git a/packages/contrail-appview/src/core/spaces/schema.ts b/packages/contrail-appview/src/core/spaces/schema.ts index ae33a7c..fcdf0cd 100644 --- a/packages/contrail-appview/src/core/spaces/schema.ts +++ b/packages/contrail-appview/src/core/spaces/schema.ts @@ -5,7 +5,7 @@ import { buildCollectionTables, buildDynamicIndexes, buildFtsTables, - buildCountColumns, + applyCountColumns, } from "../db/schema"; /** Spaces metadata tables — spaces, members, invites. No per-collection tables. */ @@ -76,8 +76,9 @@ export function buildSpacesBaseSchema(dialect: SqlDialect): string[] { /** Full spaces schema (base + per-collection tables + indexes). For callers * that need a single array of statements. Note: this does NOT include FTS - * virtual tables or ALTER TABLE count columns — those must be applied with - * try/catch fallbacks and are handled by `initSchema`. */ + * virtual tables or ALTER TABLE count columns — FTS is best-effort (engine + * may not be present) and count columns require dialect-aware idempotent + * ALTER. Both are handled by `initSchema` / `initSpacesSchema`. */ export function buildSpacesSchema(db: Database, config?: ContrailConfig): string[] { const dialect = getDialect(db); const base = buildSpacesBaseSchema(dialect); @@ -94,10 +95,10 @@ export async function initSpacesSchema(db: Database, config?: ContrailConfig): P const stmts = buildSpacesSchema(db, config); await db.batch(stmts.map((s) => db.prepare(s))); if (!config) return; + // FTS virtual tables: best-effort; the runtime may not have FTS5 compiled + // in (e.g. node:sqlite). Other DDL failures (count columns) propagate. for (const stmt of buildFtsTables(config, dialect, { forSpaces: true })) { - try { await db.prepare(stmt).run(); } catch { /* ignore */ } - } - for (const stmt of buildCountColumns(config, { forSpaces: true })) { - try { await db.prepare(stmt).run(); } catch { /* ignore */ } + try { await db.prepare(stmt).run(); } catch { /* FTS5 unavailable */ } } + await applyCountColumns(db, config, { forSpaces: true }); } diff --git a/packages/contrail-base/src/client.ts b/packages/contrail-base/src/client.ts index f37bea3..cd154ef 100644 --- a/packages/contrail-base/src/client.ts +++ b/packages/contrail-base/src/client.ts @@ -2,11 +2,12 @@ import { CompositeDidDocumentResolver, PlcDidDocumentResolver, WebDidDocumentResolver, + type DidDocumentResolver, } from "@atcute/identity-resolver"; import { type Did } from "@atcute/lexicons"; import { Client, simpleFetchHandler } from "@atcute/client"; import type {} from "@atcute/atproto"; -import type { Database } from "./types"; +import type { ContrailConfig, Database } from "./types"; // Slingshot-first PDS resolution with fallback to DID document resolution const SLINGSHOT_URL = @@ -18,28 +19,47 @@ export interface ResolvedIdentity { pds: string | null; } -/** Reject PDS URLs that point to private/internal addresses or non-HTTPS */ -function validatePdsUrl(url: string): boolean { +/** Reject external URLs (PDS, labeler, …) that point to private/internal + * addresses or non-HTTPS. The single SSRF guard shared across packages — + * callers MUST route every externally-resolved endpoint through this so the + * allowlist rules live in exactly one place. + * + * Hostnames in `additionalAllowedHosts` skip both checks. Match is exact, + * case-insensitive (allowlist entries are lowercased on compare; `URL.hostname` + * is already lowercased), and port-agnostic. + * + * Scope: best-effort guard against the obvious internal-address classes + * (private/link-local IPv4 literals, localhost, non-HTTPS). It does NOT + * resolve DNS, so a public hostname that resolves to a private address is not + * caught here, and IPv6 / non-canonical IP encodings are only partially + * covered. Defense-in-depth (egress network policy) is expected when resolver + * inputs are fully untrusted. */ +export function validateExternalUrl(url: string, additionalAllowedHosts?: string[]): boolean { + let parsed: URL; try { - const parsed = new URL(url); - if (parsed.protocol !== "https:") return false; - const host = parsed.hostname; - // Block private/internal IP ranges - if (host === "localhost" || host === "127.0.0.1" || host === "[::1]") return false; - if (host.startsWith("10.")) return false; - if (host.startsWith("192.168.")) return false; - if (host.startsWith("169.254.")) return false; - if (/^172\.(1[6-9]|2\d|3[01])\./.test(host)) return false; - return true; + parsed = new URL(url); } catch { return false; } + if (additionalAllowedHosts?.some((h) => h.toLowerCase() === parsed.hostname)) { + return true; + } + if (parsed.protocol !== "https:") return false; + const host = parsed.hostname; + // Block private/internal IP ranges + if (host === "localhost" || host === "127.0.0.1" || host === "[::1]") return false; + if (host.startsWith("10.")) return false; + if (host.startsWith("192.168.")) return false; + if (host.startsWith("169.254.")) return false; + if (/^172\.(1[6-9]|2\d|3[01])\./.test(host)) return false; + return true; } async function resolveViaSlingshot( - identifier: string + identifier: string, + slingshotUrl: string, ): Promise { - const url = new URL(SLINGSHOT_URL); + const url = new URL(slingshotUrl); url.searchParams.set("identifier", identifier); try { @@ -61,15 +81,19 @@ async function resolveViaSlingshot( } } -const didResolver = new CompositeDidDocumentResolver({ +const DEFAULT_DID_RESOLVER: DidDocumentResolver = new CompositeDidDocumentResolver({ methods: { plc: new PlcDidDocumentResolver(), web: new WebDidDocumentResolver(), }, }); -async function getPDSViaDidDoc(did: Did): Promise { - const doc = await didResolver.resolve(did as Did<"plc"> | Did<"web">); +async function getPDSViaDidDoc( + did: Did, + config?: ContrailConfig, +): Promise { + const resolver = config?.networkOverrides?.resolver ?? DEFAULT_DID_RESOLVER; + const doc = await resolver.resolve(did as Did<"plc"> | Did<"web">); return doc.service ?.find((s) => s.id === "#atproto_pds") ?.serviceEndpoint.toString(); @@ -78,21 +102,28 @@ async function getPDSViaDidDoc(did: Did): Promise { /** * Resolve identity info (did, handle, pds) for a DID or handle. * Uses slingshot first, falls back to DID doc for PDS. + * + * `config?.networkOverrides` (optional): customize the slingshot endpoint, + * the PLC URL used during DID-doc fallback, and/or which hostnames bypass + * the default SSRF guard. Omitting `config` preserves all defaults. */ export async function resolvePDS( - identifier: string + identifier: string, + config?: ContrailConfig, ): Promise { - const result = await resolveViaSlingshot(identifier); + const slingshotUrl = config?.networkOverrides?.slingshotUrl ?? SLINGSHOT_URL; + const allowed = config?.networkOverrides?.additionalAllowedHosts; + const result = await resolveViaSlingshot(identifier, slingshotUrl); if (result?.pds) { - if (!validatePdsUrl(result.pds)) return { ...result, pds: null }; + if (!validateExternalUrl(result.pds, allowed)) return { ...result, pds: null }; return result; } // Fall back to DID doc resolution (only works for DIDs, not handles) if (identifier.startsWith("did:")) { try { - const pds = await getPDSViaDidDoc(identifier as Did); - if (pds && validatePdsUrl(pds)) { + const pds = await getPDSViaDidDoc(identifier as Did, config); + if (pds && validateExternalUrl(pds, allowed)) { return { did: identifier, handle: result?.handle ?? null, @@ -107,7 +138,14 @@ export async function resolvePDS( return result; } -// In-memory PDS cache with TTL + size limit, plus in-flight deduplication +// In-memory PDS cache with TTL + size limit, plus in-flight deduplication. +// +// Keyed by DID only — this assumes a single, process-wide `networkOverrides` +// config (the deployment model: one resolver + one SSRF allowlist per process). +// Every caller in this monorepo now threads the same in-scope `config`, so a +// config-less and an override-aware resolution can never race for the same DID. +// If a future deployment ever resolves the same DID under differing overrides +// in one process, key these caches by an override fingerprint instead. const PDS_CACHE_TTL = 60 * 60 * 1000; // 1 hour const PDS_CACHE_MAX = 10_000; const pdsCache = new Map(); @@ -134,7 +172,8 @@ function pdsCacheSet(did: string, pds: string): void { export async function getPDS( did: Did, - db?: Database + db?: Database, + config?: ContrailConfig, ): Promise { const mem = pdsCacheGet(did); if (mem) return mem; @@ -143,7 +182,7 @@ export async function getPDS( const inflight = pdsInflight.get(did); if (inflight) return inflight; - const promise = resolvePDSCached(did, db); + const promise = resolvePDSCached(did, db, config); pdsInflight.set(did, promise); try { return await promise; @@ -154,7 +193,8 @@ export async function getPDS( async function resolvePDSCached( did: Did, - db?: Database + db?: Database, + config?: ContrailConfig, ): Promise { if (db) { const cached = await db @@ -167,7 +207,7 @@ async function resolvePDSCached( } } - const resolved = await resolvePDS(did); + const resolved = await resolvePDS(did, config); if (!resolved?.pds) return undefined; pdsCacheSet(did, resolved.pds); @@ -185,10 +225,21 @@ async function resolvePDSCached( return resolved.pds; } -export async function getClient(did: Did, db?: Database): Promise { - const pds = await getPDS(did, db); +export async function getClient( + did: Did, + db?: Database, + config?: ContrailConfig, +): Promise { + const pds = await getPDS(did, db, config); if (!pds) throw new Error(`PDS not found for ${did}`); return new Client({ handler: simpleFetchHandler({ service: pds }), }); } + +/** Test-only: clear module-level PDS caches. Production code MUST NOT call this. + * Exported with a `__` prefix to signal it is not part of the public API. */ +export function __resetPdsCachesForTests(): void { + pdsCache.clear(); + pdsInflight.clear(); +} diff --git a/packages/contrail-base/src/identity.ts b/packages/contrail-base/src/identity.ts index fe207b3..26e84fd 100644 --- a/packages/contrail-base/src/identity.ts +++ b/packages/contrail-base/src/identity.ts @@ -1,5 +1,5 @@ import type { Did } from "@atcute/lexicons"; -import type { Database, Logger } from "./types"; +import type { ContrailConfig, Database, Logger } from "./types"; import { isDid, isHandle } from "@atcute/lexicons/syntax"; import { resolvePDS } from "./client"; @@ -28,9 +28,10 @@ function isStale(resolvedAt: number): boolean { async function fetchAndSave( db: Database, identifier: string, - cached?: Identity | null + cached?: Identity | null, + config?: ContrailConfig, ): Promise { - const resolved = await resolvePDS(identifier); + const resolved = await resolvePDS(identifier, config); const identity: Identity = { did: resolved?.did ?? identifier, handle: resolved?.handle ?? cached?.handle ?? null, @@ -43,7 +44,8 @@ async function fetchAndSave( export async function resolveIdentity( db: Database, - did: Did + did: Did, + config?: ContrailConfig, ): Promise { const cached = await db .prepare("SELECT did, handle, pds, resolved_at FROM identities WHERE did = ?") @@ -52,12 +54,13 @@ export async function resolveIdentity( if (cached && !isStale(cached.resolved_at)) return cached; - return fetchAndSave(db, did, cached); + return fetchAndSave(db, did, cached, config); } export async function resolveIdentities( db: Database, - dids: string[] + dids: string[], + config?: ContrailConfig, ): Promise> { const map = new Map(); if (dids.length === 0) return map; @@ -80,7 +83,7 @@ export async function resolveIdentities( for (const did of dids) { if (map.has(did) || !isDid(did)) continue; try { - const identity = await fetchAndSave(db, did); + const identity = await fetchAndSave(db, did, undefined, config); map.set(did, identity); } catch { // Silently skip unresolvable identities @@ -92,7 +95,8 @@ export async function resolveIdentities( export async function resolveActor( db: Database, - actor: string + actor: string, + config?: ContrailConfig, ): Promise { if (isDid(actor)) return actor; if (!isHandle(actor)) return null; @@ -106,7 +110,7 @@ export async function resolveActor( if (cached && !isStale(cached.resolved_at)) return cached.did; // Resolve via slingshot - const resolved = await resolvePDS(actor); + const resolved = await resolvePDS(actor, config); if (!resolved?.did || !isDid(resolved.did)) return null; await saveIdentity(db, { @@ -139,7 +143,8 @@ export async function applyIdentityEvent( export async function refreshStaleIdentities( db: Database, - dids: string[] + dids: string[], + config?: ContrailConfig, ): Promise { if (dids.length === 0) return; @@ -169,7 +174,7 @@ export async function refreshStaleIdentities( for (const did of toRefresh) { try { - await fetchAndSave(db, did); + await fetchAndSave(db, did, undefined, config); } catch { // Silently skip unresolvable identities } diff --git a/packages/contrail-base/src/spaces/auth.ts b/packages/contrail-base/src/spaces/auth.ts index 4c1e817..4c255f6 100644 --- a/packages/contrail-base/src/spaces/auth.ts +++ b/packages/contrail-base/src/spaces/auth.ts @@ -12,12 +12,22 @@ import { readInProcess } from "./in-process"; export { ServiceJwtVerifier }; -/** Build a ServiceJwtVerifier from an AuthorityConfig, using the configured - * resolver or a default PLC+Web composite. The verifier checks that incoming - * JWTs target this authority's serviceDid (aud claim). */ -export function buildVerifier(authority: AuthorityConfig): ServiceJwtVerifier { +/** Build a ServiceJwtVerifier from an AuthorityConfig, using (in precedence + * order) the authority-specific resolver, then the deployment-wide + * `networkOverrides.resolver`, then a default PLC+Web composite. The verifier + * checks that incoming JWTs target this authority's serviceDid (aud claim). + * + * `networkOverrides` is optional and is the same shape carried on + * `ContrailConfig.networkOverrides` — callers with a ContrailConfig in scope + * should pass `config.networkOverrides` so private-network deployments share + * one resolver across both identity resolution and service-auth verification. */ +export function buildVerifier( + authority: AuthorityConfig, + networkOverrides?: { resolver?: DidDocumentResolver }, +): ServiceJwtVerifier { const resolver = authority.resolver ?? + networkOverrides?.resolver ?? new CompositeDidDocumentResolver({ methods: { plc: new PlcDidDocumentResolver(), diff --git a/packages/contrail-base/src/types.ts b/packages/contrail-base/src/types.ts index 9d63336..8a15a4a 100644 --- a/packages/contrail-base/src/types.ts +++ b/packages/contrail-base/src/types.ts @@ -241,6 +241,37 @@ export interface ContrailConfig { * ingests synthesized rows for any follower already in our identities * table. Lets newcomers immediately appear in existing users' feeds. */ constellation?: ConstellationConfig | false; + /** Network overrides for private-network or test deployments. + * All subfields default to current public-internet behavior; + * omitting `networkOverrides` entirely preserves current behavior. + * + * SECURITY: `resolver` and `slingshotUrl` are taken at face value and are + * NOT validated against the SSRF guard — the consumer is trusted to + * configure them. Only the PDS URL returned downstream is validated, + * and only `additionalAllowedHosts` widens that PDS validator. There is + * no "disable SSRF" flag. */ + networkOverrides?: { + /** DID document resolver used during the DID-doc PDS fallback. When + * unset, contrail constructs a default `CompositeDidDocumentResolver` + * with PLC + Web methods pointing at the upstream PLC directory. + * Pass a custom resolver to point at a private PLC mirror, inject a + * custom fetch (mTLS, retry, instrumentation), or swap in an + * alternative DID method composition. + * Mirrors the `AuthorityConfig.resolver` pattern in `spaces/types.ts`. */ + resolver?: import("@atcute/identity-resolver").DidDocumentResolver; + /** Slingshot identity resolver URL override. Trusted; not SSRF-checked. + * Default: https://slingshot.microcosm.blue/xrpc/com.bad-example.identity.resolveMiniDoc */ + slingshotUrl?: string; + /** Hostnames (DNS names or IP literals) to allow past the default SSRF + * guard when validating a resolved PDS URL. + * For listed hostnames, the non-HTTPS + private-CIDR checks are skipped. + * For all other hostnames, the default validator runs unchanged. + * Match semantics: exact hostname, case-insensitive (entries are + * lowercased on comparison; `URL.hostname` is already lowercased), + * port-agnostic. + * Example: ["pds.dev.svc.cluster.local"]. */ + additionalAllowedHosts?: string[]; + }; } export interface ConstellationConfig { diff --git a/packages/contrail-community/src/integration.ts b/packages/contrail-community/src/integration.ts index 9dca172..38b9435 100644 --- a/packages/contrail-community/src/integration.ts +++ b/packages/contrail-community/src/integration.ts @@ -60,7 +60,7 @@ export function createCommunityIntegration( registerRoutes(app, opts) { // Reuse the spaces JWT verifier — the auth model is identical. if (!config.spaces?.authority) return; - const verifier = buildVerifier(config.spaces.authority); + const verifier = buildVerifier(config.spaces.authority, config.networkOverrides); const authMiddleware = opts?.authMiddleware ?? createServiceAuthMiddleware(verifier); registerCommunityRoutes( diff --git a/packages/contrail/tests/client.test.ts b/packages/contrail/tests/client.test.ts new file mode 100644 index 0000000..7aa2143 --- /dev/null +++ b/packages/contrail/tests/client.test.ts @@ -0,0 +1,239 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { resolvePDS, getClient, getPDS, __resetPdsCachesForTests } from "../src/core/client"; +import { type DidDocumentResolver } from "@atcute/identity-resolver"; +import { createTestDbWithSchema } from "./helpers"; +import type { Did } from "@atcute/lexicons"; + +describe("validatePdsUrl via resolvePDS — regression baseline", () => { + let fetchSpy: ReturnType; + + beforeEach(() => { + fetchSpy = vi.spyOn(global, "fetch"); + }); + + afterEach(() => { + fetchSpy.mockRestore(); + }); + + it("accepts a public HTTPS PDS", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:a", handle: "a.bsky.social", pds: "https://shimeji.us-east.host.bsky.network" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:a"); + expect(r?.pds).toBe("https://shimeji.us-east.host.bsky.network"); + }); + + it("rejects non-HTTPS PDS (returns pds: null)", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:b", handle: "b.test", pds: "http://malicious.example.com" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:b"); + expect(r?.pds).toBe(null); + }); + + it("rejects 10.x private CIDR", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:c", pds: "https://10.0.0.1" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:c"); + expect(r?.pds).toBe(null); + }); + + it("rejects 192.168.x private CIDR", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:d", pds: "https://192.168.1.1" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:d"); + expect(r?.pds).toBe(null); + }); + + it("rejects 172.16-31.x private CIDR", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:e", pds: "https://172.20.0.5" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:e"); + expect(r?.pds).toBe(null); + }); + + it("rejects localhost and 169.254 link-local", async () => { + fetchSpy.mockResolvedValueOnce( + new Response(JSON.stringify({ did: "did:plc:f", pds: "https://localhost" }), { status: 200 }) + ); + expect((await resolvePDS("did:plc:f"))?.pds).toBe(null); + + fetchSpy.mockResolvedValueOnce( + new Response(JSON.stringify({ did: "did:plc:g", pds: "https://169.254.169.254" }), { status: 200 }) + ); + expect((await resolvePDS("did:plc:g"))?.pds).toBe(null); + }); +}); + +describe("validatePdsUrl via resolvePDS — additionalAllowedHosts allowlist", () => { + let fetchSpy: ReturnType; + + beforeEach(() => { + fetchSpy = vi.spyOn(global, "fetch"); + }); + + afterEach(() => { + fetchSpy.mockRestore(); + }); + + it("accepts http://pds.dev.svc.cluster.local when host is on allowlist", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:h", pds: "http://pds.dev.svc.cluster.local" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:h", { + namespace: "test", + collections: {}, + networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] }, + }); + expect(r?.pds).toBe("http://pds.dev.svc.cluster.local"); + }); + + it("still rejects http://other.private.host when not on allowlist", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:i", pds: "http://other.private.host" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:i", { + namespace: "test", + collections: {}, + networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] }, + }); + expect(r?.pds).toBe(null); + }); + + it("still rejects http://192.168.1.1 when not on allowlist", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:j", pds: "http://192.168.1.1" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:j", { + namespace: "test", + collections: {}, + networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] }, + }); + expect(r?.pds).toBe(null); + }); + + it("port-agnostic: allowlist matches hostname regardless of port", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:k", pds: "http://pds.dev.svc.cluster.local:8080" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:k", { + namespace: "test", + collections: {}, + networkOverrides: { additionalAllowedHosts: ["pds.dev.svc.cluster.local"] }, + }); + expect(r?.pds).toBe("http://pds.dev.svc.cluster.local:8080"); + }); + + it("case-insensitive: mixed-case allowlist entries match lowercase URL.hostname", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:l", pds: "http://pds.dev.svc.cluster.local" }), { status: 200 }) + ); + const r = await resolvePDS("did:plc:l", { + namespace: "test", + collections: {}, + networkOverrides: { additionalAllowedHosts: ["PDS.Dev.Svc.Cluster.Local"] }, + }); + expect(r?.pds).toBe("http://pds.dev.svc.cluster.local"); + }); +}); + +describe("getPDSViaDidDoc — resolver override", () => { + let fetchSpy: ReturnType; + + beforeEach(() => { + __resetPdsCachesForTests(); + fetchSpy = vi.spyOn(global, "fetch"); + }); + + afterEach(() => { + fetchSpy.mockRestore(); + }); + + it("falls back to DID doc when slingshot returns no pds, and uses injected resolver", async () => { + fetchSpy.mockResolvedValueOnce( + new Response(JSON.stringify({ did: "did:plc:abc", handle: "alice.test" }), { status: 200 }) + ); + + const resolveCalls: string[] = []; + const injectedResolver: DidDocumentResolver = { + async resolve(did: string) { + resolveCalls.push(did); + return { + service: [ + { id: "#atproto_pds", type: "AtprotoPersonalDataServer", serviceEndpoint: "http://pds.dev.svc.cluster.local" }, + ], + } as any; + }, + } as any; + + const r = await resolvePDS("did:plc:abc", { + namespace: "test", + collections: {}, + networkOverrides: { + resolver: injectedResolver, + additionalAllowedHosts: ["pds.dev.svc.cluster.local"], + }, + }); + expect(r?.pds).toBe("http://pds.dev.svc.cluster.local"); + expect(resolveCalls).toContain("did:plc:abc"); + }); +}); + +describe("getClient + getPDS — config plumb-through", () => { + let fetchSpy: ReturnType; + + beforeEach(() => { + __resetPdsCachesForTests(); + fetchSpy = vi.spyOn(global, "fetch"); + }); + + afterEach(() => { + fetchSpy.mockRestore(); + }); + + it("getClient with config?.networkOverrides.slingshotUrl uses the override", async () => { + fetchSpy.mockImplementation(async (input) => { + const url = String(input); + if (url.includes("my-slingshot")) { + return new Response( + JSON.stringify({ did: "did:plc:x", pds: "https://pds.allowed.test" }), + { status: 200 }, + ); + } + throw new Error(`Unexpected fetch: ${url}`); + }); + const db = await createTestDbWithSchema(); + const client = await getClient("did:plc:x" as Did, db, { + namespace: "test", + collections: {}, + networkOverrides: { + slingshotUrl: "https://my-slingshot.test/xrpc/com.bad-example.identity.resolveMiniDoc", + additionalAllowedHosts: ["pds.allowed.test"], + }, + }); + expect(client).toBeDefined(); + const calls = fetchSpy.mock.calls.map((c) => String(c[0])); + expect(calls.some((u) => u.includes("my-slingshot.test"))).toBe(true); + }); + + it("getPDS with no config uses the default slingshot URL (backward-compat)", async () => { + fetchSpy.mockImplementation(async (input) => { + const url = String(input); + if (url.includes("slingshot.microcosm.blue")) { + return new Response( + JSON.stringify({ did: "did:plc:y", pds: "https://shimeji.us-east.host.bsky.network" }), + { status: 200 }, + ); + } + throw new Error(`Unexpected fetch: ${url}`); + }); + const db = await createTestDbWithSchema(); + const pds = await getPDS("did:plc:y" as Did, db); + expect(pds).toBe("https://shimeji.us-east.host.bsky.network"); + const calls = fetchSpy.mock.calls.map((c) => String(c[0])); + expect(calls.some((u) => u.includes("slingshot.microcosm.blue"))).toBe(true); + }); +}); diff --git a/packages/contrail/tests/identity-config.test.ts b/packages/contrail/tests/identity-config.test.ts new file mode 100644 index 0000000..aa5df42 --- /dev/null +++ b/packages/contrail/tests/identity-config.test.ts @@ -0,0 +1,87 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { + resolveIdentity, + resolveIdentities, + resolveActor, + refreshStaleIdentities, +} from "../src/core/identity"; +import { __resetPdsCachesForTests } from "../src/core/client"; +import { createTestDbWithSchema } from "./helpers"; +import type { Did } from "@atcute/lexicons"; + +describe("identity.ts — config plumb-through", () => { + let fetchSpy: ReturnType; + + beforeEach(() => { + __resetPdsCachesForTests(); + fetchSpy = vi.spyOn(global, "fetch"); + }); + + afterEach(() => { + fetchSpy.mockRestore(); + }); + + const overrideConfig = { + namespace: "test", + collections: {}, + networkOverrides: { + slingshotUrl: "https://my-slingshot.test/xrpc/com.bad-example.identity.resolveMiniDoc", + additionalAllowedHosts: ["pds.allowed.test"], + }, + }; + + it("resolveIdentity routes slingshot to override URL", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:a", handle: "a.test", pds: "https://pds.allowed.test" }), { status: 200 }), + ); + const db = await createTestDbWithSchema(); + const id = await resolveIdentity(db, "did:plc:a" as Did, overrideConfig); + expect(id.pds).toBe("https://pds.allowed.test"); + expect(fetchSpy.mock.calls.some(([u]) => String(u).includes("my-slingshot.test"))).toBe(true); + }); + + it("resolveIdentities batch routes to override URL", async () => { + fetchSpy.mockImplementation(async (input) => { + const url = String(input); + const id = new URL(url).searchParams.get("identifier") ?? "did:plc:?"; + return new Response( + JSON.stringify({ did: id, handle: `${id}.test`, pds: "https://pds.allowed.test" }), + { status: 200 }, + ); + }); + const db = await createTestDbWithSchema(); + const m = await resolveIdentities(db, ["did:plc:b", "did:plc:c"], overrideConfig); + expect(m.get("did:plc:b")?.pds).toBe("https://pds.allowed.test"); + expect(m.get("did:plc:c")?.pds).toBe("https://pds.allowed.test"); + }); + + it("resolveActor routes a handle lookup to override URL", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:d", handle: "user.test", pds: "https://pds.allowed.test" }), { status: 200 }), + ); + const db = await createTestDbWithSchema(); + const did = await resolveActor(db, "user.test", overrideConfig); + expect(did).toBe("did:plc:d"); + expect(fetchSpy.mock.calls.some(([u]) => String(u).includes("my-slingshot.test"))).toBe(true); + }); + + it("refreshStaleIdentities routes to override URL", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:e", handle: "e.test", pds: "https://pds.allowed.test" }), { status: 200 }), + ); + const db = await createTestDbWithSchema(); + await refreshStaleIdentities(db, ["did:plc:e"], overrideConfig); + const row = await db.prepare("SELECT did, pds FROM identities WHERE did = ?").bind("did:plc:e").first<{ pds: string }>(); + expect(row?.pds).toBe("https://pds.allowed.test"); + }); + + it("backward-compat: no config preserves default slingshot URL", async () => { + fetchSpy.mockResolvedValue( + new Response(JSON.stringify({ did: "did:plc:f", handle: "f.bsky.social", pds: "https://shimeji.us-east.host.bsky.network" }), { status: 200 }), + ); + const db = await createTestDbWithSchema(); + const id = await resolveIdentity(db, "did:plc:f" as Did); + expect(id.pds).toBe("https://shimeji.us-east.host.bsky.network"); + expect(fetchSpy.mock.calls.some(([u]) => String(u).includes("slingshot.microcosm.blue"))).toBe(true); + }); +}); diff --git a/packages/contrail/tests/labels-resolve.test.ts b/packages/contrail/tests/labels-resolve.test.ts new file mode 100644 index 0000000..80ca73d --- /dev/null +++ b/packages/contrail/tests/labels-resolve.test.ts @@ -0,0 +1,81 @@ +import { describe, it, expect, vi } from "vitest"; +import { + CompositeDidDocumentResolver, + PlcDidDocumentResolver, + WebDidDocumentResolver, + type DidDocumentResolver, +} from "@atcute/identity-resolver"; +import { + resolveLabelerEndpoint, + validateEndpointUrl, +} from "../src/core/labels/resolve"; + +describe("validateEndpointUrl additionalAllowedHosts", () => { + it("rejects pds.dev.svc.cluster.local without override (HTTP + private hostname)", () => { + expect(validateEndpointUrl("http://pds.dev.svc.cluster.local:2583")).toBe(false); + }); + + it("accepts pds.dev.svc.cluster.local when listed in additionalAllowedHosts (case-insensitive)", () => { + expect( + validateEndpointUrl("http://PDS.dev.svc.cluster.local:2583", [ + "pds.dev.svc.cluster.local", + ]), + ).toBe(true); + }); + + it("does not relax HTTPS requirement for non-listed hosts when override is present", () => { + expect( + validateEndpointUrl("http://attacker.com", ["pds.dev.svc.cluster.local"]), + ).toBe(false); + }); + + it("ignores port differences (host-only match)", () => { + expect( + validateEndpointUrl("http://pds.dev.svc.cluster.local:9999", [ + "pds.dev.svc.cluster.local", + ]), + ).toBe(true); + }); +}); + +describe("resolveLabelerEndpoint resolver injection", () => { + it("uses networkOverrides.resolver when provided", async () => { + const mockResolver = { + resolve: vi.fn().mockResolvedValue({ + service: [ + { id: "#atproto_labeler", serviceEndpoint: "https://labeler.test" }, + ], + }), + }; + const endpoint = await resolveLabelerEndpoint("did:plc:abc123", { + resolver: mockResolver as unknown as DidDocumentResolver, + }); + expect(endpoint).toBe("https://labeler.test"); + expect(mockResolver.resolve).toHaveBeenCalledOnce(); + }); + + it("applies additionalAllowedHosts to resolved labeler endpoint", async () => { + const mockResolver = { + resolve: vi.fn().mockResolvedValue({ + service: [ + { + id: "#atproto_labeler", + serviceEndpoint: "http://labeler.dev.svc.cluster.local:2583", + }, + ], + }), + }; + // Without override the http endpoint should be rejected + const rejected = await resolveLabelerEndpoint("did:plc:abc123", { + resolver: mockResolver as unknown as DidDocumentResolver, + }); + expect(rejected).toBeNull(); + + // With override it should pass + const accepted = await resolveLabelerEndpoint("did:plc:abc123", { + resolver: mockResolver as unknown as DidDocumentResolver, + additionalAllowedHosts: ["labeler.dev.svc.cluster.local"], + }); + expect(accepted).toBe("http://labeler.dev.svc.cluster.local:2583"); + }); +}); diff --git a/packages/contrail/tests/network-overrides-appview.test.ts b/packages/contrail/tests/network-overrides-appview.test.ts new file mode 100644 index 0000000..1110a57 --- /dev/null +++ b/packages/contrail/tests/network-overrides-appview.test.ts @@ -0,0 +1,96 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { createApp } from "../src/core/router"; +import { runIngestCycle } from "../src/core/jetstream"; +import { __resetPdsCachesForTests } from "../src/core/client"; +import { createTestDbWithSchema, TEST_CONFIG } from "./helpers"; +import type { ContrailConfig } from "../src/core/types"; + +// Mock the Jetstream subscription so `runIngestCycle` ingests one synthetic +// commit without opening a real WebSocket. Everything else in the appview +// ingest path runs for real, so this exercises the live-ingest refresh path +// end-to-end (the smoking-gun call site `refreshStaleIdentities(db, dids)`). +vi.mock("@atcute/jetstream", async (importOriginal) => { + const actual = await importOriginal(); + class MockJetstreamSubscription { + cursor: number | null = null; + constructor(_opts: unknown) {} + async *[Symbol.asyncIterator]() { + yield { + kind: "commit", + // A past time_us so the ingest loop doesn't treat it as "caught up". + time_us: 1_000_000, + did: "did:plc:ingest", + commit: { + collection: "community.lexicon.calendar.event", + operation: "create", + rkey: "abc", + cid: "bafyabc", + record: { name: "Test Event", startsAt: "2026-04-01T10:00:00Z", mode: "online" }, + }, + }; + } + } + return { ...actual, JetstreamSubscription: MockJetstreamSubscription }; +}); + +const silentLogger = { log() {}, warn() {}, error() {} }; + +const OVERRIDE = { + slingshotUrl: "https://my-slingshot.test/xrpc/com.bad-example.identity.resolveMiniDoc", + additionalAllowedHosts: ["pds.allowed.test"], +}; + +function overrideConfig(): ContrailConfig { + return { ...TEST_CONFIG, logger: silentLogger, networkOverrides: OVERRIDE }; +} + +describe("networkOverrides — appview entry points thread config", () => { + let fetchSpy: ReturnType; + + beforeEach(() => { + __resetPdsCachesForTests(); + fetchSpy = vi.spyOn(globalThis, "fetch").mockResolvedValue( + new Response( + JSON.stringify({ did: "did:plc:resolved", handle: "user.test", pds: "https://pds.allowed.test" }), + { status: 200, headers: { "content-type": "application/json" } }, + ), + ); + }); + + afterEach(() => { + fetchSpy.mockRestore(); + }); + + // router actor-resolution path: GET /xrpc/.getProfile -> resolveActor + it("router getProfile resolves a handle via the override slingshot", async () => { + const db = await createTestDbWithSchema(); + const app = createApp(db, overrideConfig()); + + await app.fetch( + new Request(`http://localhost/xrpc/${TEST_CONFIG.namespace}.getProfile?actor=user.test`), + ); + + // The handle lookup (identifier=user.test) must go to the override + // slingshot, not the default public one. Before the fix `resolveActor` + // was called without `config`, so this fetch hit the default URL. + const hitOverrideForHandle = fetchSpy.mock.calls.some(([u]) => { + const s = String(u); + return s.includes("my-slingshot.test") && s.includes("identifier=user.test"); + }); + expect(hitOverrideForHandle).toBe(true); + }); + + // live-ingest refresh path: runIngestCycle -> refreshStaleIdentities + it("jetstream ingest cycle refreshes identities via the override slingshot", async () => { + const db = await createTestDbWithSchema(); + + await runIngestCycle(db, overrideConfig(), 1_000); + + // refreshStaleIdentities resolved the ingested DID; before the fix it was + // called without `config`, so the resolve hit the default slingshot. + const hitOverride = fetchSpy.mock.calls.some(([u]) => + String(u).includes("my-slingshot.test"), + ); + expect(hitOverride).toBe(true); + }); +}); diff --git a/packages/contrail/tests/network-overrides.test.ts b/packages/contrail/tests/network-overrides.test.ts new file mode 100644 index 0000000..011ab76 --- /dev/null +++ b/packages/contrail/tests/network-overrides.test.ts @@ -0,0 +1,189 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from "vitest"; +import http from "node:http"; +import type { AddressInfo } from "node:net"; +import { resolvePDS, getClient, __resetPdsCachesForTests } from "../src/core/client"; +import { refreshStaleIdentities } from "../src/core/identity"; +import { createTestDbWithSchema } from "./helpers"; +import type { ContrailConfig } from "../src/core/types"; +import type { Did } from "@atcute/lexicons"; +import { + CompositeDidDocumentResolver, + PlcDidDocumentResolver, + WebDidDocumentResolver, +} from "@atcute/identity-resolver"; + +type Hit = { method: string; url: string }; + +interface Stub { + url: string; + hits: Hit[]; + setHandler: (h: (req: http.IncomingMessage, res: http.ServerResponse) => void) => void; + close: () => Promise; +} + +async function startStub(): Promise { + const hits: Hit[] = []; + let handler: (req: http.IncomingMessage, res: http.ServerResponse) => void = (_req, res) => { + res.writeHead(404); + res.end(); + }; + const server = http.createServer((req, res) => { + hits.push({ method: req.method ?? "?", url: req.url ?? "" }); + handler(req, res); + }); + await new Promise((r) => server.listen(0, "127.0.0.1", r)); + const { port } = server.address() as AddressInfo; + return { + url: `http://127.0.0.1:${port}`, + hits, + setHandler: (h) => { + handler = h; + }, + close: () => new Promise((r) => server.close(() => r())), + }; +} + +let plc: Stub; +let slingshot: Stub; + +const baseConfig: ContrailConfig = { + namespace: "test", + collections: {}, +}; + +beforeAll(async () => { + plc = await startStub(); + slingshot = await startStub(); +}); + +afterAll(async () => { + await plc.close(); + await slingshot.close(); +}); + +beforeEach(() => { + plc.hits.length = 0; + slingshot.hits.length = 0; + __resetPdsCachesForTests(); +}); + +describe("networkOverrides — full chain integration", () => { + it("resolvePDS routes slingshot fetch to slingshotUrl override", async () => { + slingshot.setHandler((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ did: "did:plc:abc", handle: "alice.test", pds: "http://pds.private.test" })); + }); + + const config: ContrailConfig = { + ...baseConfig, + networkOverrides: { + slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`, + additionalAllowedHosts: ["pds.private.test"], + }, + }; + const result = await resolvePDS("did:plc:abc", config); + expect(result?.pds).toBe("http://pds.private.test"); + expect(slingshot.hits.length).toBeGreaterThan(0); + expect(slingshot.hits[0].url).toContain("identifier=did%3Aplc%3Aabc"); + }); + + it("rejects pds when not in allowlist (default validator still applies)", async () => { + slingshot.setHandler((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ did: "did:plc:bcd", pds: "http://pds.private.test" })); + }); + + const config: ContrailConfig = { + ...baseConfig, + networkOverrides: { + slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`, + }, + }; + const result = await resolvePDS("did:plc:bcd", config); + expect(result?.pds).toBe(null); + }); + + it("falls back to plcUrl when slingshot returns no pds", async () => { + slingshot.setHandler((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ did: "did:plc:cde", handle: "carol.test" })); + }); + plc.setHandler((req, res) => { + const decoded = req.url ? decodeURIComponent(req.url) : ""; + if (decoded.includes("did:plc:cde")) { + res.writeHead(200, { "content-type": "application/did+ld+json" }); + res.end(JSON.stringify({ + "@context": ["https://www.w3.org/ns/did/v1"], + id: "did:plc:cde", + alsoKnownAs: ["at://carol.test"], + verificationMethod: [], + service: [ + { id: "#atproto_pds", type: "AtprotoPersonalDataServer", serviceEndpoint: "http://pds.private.test" }, + ], + })); + } else { + res.writeHead(404); + res.end(); + } + }); + + const resolver = new CompositeDidDocumentResolver({ + methods: { + plc: new PlcDidDocumentResolver({ apiUrl: plc.url }), + web: new WebDidDocumentResolver(), + }, + }); + const config: ContrailConfig = { + ...baseConfig, + networkOverrides: { + slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`, + resolver, + additionalAllowedHosts: ["pds.private.test"], + }, + }; + const result = await resolvePDS("did:plc:cde", config); + expect(result?.pds).toBe("http://pds.private.test"); + expect(plc.hits.length).toBeGreaterThan(0); + }); + + it("refreshStaleIdentities persists pds via override path", async () => { + slingshot.setHandler((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ did: "did:plc:def", handle: "dave.test", pds: "http://pds.private.test" })); + }); + const config: ContrailConfig = { + ...baseConfig, + networkOverrides: { + slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`, + additionalAllowedHosts: ["pds.private.test"], + }, + }; + const db = await createTestDbWithSchema(); + await refreshStaleIdentities(db, ["did:plc:def"], config); + + const row = await db + .prepare("SELECT did, pds FROM identities WHERE did = ?") + .bind("did:plc:def") + .first<{ did: string; pds: string }>(); + expect(row?.pds).toBe("http://pds.private.test"); + expect(slingshot.hits.length).toBeGreaterThan(0); + }); + + it("getClient with override config resolves PDS via stubbed slingshot", async () => { + slingshot.setHandler((_req, res) => { + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ did: "did:plc:efg", handle: "eve.test", pds: "http://pds.private.test" })); + }); + const config: ContrailConfig = { + ...baseConfig, + networkOverrides: { + slingshotUrl: `${slingshot.url}/xrpc/com.bad-example.identity.resolveMiniDoc`, + additionalAllowedHosts: ["pds.private.test"], + }, + }; + const db = await createTestDbWithSchema(); + const client = await getClient("did:plc:efg" as Did, db, config); + expect(client).toBeDefined(); + expect(slingshot.hits.some((h) => h.url.includes("identifier=did%3Aplc%3Aefg"))).toBe(true); + }); +}); diff --git a/packages/contrail/tests/postgres-concurrent-init.test.ts b/packages/contrail/tests/postgres-concurrent-init.test.ts new file mode 100644 index 0000000..4545a0d --- /dev/null +++ b/packages/contrail/tests/postgres-concurrent-init.test.ts @@ -0,0 +1,92 @@ +import { describe, it, expect, beforeAll, afterAll, beforeEach } from "vitest"; +import pg from "pg"; +import { createPostgresDatabase } from "../src/adapters/postgres"; +import { initSchema } from "../src/core/db/schema"; +import { resolveConfig } from "../src/core/types"; + +/** + * Postgres-dialect concurrent-init race. + * + * SQLite serializes DDL globally, so the existing `schema-idempotency.test.ts` + * (which uses `createSqliteDatabase`) can't surface the Postgres-specific race + * where two concurrent `CREATE TABLE IF NOT EXISTS` statements both pass the + * existence check and then both try to insert into pg_class/pg_type, with the + * loser raising 23505 on `pg_type_typname_nsp_index` (the unique index on + * (typname, typnamespace)). + * + * Real-world hit: discovered during PR44 Phase C local validation when the OM + * API consumer's `contrail-init-idempotency.spec.ts` ran three parallel + * `contrail.init(db)` calls against a fresh Postgres schema. + */ + +const TEST_CONFIG = resolveConfig({ + namespace: "com.example", + collections: { + event: { + collection: "community.lexicon.calendar.event", + queryable: { mode: {}, name: {}, startsAt: { type: "range" } }, + searchable: ["name", "description"], + relations: { + rsvps: { + collection: "rsvp", + groupBy: "status", + groups: { + going: "community.lexicon.calendar.rsvp#going", + }, + }, + }, + }, + rsvp: { + collection: "community.lexicon.calendar.rsvp", + references: { + event: { + collection: "event", + field: "subject.uri", + }, + }, + }, + }, +}); + +const PG_URL = process.env.TEST_DATABASE_URL; +if (!PG_URL) { + describe.skip("PostgreSQL concurrent init (TEST_DATABASE_URL not set)", () => { + it("skipped", () => {}); + }); +} else { + let pool: pg.Pool; + let db: ReturnType; + + beforeAll(async () => { + pool = new pg.Pool({ connectionString: PG_URL }); + await pool.query("SELECT 1"); + db = createPostgresDatabase(pool); + }); + + afterAll(async () => { + await pool?.end(); + }); + + beforeEach(async () => { + const tables = await pool.query( + `SELECT tablename FROM pg_tables WHERE schemaname = 'public' + AND (tablename LIKE 'records_%' OR tablename LIKE 'fts_%' + OR tablename IN ('backfills', 'discovery', 'cursor', 'identities', 'feed_items', 'feed_backfills'))` + ); + for (const { tablename } of tables.rows) { + await pool.query(`DROP TABLE IF EXISTS ${tablename} CASCADE`); + } + }); + + describe("PostgreSQL initSchema under concurrency", () => { + it("is safe to call three times concurrently against a fresh schema", async () => { + await expect( + Promise.all([ + initSchema(db, TEST_CONFIG), + initSchema(db, TEST_CONFIG), + initSchema(db, TEST_CONFIG), + ]) + ).resolves.not.toThrow(); + }); + }); +} diff --git a/packages/contrail/tests/schema-idempotency.test.ts b/packages/contrail/tests/schema-idempotency.test.ts new file mode 100644 index 0000000..9151132 --- /dev/null +++ b/packages/contrail/tests/schema-idempotency.test.ts @@ -0,0 +1,139 @@ +import { describe, it, expect } from "vitest"; +import { initSchema, addColumnIfNotExists } from "../src/core/db/schema"; +import type { Database } from "../src/core/types"; +import { resolveConfig } from "../src/core/types"; +import { createTestDb, TEST_CONFIG } from "./helpers"; + +/** + * L3 — schema idempotency. + * + * These tests pin the contract that `initSchema` is safe to call repeatedly + * (sequentially or concurrently) and that real DDL errors are NOT silently + * swallowed. The previous implementation wrapped ALTER TABLE statements in + * `try { ... } catch { /* ignore *\/ }` blocks that masked syntax errors, + * missing tables, and type mismatches alongside the intended duplicate-column + * case. After L3, only duplicate-column races are absorbed (via dialect-aware + * IF-NOT-EXISTS / PRAGMA pre-check); other DDL failures surface. + */ + +describe("initSchema idempotency", () => { + it("is safe to call twice sequentially", async () => { + const db = createTestDb(); + await initSchema(db, TEST_CONFIG); + await expect(initSchema(db, TEST_CONFIG)).resolves.not.toThrow(); + }); + + it("is safe to call concurrently", async () => { + const db = createTestDb(); + await Promise.all([ + initSchema(db, TEST_CONFIG), + initSchema(db, TEST_CONFIG), + initSchema(db, TEST_CONFIG), + ]); + + // Verify count column on records_event exists exactly once. The count + // columns are added via ALTER TABLE; duplicate adds would have failed + // without idempotent ALTER. + const cols = await db + .prepare("PRAGMA table_info(records_event)") + .all<{ name: string }>(); + const names = cols.results.map((c) => c.name); + + // There should be exactly one of each grouped-count column (sanity check + // that no parallel run got further than the first). + const countCols = names.filter((n) => n.startsWith("count_")); + const dedup = new Set(countCols); + expect(countCols.length).toBe(dedup.size); + // And we should have at least one count column (config defines rsvp groups). + expect(countCols.length).toBeGreaterThan(0); + }); + + it("does not leave duplicate count columns after repeated init", async () => { + const db = createTestDb(); + for (let i = 0; i < 5; i++) { + await initSchema(db, TEST_CONFIG); + } + const cols = await db + .prepare("PRAGMA table_info(records_event)") + .all<{ name: string }>(); + const names = cols.results.map((c) => c.name); + expect(new Set(names).size).toBe(names.length); + }); +}); + +describe("addColumnIfNotExists", () => { + // The helper is the seam where dialect-aware idempotent ALTER lives. We + // verify it both adds a column when absent and is a no-op when present. + it("adds the column when absent", async () => { + const db = createTestDb(); + await db.prepare("CREATE TABLE t (id INTEGER PRIMARY KEY)").run(); + + await addColumnIfNotExists(db, "t", "extra", "INTEGER NOT NULL DEFAULT 0"); + + const cols = await db.prepare("PRAGMA table_info(t)").all<{ name: string }>(); + expect(cols.results.map((c) => c.name)).toContain("extra"); + }); + + it("is a no-op when the column already exists", async () => { + const db = createTestDb(); + await db.prepare("CREATE TABLE t (id INTEGER PRIMARY KEY, extra INTEGER)").run(); + + // First call: column already present — should not throw. + await expect( + addColumnIfNotExists(db, "t", "extra", "INTEGER NOT NULL DEFAULT 0") + ).resolves.not.toThrow(); + + // Calling it again should still be a no-op. + await expect( + addColumnIfNotExists(db, "t", "extra", "INTEGER NOT NULL DEFAULT 0") + ).resolves.not.toThrow(); + + // And we still have exactly one `extra` column. + const cols = await db.prepare("PRAGMA table_info(t)").all<{ name: string }>(); + const extras = cols.results.filter((c) => c.name === "extra"); + expect(extras.length).toBe(1); + }); + + it("surfaces real DDL errors (target table does not exist)", async () => { + // The previous swallow-all `try { ... } catch { /* ignore */ }` masked + // *any* DDL failure, including target-table-missing. The new helper must + // only absorb the duplicate-column case and surface everything else. + const db = createTestDb(); + + await expect( + addColumnIfNotExists(db, "no_such_table", "x", "INTEGER") + ).rejects.toThrow(); + }); +}); + +describe("initSchema with extra schemas — real DDL errors surface", () => { + // Belt-and-suspenders: confirm that a genuine failure inside an extension + // schema module propagates rather than being absorbed. This is the + // user-visible behavior change from L3. + it("propagates errors thrown from an extra schema", async () => { + const db = createTestDb(); + const broken = async (_db: Database) => { + throw new Error("synthetic DDL failure"); + }; + + await expect( + initSchema(db, TEST_CONFIG, { extraSchemas: [broken] }) + ).rejects.toThrow("synthetic DDL failure"); + }); + + it("treats a config without spaces/labels/feeds the same way (sanity)", async () => { + // Minimal config with no relations — should still init cleanly twice. + const minimal = resolveConfig({ + namespace: "com.example", + collections: { + foo: { + collection: "com.example.foo", + queryable: {}, + }, + }, + }); + const db = createTestDb(); + await initSchema(db, minimal); + await expect(initSchema(db, minimal)).resolves.not.toThrow(); + }); +}); diff --git a/packages/contrail/tests/spaces-auth.test.ts b/packages/contrail/tests/spaces-auth.test.ts new file mode 100644 index 0000000..805c844 --- /dev/null +++ b/packages/contrail/tests/spaces-auth.test.ts @@ -0,0 +1,65 @@ +import { describe, it, expect } from "vitest"; +import { + PlcDidDocumentResolver, + CompositeDidDocumentResolver, + WebDidDocumentResolver, +} from "@atcute/identity-resolver"; +import { buildVerifier } from "../src/core/spaces/auth"; +import type { AuthorityConfig } from "../src/core/spaces/types"; + +const CUSTOM_PLC = "http://custom-plc.test"; + +function makeAuthority(overrides: Partial = {}): AuthorityConfig { + return { + type: "tools.atmo.event.space", + serviceDid: "did:web:authority.test", + ...overrides, + } as AuthorityConfig; +} + +// ServiceJwtVerifier from @atcute/xrpc-server@0.1.12 exposes the resolver as +// the public instance field `didDocResolver` (verified against +// node_modules/.../auth/jwt-verifier.d.ts). The plan's hint at `.resolver` was +// a guess — we use the real field here so the test verifies the resolver +// actually wired into the verifier instance. +describe("buildVerifier resolver precedence", () => { + it("uses AuthorityConfig.resolver when provided (most-specific wins)", () => { + const specific = new CompositeDidDocumentResolver({ + methods: { + plc: new PlcDidDocumentResolver({ apiUrl: "http://specific.test" }), + web: new WebDidDocumentResolver(), + }, + }); + const network = new CompositeDidDocumentResolver({ + methods: { + plc: new PlcDidDocumentResolver({ apiUrl: CUSTOM_PLC }), + web: new WebDidDocumentResolver(), + }, + }); + const verifier = buildVerifier(makeAuthority({ resolver: specific }), { + resolver: network, + }); + expect(verifier.didDocResolver).toBe(specific); + }); + + it("falls back to networkOverrides.resolver when authority resolver is absent", () => { + const network = new CompositeDidDocumentResolver({ + methods: { + plc: new PlcDidDocumentResolver({ apiUrl: CUSTOM_PLC }), + web: new WebDidDocumentResolver(), + }, + }); + const verifier = buildVerifier(makeAuthority(), { resolver: network }); + expect(verifier.didDocResolver).toBe(network); + }); + + it("falls back to default composite when both are absent", () => { + const verifier = buildVerifier(makeAuthority(), {}); + expect(verifier.didDocResolver).toBeInstanceOf(CompositeDidDocumentResolver); + }); + + it("treats omitted second arg the same as empty networkOverrides (backward-compat)", () => { + const verifier = buildVerifier(makeAuthority()); + expect(verifier.didDocResolver).toBeInstanceOf(CompositeDidDocumentResolver); + }); +}); -- 2.51.2