From 41f4918d0502462cef28cacecd07d5f886e238ac Mon Sep 17 00:00:00 2001 From: ditadi Date: Fri, 23 Oct 2026 14:00:00 +0100 Subject: [PATCH] feat(appkit): verify database callers and authorize named operations Add the operation request pipeline: verify the caller with the workspace, refuse cross-site requests, parse the input, run the policy once and execute once, then validate and project the answer. The pipeline is not registered yet, so no request reaches it. Default outputs answer timestamps in RFC 3339, and span redaction matches the routed path. Signed-off-by: ditadi --- NOTICE.md | 1 + packages/appkit/package.json | 1 + packages/appkit/src/core/appkit.ts | 9 + packages/appkit/src/core/plugin-context.ts | 18 + .../appkit/src/core/tests/service-api.test.ts | 51 + packages/appkit/src/database/errors.ts | 6 +- .../src/plugins/database/crud/codecs.ts | 23 +- .../database/crud/tests/codecs.test.ts | 17 + .../database/crud/tests/routes.test.ts | 4 +- .../appkit/src/plugins/database/defaults.ts | 2 + .../src/plugins/database/operation-actor.ts | 234 +++ .../plugins/database/operation-metadata.ts | 2 +- .../src/plugins/database/operation-runtime.ts | 695 +++++++++ .../database/tests/operation-actor.test.ts | 61 + .../plugins/database/tests/operation-edges.ts | 197 +++ .../database/tests/operation-runtime.test.ts | 1380 +++++++++++++++++ .../src/plugins/server/dev-obo-middleware.ts | 3 +- .../appkit/src/telemetry/instrumentations.ts | 34 +- packages/appkit/src/utils/routed-path.ts | 11 + .../src/utils/tests/routed-path.test.ts | 22 + packages/shared/src/database/errors.ts | 2 + .../shared/src/database/query-codec.test.ts | 4 +- pnpm-lock.yaml | 3 + 23 files changed, 2771 insertions(+), 9 deletions(-) create mode 100644 packages/appkit/src/core/tests/service-api.test.ts create mode 100644 packages/appkit/src/plugins/database/operation-actor.ts create mode 100644 packages/appkit/src/plugins/database/operation-runtime.ts create mode 100644 packages/appkit/src/plugins/database/tests/operation-actor.test.ts create mode 100644 packages/appkit/src/plugins/database/tests/operation-edges.ts create mode 100644 packages/appkit/src/plugins/database/tests/operation-runtime.test.ts create mode 100644 packages/appkit/src/utils/routed-path.ts create mode 100644 packages/appkit/src/utils/tests/routed-path.test.ts diff --git a/NOTICE.md b/NOTICE.md index c732ed57d..fbfaa5149 100644 --- a/NOTICE.md +++ b/NOTICE.md @@ -11,6 +11,7 @@ This Software contains code from the following open source projects: | [@opentelemetry/api](https://www.npmjs.com/package/@opentelemetry/api) | 1.9.0 | Apache-2.0 | https://github.com/open-telemetry/opentelemetry-js/tree/main/api | | [@opentelemetry/api-logs](https://www.npmjs.com/package/@opentelemetry/api-logs) | 0.219.0 | Apache-2.0 | https://github.com/open-telemetry/opentelemetry-js/tree/main/experimental/packages/api-logs | | [@opentelemetry/auto-instrumentations-node](https://www.npmjs.com/package/@opentelemetry/auto-instrumentations-node) | 0.77.0 | Apache-2.0 | https://github.com/open-telemetry/opentelemetry-js-contrib/tree/main/packages/auto-instrumentations-node#readme | +| [@opentelemetry/core](https://www.npmjs.com/package/@opentelemetry/core) | 2.8.0 | Apache-2.0 | https://github.com/open-telemetry/opentelemetry-js/tree/main/packages/opentelemetry-core | | [@opentelemetry/exporter-logs-otlp-proto](https://www.npmjs.com/package/@opentelemetry/exporter-logs-otlp-proto) | 0.219.0 | Apache-2.0 | https://github.com/open-telemetry/opentelemetry-js/tree/main/experimental/packages/exporter-logs-otlp-proto | | [@opentelemetry/exporter-metrics-otlp-proto](https://www.npmjs.com/package/@opentelemetry/exporter-metrics-otlp-proto) | 0.219.0 | Apache-2.0 | https://github.com/open-telemetry/opentelemetry-js/tree/main/experimental/packages/opentelemetry-exporter-metrics-otlp-proto | | [@opentelemetry/exporter-trace-otlp-proto](https://www.npmjs.com/package/@opentelemetry/exporter-trace-otlp-proto) | 0.219.0 | Apache-2.0 | https://github.com/open-telemetry/opentelemetry-js/tree/main/experimental/packages/exporter-trace-otlp-proto | diff --git a/packages/appkit/package.json b/packages/appkit/package.json index 968531019..03dd40704 100644 --- a/packages/appkit/package.json +++ b/packages/appkit/package.json @@ -93,6 +93,7 @@ "@opentelemetry/api": "1.9.0", "@opentelemetry/api-logs": "0.219.0", "@opentelemetry/auto-instrumentations-node": "0.77.0", + "@opentelemetry/core": "2.8.0", "@opentelemetry/exporter-logs-otlp-proto": "0.219.0", "@opentelemetry/exporter-metrics-otlp-proto": "0.219.0", "@opentelemetry/exporter-trace-otlp-proto": "0.219.0", diff --git a/packages/appkit/src/core/appkit.ts b/packages/appkit/src/core/appkit.ts index 7de60ce2a..a14deac8c 100644 --- a/packages/appkit/src/core/appkit.ts +++ b/packages/appkit/src/core/appkit.ts @@ -135,6 +135,15 @@ export class AppKit { }); } } + + const serviceApi: Record = Object.create(null); + for (const name of Object.keys(this.#pluginInstances)) { + Object.defineProperty(serviceApi, name, { + enumerable: true, + get: () => this.wrapWithAsUser(this.#pluginInstances[name]), + }); + } + this.#context.registerServiceApi(Object.freeze(serviceApi)); } private createAndRegisterPlugin( diff --git a/packages/appkit/src/core/plugin-context.ts b/packages/appkit/src/core/plugin-context.ts index 27603bdc1..e31746a2d 100644 --- a/packages/appkit/src/core/plugin-context.ts +++ b/packages/appkit/src/core/plugin-context.ts @@ -73,6 +73,7 @@ export class PluginContext { LifecycleEvent, Set<() => void | Promise> >(); + private serviceApi: Readonly> | undefined; private telemetry: ITelemetry; private createCallerContext: typeof ServiceContext.createCallerContext; @@ -185,6 +186,23 @@ export class PluginContext { this.plugins.set(name, instance); } + /** + * Publish the service-principal handle: each plugin's exports, read the way + * `appkit.` reads them. + * @internal Called once by AppKit core after every plugin is registered. + */ + registerServiceApi(api: Readonly>): void { + this.serviceApi = api; + } + + /** + * The service-principal handle keyed by plugin name, for plugins that run + * application code against sibling plugins. `undefined` outside AppKit core. + */ + getServiceApi(): Readonly> | undefined { + return this.serviceApi; + } + /** * Returns all registered plugin instances keyed by name. * Used by the server plugin for route injection, client config, diff --git a/packages/appkit/src/core/tests/service-api.test.ts b/packages/appkit/src/core/tests/service-api.test.ts new file mode 100644 index 000000000..fcbe95702 --- /dev/null +++ b/packages/appkit/src/core/tests/service-api.test.ts @@ -0,0 +1,51 @@ +import { describe, expect, test } from "vitest"; + +import { getCurrentPrincipalKey } from "../../context"; +import { Plugin, toPlugin } from "../../plugin"; +import { createTestApp } from "../../testing"; + +const manifest = { + displayName: "Probe", + description: "Service handle probe", + resources: { required: [], optional: [] }, +}; + +class LedgerPlugin extends Plugin { + static manifest = { ...manifest, name: "ledger" as const }; + private entries = 0; + record() { + this.entries += 1; + return `${getCurrentPrincipalKey()}:${this.entries}`; + } + exports() { + return { record: this.record }; + } +} + +class RunnerPlugin extends Plugin { + static manifest = { ...manifest, name: "runner" as const }; + /** What a plugin running application code would hand that code. */ + serviceApi() { + return this.context?.getServiceApi(); + } + exports() { + return { serviceApi: () => this.serviceApi() }; + } +} + +describe("PluginContext service handle", () => { + test("lets a plugin reach every sibling's service-principal exports", async () => { + await using app = await createTestApp({ + plugins: [toPlugin(LedgerPlugin)(), toPlugin(RunnerPlugin)()], + server: false, + }); + + const api = app.plugins.runner.serviceApi() as Record; + + expect(Object.keys(api).sort()).toEqual(["ledger", "runner"]); + expect(Object.isFrozen(api)).toBe(true); + // The same bound handle `appkit.ledger` returns, in service scope. + expect(api.ledger.record()).toBe("app:1"); + expect(app.plugins.ledger.record()).toBe("app:2"); + }); +}); diff --git a/packages/appkit/src/database/errors.ts b/packages/appkit/src/database/errors.ts index 12d810d34..1fa48fa59 100644 --- a/packages/appkit/src/database/errors.ts +++ b/packages/appkit/src/database/errors.ts @@ -25,6 +25,10 @@ const definitions: Record< DatabaseErrorCategory, { readonly message: string; readonly statusCode: number } > = { + UNAUTHENTICATED: { + message: "Database request is not authenticated", + statusCode: 401, + }, INVALID_REQUEST: { message: "Invalid database request", statusCode: 400 }, VALIDATION_FAILED: { message: "Database request failed validation", @@ -42,7 +46,7 @@ const definitions: Record< statusCode: 415, }, PAYLOAD_TOO_LARGE: { - message: "Database response is too large", + message: "Database payload is too large", statusCode: 413, }, INTERNAL: { message: "Database operation failed", statusCode: 500 }, diff --git a/packages/appkit/src/plugins/database/crud/codecs.ts b/packages/appkit/src/plugins/database/crud/codecs.ts index a474859fc..aea42c847 100644 --- a/packages/appkit/src/plugins/database/crud/codecs.ts +++ b/packages/appkit/src/plugins/database/crud/codecs.ts @@ -24,6 +24,25 @@ export interface CompiledColumn { /** Decimal integer with no sign padding, leading zeros, or exponent. */ const DECIMAL_INT = /^-?(0|[1-9]\d*)$/; +/** + * PostgreSQL's ISO text output for a timestamp, such as + * `2024-01-01 10:00:00.5` or `2024-01-01 10:00:00+05:30`. + */ +const PG_TIMESTAMP = + /^(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}:\d{2}(?:\.\d+)?)(?:([+-]\d{2})(?::(\d{2}))?)?$/; + +/** + * Answer a timestamp in RFC 3339 form, which browsers parse and the decoder + * accepts back: a `T` between date and time, and a `±HH:MM` offset. Any other + * text, such as `infinity`, is answered as the driver gave it. + */ +function canonicalTimestamp(value: string): string { + const match = PG_TIMESTAMP.exec(value); + if (!match) return value; + const [, date, time, hours, minutes = "00"] = match; + return `${date}T${time}${hours === undefined ? "" : `${hours}:${minutes}`}`; +} + /** * Map a wire value onto the runtime representation the schema validators and * Drizzle columns expect. Only path segments and JSON scalars reach this, so @@ -75,9 +94,11 @@ function encodeValue(meta: ColumnMeta, value: unknown): JsonValue { case "string": case "uuid": case "enum": - case "date": if (typeof value === "string") return value; break; + case "date": + if (typeof value === "string") return canonicalTimestamp(value); + break; case "json": if (isJsonValue(value)) return value; break; diff --git a/packages/appkit/src/plugins/database/crud/tests/codecs.test.ts b/packages/appkit/src/plugins/database/crud/tests/codecs.test.ts index 46ce5387c..008afb20d 100644 --- a/packages/appkit/src/plugins/database/crud/tests/codecs.test.ts +++ b/packages/appkit/src/plugins/database/crud/tests/codecs.test.ts @@ -90,6 +90,23 @@ describe("compileColumn encode", () => { expect(column("name").encode(undefined)).toBeNull(); }); + it("answers PostgreSQL timestamp text in a form the decoder accepts back", () => { + const createdAt = column("createdAt"); + const samples: [string, string][] = [ + ["2024-01-01 10:00:00", "2024-01-01T10:00:00"], + ["2024-01-01 10:00:00.123456", "2024-01-01T10:00:00.123456"], + ["2024-01-01 10:00:00+00", "2024-01-01T10:00:00+00:00"], + ["2024-01-01 10:00:00.5-03", "2024-01-01T10:00:00.5-03:00"], + ["2024-01-01 10:00:00+05:30", "2024-01-01T10:00:00+05:30"], + ]; + for (const [text, wire] of samples) { + expect(createdAt.encode(text)).toBe(wire); + expect(createdAt.decode(createdAt.encode(text))).toBe(wire); + } + // Text that is not an ISO timestamp is answered as the driver gave it. + expect(createdAt.encode("infinity")).toBe("infinity"); + }); + it("fails closed when a driver value contradicts its column", () => { expect(() => column("name").encode(7)).toThrow(DatabasePluginError); expect(() => column("createdAt").encode(new Date())).toThrow( diff --git a/packages/appkit/src/plugins/database/crud/tests/routes.test.ts b/packages/appkit/src/plugins/database/crud/tests/routes.test.ts index 5dda67821..161cf00e5 100644 --- a/packages/appkit/src/plugins/database/crud/tests/routes.test.ts +++ b/packages/appkit/src/plugins/database/crud/tests/routes.test.ts @@ -340,7 +340,7 @@ describe("projection and response limits", () => { ); expect(response.sent.status).toBe(413); expect(response.json()).toEqual({ - error: "Database response is too large", + error: "Database payload is too large", }); expect(MAX_RESPONSE_BYTES).toBeLessThan(6 * wide.length); }); @@ -364,7 +364,7 @@ describe("projection and response limits", () => { ); expect(response.sent.status).toBe(413); expect(response.json()).toEqual({ - error: "Database response is too large", + error: "Database payload is too large", }); expect(touchedSentinel).toBe(false); }); diff --git a/packages/appkit/src/plugins/database/defaults.ts b/packages/appkit/src/plugins/database/defaults.ts index 2a473ce84..3cc4e8fc0 100644 --- a/packages/appkit/src/plugins/database/defaults.ts +++ b/packages/appkit/src/plugins/database/defaults.ts @@ -32,6 +32,8 @@ export const TRANSACTION_TIMEOUT_MS = 30_000; // Request decoding: rejected before any database work is scheduled. /** Max encoded size of a generated read query string. */ export const MAX_QUERY_BYTES = 8 * 1024; +/** Max UTF-8 size of one operation request body. */ +export const MAX_OPERATION_BODY_BYTES = 1024 * 1024; /** Max nesting depth of `and`/`or` groups in one generated filter. */ export const MAX_WHERE_DEPTH = 5; /** Max column conditions across one generated filter. */ diff --git a/packages/appkit/src/plugins/database/operation-actor.ts b/packages/appkit/src/plugins/database/operation-actor.ts new file mode 100644 index 000000000..6d3106a78 --- /dev/null +++ b/packages/appkit/src/plugins/database/operation-actor.ts @@ -0,0 +1,234 @@ +import { createHash } from "node:crypto"; + +import type { Request } from "express"; +import { + createDevOboIdentityProvider, + loadDevOboIdentityFromEnvironment, +} from "shared"; + +import { getClientOptions } from "../../context/client-options"; +import { DatabasePluginError } from "../../database/errors"; +import { createLogger } from "../../logging/logger"; +import { ApiError, createWorkspaceClient } from "../../workspace-client"; +import { isLocalRequest } from "../server/dev-obo-middleware"; +import type { Actor } from "./operations"; + +const logger = createLogger("database:operations"); + +/** How long the workspace may take to name the caller. */ +const ME_TIMEOUT_MS = 10_000; +/** An OAuth access token is a few kilobytes; a longer header is not one. */ +const MAX_TOKEN_BYTES = 16 * 1024; +const MAX_ORIGIN_LENGTH = 2048; + +/** The signed-in developer the local OBO loader verified with the workspace. */ +type LocalIdentity = Awaited< + ReturnType +>; + +/** + * Where operation requests come from, chosen once from deployment + * configuration. Choosing a boundary authenticates nobody: an Apps request + * still needs a token the workspace accepts, and a local request still needs + * the signed-in developer and a loopback origin. + */ +type ActorBoundary = + | { readonly kind: "apps"; readonly host: string } + | { + readonly kind: "local"; + readonly host: string; + readonly identity: () => Promise; + } + | { readonly kind: "unsupported" }; + +/** A verified caller and the opaque session discriminator its responses carry. */ +export interface AuthenticatedActor { + readonly actor: Actor; + readonly discriminator: string; +} + +/** Resolve the caller once per request; `undefined` means unauthenticated. */ +export type ActorResolver = ( + req: Request, +) => Promise; + +function canonicalHost(host: string): string | undefined { + try { + return new URL(/^https?:\/\//i.test(host) ? host : `https://${host}`) + .origin; + } catch { + return undefined; + } +} + +/** + * Pick the boundary from deployment configuration. `host` is the workspace + * the service principal is configured for, never a request header. Local + * development needs the explicit credentials the local OBO middleware uses. + * Every other deployment fails closed. + */ +export function selectActorBoundary( + host: string | undefined, + env: NodeJS.ProcessEnv = process.env, +): ActorBoundary { + const workspace = host ? canonicalHost(host) : undefined; + if (!workspace) return { kind: "unsupported" }; + if (env.NODE_ENV === "development") { + const explicit = + env.DATABRICKS_TOKEN?.trim() || env.DATABRICKS_CONFIG_PROFILE?.trim(); + if (!explicit || env.APPKIT_DEV_OBO === "false") { + return { kind: "unsupported" }; + } + return { + kind: "local", + host: workspace, + identity: createDevOboIdentityProvider(() => + loadDevOboIdentityFromEnvironment(env), + ), + }; + } + return env.DATABRICKS_APP_NAME + ? { kind: "apps", host: workspace } + : { kind: "unsupported" }; +} + +function errorName(error: unknown): string { + return error instanceof Error ? error.name : typeof error; +} + +/** Read the workspace's answer without trusting any field it lacks. */ +function actorFromUser(user: { + readonly id?: string; + readonly userName?: string; + readonly active?: boolean; + readonly schemas?: readonly string[]; +}): Actor | undefined { + if ( + typeof user.id !== "string" || + user.id.length === 0 || + user.active === false || + user.schemas?.some((schema) => String(schema).endsWith(":ServicePrincipal")) + ) { + return undefined; + } + return Object.freeze( + typeof user.userName === "string" + ? { id: user.id, email: user.userName } + : { id: user.id }, + ); +} + +/** + * Ask the workspace who owns the forwarded user token. The client carries + * only that token, so a rejected token can never fall back to the service + * principal, and identity headers beside it are never read. + */ +async function verifyForwardedToken( + host: string, + req: Request, +): Promise { + const header = req.headers["x-forwarded-access-token"]; + const token = typeof header === "string" ? header.trim() : ""; + if (!token || Buffer.byteLength(token, "utf8") > MAX_TOKEN_BYTES) { + return undefined; + } + const client = createWorkspaceClient({ + token, + host, + clientOptions: getClientOptions(), + }); + try { + return actorFromUser( + await client.currentUser.me( + {}, + { signal: AbortSignal.timeout(ME_TIMEOUT_MS) }, + ), + ); + } catch (error) { + if ( + error instanceof ApiError && + (error.statusCode === 401 || error.statusCode === 403) + ) { + return undefined; + } + logger.warn( + "Workspace identity check failed (%s); answering 503", + errorName(error), + ); + throw new DatabasePluginError("TRANSIENT", "read"); + } +} + +async function localActor( + identity: () => Promise, +): Promise { + try { + const { userId, email } = await identity(); + return Object.freeze(email ? { id: userId, email } : { id: userId }); + } catch { + // The loader's failure can carry CLI output; only the fact is logged. + logger.warn( + "Local database operations could not load the signed-in Databricks identity", + ); + return undefined; + } +} + +/** SHA-256 of the workspace and the canonical actor id, joined by NUL. */ +function actorDiscriminator(host: string, actorId: string): string { + return createHash("sha256").update(`${host}\0${actorId}`).digest("hex"); +} + +/** Resolve operation callers at the selected boundary. */ +export function createActorResolver(boundary: ActorBoundary): ActorResolver { + const authenticated = ( + host: string, + actor: Actor | undefined, + ): AuthenticatedActor | undefined => + actor && { actor, discriminator: actorDiscriminator(host, actor.id) }; + + switch (boundary.kind) { + case "apps": + return async (req) => + authenticated( + boundary.host, + await verifyForwardedToken(boundary.host, req), + ); + case "local": + // Forwarded headers are ignored: only the signed-in developer, on loopback. + return async (req) => + isLocalRequest(req) + ? authenticated(boundary.host, await localActor(boundary.identity)) + : undefined; + case "unsupported": + return async () => undefined; + } +} + +function forwardedHost(req: Request): string | undefined { + const forwarded = req.headers["x-forwarded-host"]; + if (typeof forwarded === "string") return forwarded.split(",")[0]?.trim(); + return req.headers.host; +} + +/** + * Whether a browser could have sent this request only from the app's own + * pages. Fetch metadata decides when present; reads also accept a direct + * navigation. Without it, an `Origin` must name the host the request reached. + * A request with neither header did not come from a browser page. + */ +export function isSameOriginRequest(req: Request): boolean { + const site = req.headers["sec-fetch-site"]; + if (site !== undefined) { + return site === "same-origin" || (site === "none" && req.method === "GET"); + } + const origin = req.headers.origin; + if (origin === undefined) return true; + if (origin.length > MAX_ORIGIN_LENGTH) return false; + try { + const url = new URL(origin); + return url.origin === origin && url.host === forwardedHost(req); + } catch { + return false; + } +} diff --git a/packages/appkit/src/plugins/database/operation-metadata.ts b/packages/appkit/src/plugins/database/operation-metadata.ts index 337e7b5e4..1f80f8825 100644 --- a/packages/appkit/src/plugins/database/operation-metadata.ts +++ b/packages/appkit/src/plugins/database/operation-metadata.ts @@ -11,7 +11,7 @@ const OPERATION_NAME = /^[A-Za-z][A-Za-z0-9_-]{0,63}(?:\.[A-Za-z][A-Za-z0-9_-]{0,63})*$/; /** Longest operation name, so a request path stays bounded. */ -const MAX_OPERATION_NAME_LENGTH = 255; +export const MAX_OPERATION_NAME_LENGTH = 255; /** Names a plain object cannot hold as its own key, as the boot payload does. */ const RESERVED_NAMES: ReadonlySet = new Set([ diff --git a/packages/appkit/src/plugins/database/operation-runtime.ts b/packages/appkit/src/plugins/database/operation-runtime.ts new file mode 100644 index 000000000..0fae4ea94 --- /dev/null +++ b/packages/appkit/src/plugins/database/operation-runtime.ts @@ -0,0 +1,695 @@ +import { context, type Span, SpanStatusCode } from "@opentelemetry/api"; +import { suppressTracing } from "@opentelemetry/core"; +import express, { + type Request, + type RequestHandler, + type Response, +} from "express"; + +import { + classifyDatabaseError, + DatabasePluginError, + databaseErrorFromStatus, + databaseSetupFailed, + invalidDatabaseInput, +} from "../../database/errors"; +import type { Row } from "../../database/runtime"; +import type { Schema } from "../../database/schema-builder"; +import { DatabaseValidationError } from "../../errors"; +import { createLogger } from "../../logging/logger"; +import type { ITelemetry } from "../../telemetry"; +import type { JsonValue } from "./crud/codecs"; +import { + boundedJson, + compileCrudTables, + type CrudTable, + isPlainObject, +} from "./crud/contract"; +import { decodeCreateBody, decodeUpdateBody } from "./crud/request"; +import { sendEmpty, sendJson } from "./crud/response"; +import { MAX_OPERATION_BODY_BYTES, MAX_QUERY_BYTES } from "./defaults"; +import type { DatabaseExports } from "./entity-types"; +import { type ActorResolver, isSameOriginRequest } from "./operation-actor"; +import { + MAX_OPERATION_NAME_LENGTH, + type validateOperationMetadata, +} from "./operation-metadata"; +import type { OperationAppKit, OperationMethod } from "./operations"; + +const logger = createLogger("database:operations"); + +/** One declaration exactly as the shared metadata validator admits it. */ +type ValidatedOperation = ReturnType< + typeof validateOperationMetadata +>["operations"][number]; + +/** + * Every authenticated response carries this opaque session discriminator: the + * SHA-256 of the workspace host, NUL, and the canonical actor id. It is not a + * credential and names no one; a 401 never carries it. + */ +const DATABASE_ACTOR_HEADER = "x-appkit-database-actor"; + +/** Mount point below the plugin router: `router.use(OPERATIONS_PATH, handler)`. */ +export const OPERATIONS_PATH = "/operations"; + +/** Methods whose `null` result is a missing record unless an output accepts it. */ +const NULLABLE_METHODS: ReadonlySet = new Set([ + "GET", + "PATCH", + "PUT", +]); + +/** The service-principal handles an operation reaches. */ +interface OperationServices { + readonly db: DatabaseExports; + readonly appkit: OperationAppKit; +} + +interface OperationHandlerOptions { + readonly schema: Schema; + readonly operations: readonly ValidatedOperation[]; + readonly resolveActor: ActorResolver; + /** Read per admitted request, so a draining plugin serves no stale client. */ + readonly services: () => OperationServices; + readonly telemetry: Pick; +} + +type Reply = + | { readonly status: 200 | 201; readonly body: JsonValue } + | { readonly status: 204 }; + +type Decode = (raw: JsonValue) => unknown; + +/** Validate an answer and shape the reply; a schema may refine asynchronously. */ +type Respond = (result: unknown) => Reply | Promise; + +interface CompiledOperation { + readonly name: string; + readonly method: OperationMethod; + readonly decode: Decode; + readonly authorize: ValidatedOperation["authorize"]; + readonly execute: ValidatedOperation["execute"]; + readonly reply: Respond; +} + +/** Where a request stopped. A span attribute, never part of a response. */ +type Stage = + | "actor" + | "origin" + | "route" + | "input" + | "resource" + | "allow" + | "execute" + | "output"; + +function errorName(error: unknown): string { + return error instanceof Error ? error.name : typeof error; +} + +/** Public scalar columns only; a relation is never part of a default reply. */ +function publicColumns(table: CrudTable, row: unknown): JsonValue { + if (!isPlainObject(row)) throw new DatabasePluginError("INTERNAL", "read"); + const out: Record = {}; + for (const [name, value] of Object.entries(row)) { + const column = table.columns.get(name); + if (column && !column.meta.isPrivate) out[name] = column.encode(value); + } + return out; +} + +function rowOrMissing(table: CrudTable): (result: unknown) => Reply { + return (result) => { + if (result === null) throw new DatabasePluginError("NOT_FOUND", "read"); + return { status: 200, body: publicColumns(table, result) }; + }; +} + +/** Refuse a body that leaves out a column the contract requires. */ +function requireColumns(columns: Iterable, values: Row): Row { + for (const column of columns) { + if (!Object.hasOwn(values, column)) { + throw invalidDatabaseInput([column], "Required"); + } + } + return values; +} + +/** + * Decode one member of a keyed input, naming its refusals inside it: a field + * detail reads `["changes", "title"]`, and a refusal of the whole member reads + * `["changes"]`. + */ +function member( + name: "changes" | "values", + decode: (raw: unknown) => Row, +): { + readonly name: "changes" | "values"; + readonly decode: (raw: unknown) => Row; +} { + return { + name, + decode: (raw) => { + try { + return decode(raw); + } catch (error) { + if (!(error instanceof DatabasePluginError) || !error.details) { + throw error; + } + throw new DatabasePluginError( + error.category, + error.phase, + undefined, + error.details.map((detail) => ({ + path: + detail.path.length === 1 && detail.path[0] === "body" + ? [name] + : [name, ...detail.path], + message: detail.message, + })), + ); + } + }, + }; +} + +/** `{ id }`, plus one member of column values for an update. */ +function keyedInput( + table: CrudTable, + member?: { + readonly name: "changes" | "values"; + readonly decode: (raw: unknown) => Row; + }, +): Decode | undefined { + const key = table.primaryKey; + if (!key) return undefined; + const members = new Set(member ? ["id", member.name] : ["id"]); + return (raw) => { + if ( + !isPlainObject(raw) || + Object.keys(raw).some((name) => !members.has(name)) + ) { + throw invalidDatabaseInput(["input"], "Unexpected input shape"); + } + const id = key.decode(raw.id); + if (id === undefined) { + throw invalidDatabaseInput(["id"], "Not a valid identifier"); + } + return member + ? { id, [member.name]: member.decode(raw[member.name]) } + : { id }; + }; +} + +/** + * The conventional contract of each method, compiled from the table's public + * HTTP facet. An `undefined` input means the table has no default for that + * method, so the operation must declare its own. + */ +const DEFAULTS: Readonly< + Record< + OperationMethod, + { + readonly input: (table: CrudTable) => Decode | undefined; + readonly reply: (table: CrudTable) => (result: unknown) => Reply; + } + > +> = { + GET: { input: (table) => keyedInput(table), reply: rowOrMissing }, + POST: { + input: (table) => { + const required = [...table.creatable].filter((name) => { + const meta = table.columns.get(name)?.meta; + return meta?.notNull && !meta.hasDefault; + }); + return (raw) => requireColumns(required, decodeCreateBody(table, raw)); + }, + reply: (table) => (result) => ({ + status: 201, + body: publicColumns(table, result), + }), + }, + PATCH: { + input: (table) => + table.updatable.size === 0 + ? undefined + : keyedInput( + table, + member("changes", (raw) => { + const changes = decodeUpdateBody(table, raw); + if (Object.keys(changes).length === 0) { + throw invalidDatabaseInput( + ["body"], + "Expected at least one change", + ); + } + return changes; + }), + ), + reply: rowOrMissing, + }, + PUT: { + input: (table) => + table.updatable.size === 0 + ? undefined + : keyedInput( + table, + member("values", (raw) => + requireColumns(table.updatable, decodeUpdateBody(table, raw)), + ), + ), + reply: rowOrMissing, + }, + DELETE: { + input: (table) => keyedInput(table), + reply: () => (result) => { + if (result === true) return { status: 204 }; + throw new DatabasePluginError( + result === false ? "NOT_FOUND" : "INTERNAL", + "write", + ); + }, + }, +}; + +type Parse = ( + value: unknown, +) => Promise< + { readonly ok: true; readonly data: unknown } | { readonly ok: false } +>; + +function setupFailed( + operation: ValidatedOperation, + problem: string, +): DatabasePluginError { + return databaseSetupFailed( + `${operation.file}#${operation.exportName} ${problem}`, + ); +} + +/** + * Narrow a validated schema slot once; only the parse result is read back. + * Parsing is asynchronous, so async refinements and transforms run too. + */ +function parserOf(operation: ValidatedOperation, schema: object): Parse { + const safeParseAsync: unknown = Reflect.get(schema, "safeParseAsync"); + if (typeof safeParseAsync !== "function") { + throw setupFailed(operation, "has a schema without safeParseAsync."); + } + return async (value) => { + const result: unknown = await Reflect.apply(safeParseAsync, schema, [ + value, + ]); + return typeof result === "object" && + result !== null && + Reflect.get(result, "success") === true + ? { ok: true, data: Reflect.get(result, "data") } + : { ok: false }; + }; +} + +function schemaInput(parse: Parse): Decode { + return async (raw) => { + const parsed = await parse(raw); + // Zod issues can quote caller keys, so the refusal stays opaque. + if (!parsed.ok) throw new DatabasePluginError("INVALID_REQUEST", "read"); + return parsed.data; + }; +} + +function schemaReply(method: OperationMethod, parse: Parse): Respond { + return async (result) => { + const parsed = await parse(result); + if (!parsed.ok) { + throw new DatabasePluginError( + result === null && NULLABLE_METHODS.has(method) + ? "NOT_FOUND" + : "INTERNAL", + "read", + ); + } + if (parsed.data === undefined) return { status: 204 }; + let body: JsonValue; + try { + body = boundedJson(parsed.data); + } catch { + throw new DatabasePluginError("INTERNAL", "read"); + } + return { status: method === "POST" ? 201 : 200, body }; + }; +} + +/** Compile every declaration once; a contract that cannot be honored fails setup. */ +function compileOperations( + schema: Schema, + operations: readonly ValidatedOperation[], +): ReadonlyMap { + const tables = compileCrudTables(schema.$tables); + const registry = new Map(); + for (const operation of operations) { + const table = tables.get(operation.entity); + if (!table) { + throw setupFailed(operation, "names an entity the schema lacks."); + } + const defaults = DEFAULTS[operation.method]; + const decode = operation.input + ? schemaInput(parserOf(operation, operation.input)) + : defaults.input(table); + if (!decode) { + throw setupFailed( + operation, + `needs an explicit input: ${operation.entity} has no default ${operation.method} input.`, + ); + } + registry.set(operation.name, { + name: operation.name, + method: operation.method, + decode, + authorize: operation.authorize, + execute: operation.execute, + reply: operation.output + ? schemaReply(operation.method, parserOf(operation, operation.output)) + : defaults.reply(table), + }); + } + return registry; +} + +/** Express keeps the raw query in the URL; decoding it is the handler's job. */ +function rawQuery(req: Request): string { + const url = req.originalUrl ?? req.url; + const start = url.indexOf("?"); + return start === -1 ? "" : url.slice(start + 1); +} + +/** A GET carries exactly one `input` parameter holding bounded JSON. */ +function queryInput(req: Request): JsonValue { + const query = rawQuery(req); + if (Buffer.byteLength(query, "utf8") > MAX_QUERY_BYTES) { + throw invalidDatabaseInput(["input"], "Input exceeds the maximum size"); + } + const [entry, ...rest] = new URLSearchParams(query); + if (!entry || rest.length > 0 || entry[0] !== "input") { + throw invalidDatabaseInput(["input"], "Expected one JSON input parameter"); + } + let value: unknown; + try { + value = JSON.parse(entry[1]); + } catch { + throw invalidDatabaseInput(["input"], "Input is not valid JSON"); + } + return boundedJson(value); +} + +const parseJsonBody = express.json({ + limit: MAX_OPERATION_BODY_BYTES, + inflate: false, + strict: false, + type: () => true, +}); + +function statusOf(error: unknown): number { + const status = + typeof error === "object" && error !== null + ? Reflect.get(error, "status") + : undefined; + return typeof status === "number" ? status : 500; +} + +/** A write's whole input is one bounded JSON body, read only after authentication. */ +async function bodyInput(req: Request, res: Response): Promise { + if (rawQuery(req) !== "") { + throw invalidDatabaseInput(["query"], "Writes accept no query parameters"); + } + if (req.is("application/json") === false) { + throw new DatabasePluginError("UNSUPPORTED_MEDIA_TYPE", "write"); + } + // Something read this body before the caller was known. + if (req.readableEnded) throw new DatabasePluginError("INTERNAL", "write"); + await new Promise((resolve, reject) => { + parseJsonBody(req, res, (error?: unknown) => + error === undefined + ? resolve() + : reject(databaseErrorFromStatus(statusOf(error), "write")), + ); + }); + return boundedJson(req.body); +} + +type Called = + | { readonly ok: true; readonly value: unknown } + | { readonly ok: false }; + +/** Run one authorization callback; a throw is recorded as a refusal. */ +async function invoke( + operation: CompiledOperation, + callback: (...args: never[]) => unknown, + argument: object, +): Promise { + try { + const value: unknown = await Reflect.apply(callback, undefined, [argument]); + return { ok: true, value }; + } catch (error) { + logger.warn( + "Operation %s refused: its authorization threw %s", + operation.name, + errorName(error), + ); + return { ok: false }; + } +} + +/** Only literal `true` admits; an explicit denial picks 403 or 404. */ +function enforce(called: Called, refusal: "FORBIDDEN" | "NOT_FOUND"): void { + if (called.ok && called.value === true) return; + const deny = + called.ok && typeof called.value === "object" && called.value !== null + ? Reflect.get(called.value, "deny") + : undefined; + throw new DatabasePluginError( + deny === "forbidden" + ? "FORBIDDEN" + : deny === "not-found" + ? "NOT_FOUND" + : refusal, + "read", + ); +} + +/** The category a failure may reveal; field details only from input decoding. */ +function safeFailure( + error: unknown, + stage: Stage, + method: string, +): DatabasePluginError { + if (error instanceof DatabaseValidationError) { + return new DatabasePluginError("VALIDATION_FAILED", "write"); + } + const safe = classifyDatabaseError( + error, + method === "GET" ? "read" : "write", + ); + // Only authentication answers 401, so a client never drops a session otherwise. + if (safe.category === "UNAUTHENTICATED" && stage !== "actor") { + return new DatabasePluginError("INTERNAL", safe.phase); + } + return stage === "input" || !safe.details + ? safe + : new DatabasePluginError(safe.category, safe.phase); +} + +/** + * A write whose answer failed after it ran. The client reports + * `OUTCOME_UNKNOWN` and refreshes, since the change may have committed. + */ +function sendOutcomeUnknown(res: Response): void { + if (res.headersSent) return; + sendJson(res, 500, { + error: "Database write may have completed", + outcome: "unknown", + }); +} + +function sendFailure(res: Response, failure: DatabasePluginError): void { + if (res.headersSent) return; + const details = failure.details?.map((detail) => ({ + path: [...detail.path], + message: detail.message, + })); + sendJson( + res, + failure.statusCode, + details && details.length > 0 + ? { error: failure.clientMessage, details } + : { error: failure.clientMessage }, + ); +} + +/** + * The private HTTP runtime for named operations. Mount it with + * `router.use(OPERATIONS_PATH, handler)` and keep every body parser off the + * mount: the caller is authenticated before the body or query is read. Each + * request resolves its actor once, routes by encoded name and exact method, + * decodes input, loads at most one resource, admits only literal `true`, then + * executes once. No response is cacheable. + */ +export function createOperationHandler( + options: OperationHandlerOptions, +): RequestHandler { + const registry = compileOperations(options.schema, options.operations); + + function route(req: Request): CompiledOperation | undefined { + const segment = /^\/([^/]+)$/.exec(req.path)?.[1]; + // A percent-encoded character takes three bytes, so this bounds decoding. + if (!segment || segment.length > MAX_OPERATION_NAME_LENGTH * 3) { + return undefined; + } + let name: string; + try { + name = decodeURIComponent(segment); + } catch { + return undefined; + } + const operation = registry.get(name); + // Express answers HEAD with a GET route; an operation answers its own method only. + return operation?.method === req.method ? operation : undefined; + } + + async function admit( + operation: CompiledOperation, + base: object, + progress: { stage: Stage }, + ): Promise { + const { authorize } = operation; + progress.stage = "allow"; + if (typeof authorize === "function") { + enforce( + await invoke(operation, authorize, Object.freeze({ ...base })), + "FORBIDDEN", + ); + return undefined; + } + if (authorize.resource === undefined) { + enforce( + await invoke( + operation, + authorize.allow, + Object.freeze({ ...base, resource: undefined }), + ), + "FORBIDDEN", + ); + return undefined; + } + progress.stage = "resource"; + const found = await invoke( + operation, + authorize.resource, + Object.freeze({ ...base }), + ); + // `0`, `false`, and `""` are resources; only null and undefined are missing. + if (!found.ok || found.value === null || found.value === undefined) { + throw new DatabasePluginError("NOT_FOUND", "read"); + } + progress.stage = "allow"; + enforce( + await invoke( + operation, + authorize.allow, + Object.freeze({ ...base, resource: found.value }), + ), + "NOT_FOUND", + ); + return found.value; + } + + async function run( + req: Request, + res: Response, + span: Span, + progress: { stage: Stage }, + ): Promise { + const caller = await options.resolveActor(req); + if (!caller) throw new DatabasePluginError("UNAUTHENTICATED", "read"); + res.setHeader(DATABASE_ACTOR_HEADER, caller.discriminator); + + progress.stage = "origin"; + if (!isSameOriginRequest(req)) { + throw new DatabasePluginError("FORBIDDEN", "read"); + } + + progress.stage = "route"; + const operation = route(req); + if (!operation) throw new DatabasePluginError("NOT_FOUND", "read"); + span.setAttribute("database.operation.name", operation.name); + + progress.stage = "input"; + const input = await operation.decode( + operation.method === "GET" ? queryInput(req) : await bodyInput(req, res), + ); + + progress.stage = "allow"; + const { db, appkit } = options.services(); + const base = { actor: caller.actor, db, appkit, input }; + const resource = await admit(operation, base, progress); + + progress.stage = "execute"; + const result: unknown = await Reflect.apply(operation.execute, undefined, [ + Object.freeze({ ...base, resource }), + ]); + + progress.stage = "output"; + return await operation.reply(result); + } + + return (req, res) => { + res.setHeader("Cache-Control", "no-store"); + return options.telemetry + .startActiveSpan( + "database.operation", + { attributes: { "http.request.method": req.method } }, + async (span) => { + const progress: { stage: Stage } = { stage: "actor" }; + try { + // Lower clients and callbacks run untraced: their spans could + // carry SQL text, row values, tokens, or raw exception messages. + const reply = await context.with( + suppressTracing(context.active()), + () => run(req, res, span, progress), + ); + if (reply.status === 204) sendEmpty(res, 204); + else sendJson(res, reply.status, reply.body); + span.setAttributes({ + "database.operation.outcome": "success", + "http.response.status_code": reply.status, + }); + } catch (error) { + const failure = safeFailure(error, progress.stage, req.method); + // Once a write's execute returned, its changes may have committed: + // anything but the contract's own not-found is an unknown outcome. + const unknown = + progress.stage === "output" && + req.method !== "GET" && + failure.category !== "NOT_FOUND"; + if (unknown) sendOutcomeUnknown(res); + else sendFailure(res, failure); + span.setAttributes({ + "database.operation.outcome": unknown + ? "OUTCOME_UNKNOWN" + : failure.category, + "database.operation.stage": progress.stage, + "http.response.status_code": failure.statusCode, + }); + if (failure.statusCode >= 500) { + span.setStatus({ code: SpanStatusCode.ERROR }); + } + } finally { + span.end(); + } + }, + ) + .catch(() => { + if (!res.headersSent) res.status(500).end(); + }); + }; +} diff --git a/packages/appkit/src/plugins/database/tests/operation-actor.test.ts b/packages/appkit/src/plugins/database/tests/operation-actor.test.ts new file mode 100644 index 000000000..7ae3ce15d --- /dev/null +++ b/packages/appkit/src/plugins/database/tests/operation-actor.test.ts @@ -0,0 +1,61 @@ +import { describe, expect, test } from "vitest"; + +import { selectActorBoundary } from "../operation-actor"; + +describe("selectActorBoundary", () => { + const host = "workspace.cloud.databricks.com"; + + test("a deployed Databricks App verifies forwarded user tokens with its workspace", () => { + expect( + selectActorBoundary(host, { + NODE_ENV: "production", + DATABRICKS_APP_NAME: "notes", + }), + ).toEqual({ kind: "apps", host: `https://${host}` }); + }); + + test("local development needs explicit credentials and keeps local OBO switchable", () => { + const local = selectActorBoundary(`https://${host}/`, { + NODE_ENV: "development", + DATABRICKS_CONFIG_PROFILE: "dev", + }); + expect(local).toMatchObject({ kind: "local", host: `https://${host}` }); + + for (const env of [ + { NODE_ENV: "development" }, + { NODE_ENV: "development", DATABRICKS_TOKEN: " " }, + { + NODE_ENV: "development", + DATABRICKS_TOKEN: "dapi-dev", + APPKIT_DEV_OBO: "false", + }, + ]) { + expect(selectActorBoundary(host, env)).toEqual({ kind: "unsupported" }); + } + }); + + test("any other hosting, or no workspace, authenticates no one", () => { + expect(selectActorBoundary(host, { NODE_ENV: "production" })).toEqual({ + kind: "unsupported", + }); + // Request headers alone never select a boundary. + expect( + selectActorBoundary(host, { + NODE_ENV: "production", + HTTP_X_FORWARDED_ACCESS_TOKEN: "t", + }), + ).toEqual({ kind: "unsupported" }); + expect( + selectActorBoundary(undefined, { + NODE_ENV: "production", + DATABRICKS_APP_NAME: "notes", + }), + ).toEqual({ kind: "unsupported" }); + expect( + selectActorBoundary("http://[bad", { + NODE_ENV: "production", + DATABRICKS_APP_NAME: "notes", + }), + ).toEqual({ kind: "unsupported" }); + }); +}); diff --git a/packages/appkit/src/plugins/database/tests/operation-edges.ts b/packages/appkit/src/plugins/database/tests/operation-edges.ts new file mode 100644 index 000000000..a36ce987f --- /dev/null +++ b/packages/appkit/src/plugins/database/tests/operation-edges.ts @@ -0,0 +1,197 @@ +import http, { type Server } from "node:http"; +import type { AddressInfo } from "node:net"; + +import { context, trace } from "@opentelemetry/api"; +import { suppressTracing } from "@opentelemetry/core"; + +import type { DataPath, Row } from "../../../database/runtime"; + +/** Values no response, log, or span may ever carry. */ +export const SENTINEL = { + token: "token-sentinel-7f3a", + actorId: "actor-sentinel-1001", + email: "sentinel.person@example.com", + input: "input-sentinel-41", + row: "row-sentinel-secret", + error: "exception-sentinel-boom", + reason: "reason-sentinel-denied", +}; + +type Store = Record; + +export const stringify = (value: unknown) => + JSON.stringify(value, (_key, item) => + typeof item === "bigint" ? item.toString() : item, + ); + +/** + * The external database edge: an in-memory DataPath over `tables` that stages + * a transaction and opens the span a pg instrumentation would, with the SQL + * text, the row values, and the raw driver exception on it. Inserting the + * value `"explode"` fails like a constraint violation. + */ +export function memoryDatabase(tables: readonly string[]) { + const committed: Store = Object.fromEntries(tables.map((name) => [name, []])); + const log: string[] = []; + let nextId = 1; + const lower = (statement: string, run: () => T): T => { + const span = trace + .getTracer("@opentelemetry/instrumentation-pg") + .startSpan("pg.query", { attributes: { "db.statement": statement } }); + try { + return run(); + } catch (error) { + span.recordException(error as Error); + throw error; + } finally { + span.end(); + } + }; + const build = (store: Store): DataPath => ({ + select: async (table) => store[table.$name].slice(), + findOne: async (table, value) => + lower(`select * from ${table.$name} where id = ${String(value)}`, () => { + const row = store[table.$name].find((entry) => entry.id === value); + return row ? { ...row } : null; + }), + count: async (table) => store[table.$name].length, + insert: async (table, values) => + lower(`insert ${stringify(values)}`, () => { + if (Object.values(values).includes("explode")) { + throw new Error(`duplicate key value ${SENTINEL.error}`); + } + const row = { id: nextId++, ...values }; + store[table.$name].push(row); + log.push(`insert:${table.$name}`); + return { ...row }; + }), + update: async (table, value, values) => + lower(`update ${stringify(values)}`, () => { + const row = store[table.$name].find((entry) => entry.id === value); + if (!row) return null; + Object.assign(row, values); + return { ...row }; + }), + upsert: async () => { + throw new Error("not used"); + }, + delete: async (table, value) => { + const index = store[table.$name].findIndex((row) => row.id === value); + if (index === -1) return false; + store[table.$name].splice(index, 1); + return true; + }, + raw: async () => [], + transaction: async (callback) => { + log.push("begin"); + const staged: Store = Object.fromEntries( + Object.entries(store).map(([name, rows]) => [ + name, + rows.map((row) => ({ ...row })), + ]), + ); + try { + const result = await callback(build(staged)); + for (const name of Object.keys(store)) store[name] = staged[name]; + log.push("commit"); + return result; + } catch (error) { + log.push("rollback"); + throw error; + } + }, + }); + return { path: build(committed), committed, log }; +} + +/** The external identity edge: a workspace that answers SCIM `Me` per bearer token. */ +export async function startWorkspace() { + const users: Record = { + [`Bearer ${SENTINEL.token}`]: { + id: SENTINEL.actorId, + userName: SENTINEL.email, + active: true, + schemas: ["urn:ietf:params:scim:schemas:core:2.0:User"], + }, + "Bearer token-bob": { id: "bob-2002", userName: "bob@example.com" }, + "Bearer token-sp": { + id: "sp-3003", + schemas: ["urn:ietf:params:scim:schemas:core:2.0:ServicePrincipal"], + }, + }; + const authorizations: string[] = []; + const server = http.createServer((req, res) => { + const authorization = req.headers.authorization ?? ""; + authorizations.push(authorization); + req.resume(); + if (req.method !== "GET" || req.url !== "/api/2.0/preview/scim/v2/Me") { + res.writeHead(404).end(); + return; + } + const user = users[authorization]; + const status = + authorization === "Bearer token-outage" ? 500 : user ? 200 : 401; + res.writeHead(status, { "content-type": "application/json" }); + res.end( + JSON.stringify( + user ?? { error_code: "UNAUTHENTICATED", message: "Bad token" }, + ), + ); + }); + const url = await listen(server); + return { url, authorizations, close: () => close(server) }; +} + +export async function listen(server: Server): Promise { + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + return `http://127.0.0.1:${(server.address() as AddressInfo).port}`; +} + +export async function close(server: Server): Promise { + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); +} + +export interface Reply { + status: number; + headers: http.IncomingHttpHeaders; + body: string; +} + +/** + * Raw HTTP, so tests control Host, Origin, and fetch metadata exactly. The + * harness's own client is untraced: in production the browser is the client. + */ +export function send( + url: string, + options: { method?: string; headers?: Record; body?: string }, +): Promise { + const headers = { ...options.headers }; + if (options.body !== undefined) { + headers["content-length"] = String(Buffer.byteLength(options.body)); + } + return context.with( + suppressTracing(context.active()), + () => + new Promise((resolve, reject) => { + const req = http.request( + url, + { method: options.method ?? "GET", headers }, + (res) => { + let body = ""; + res.setEncoding("utf8"); + res.on("data", (chunk) => (body += chunk)); + res.on("end", () => + resolve({ + status: res.statusCode ?? 0, + headers: res.headers, + body, + }), + ); + }, + ); + req.on("error", reject); + req.end(options.body); + }), + ); +} diff --git a/packages/appkit/src/plugins/database/tests/operation-runtime.test.ts b/packages/appkit/src/plugins/database/tests/operation-runtime.test.ts new file mode 100644 index 000000000..46a13de4a --- /dev/null +++ b/packages/appkit/src/plugins/database/tests/operation-runtime.test.ts @@ -0,0 +1,1380 @@ +import { createHash } from "node:crypto"; +import http, { type Server } from "node:http"; +import { inspect } from "node:util"; + +import { context, SpanKind, trace } from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from "@opentelemetry/sdk-trace-base"; +import { mockServiceContext } from "@tools/test-helpers"; +import express from "express"; +import { + afterAll, + afterEach, + beforeAll, + beforeEach, + describe, + expect, + test, + vi, +} from "vitest"; +import { z } from "zod"; + +import { DatabasePluginError } from "../../../database/errors"; +import { + bigint, + defineSchema, + fk, + id, + text, +} from "../../../database/schema-builder"; +import { TelemetryManager } from "../../../telemetry"; +import type { ITelemetry } from "../../../telemetry"; + +const mocks = vi.hoisted(() => ({ + initializeLakebasePool: vi.fn(), + createDrizzleDb: vi.fn(), + createDrizzleDataPath: vi.fn(), +})); + +vi.mock("../../../connectors/lakebase", () => ({ + initializeLakebasePool: mocks.initializeLakebasePool, +})); +vi.mock("../../../database/runtime/engine/drizzle-data-path", () => ({ + createDrizzleDb: mocks.createDrizzleDb, + createDrizzleDataPath: mocks.createDrizzleDataPath, +})); +vi.mock("../schema-check", () => ({ + assertSchemaMatchesDatabase: async () => undefined, +})); + +import { DatabasePlugin } from "../database"; +import { type ActorResolver, createActorResolver } from "../operation-actor"; +import { validateOperationMetadata } from "../operation-metadata"; +import { createOperationHandler, OPERATIONS_PATH } from "../operation-runtime"; +import { defineOperation, definePolicy } from "../operations"; +import { + close, + listen, + memoryDatabase, + type Reply, + SENTINEL, + send, + startWorkspace, + stringify, +} from "./operation-edges"; + +// The package has no generated registry, so these calls take the untyped JS path. +const define = defineOperation as unknown as ( + definition: Record, +) => unknown; +const policy = definePolicy as unknown as ( + definition: Record, +) => unknown; +type Ctx = { + actor: { id: string; email?: string }; + db: Record; + input: any; + resource?: any; +}; + +const ADA = { id: SENTINEL.actorId, email: SENTINEL.email }; + +const schema = defineSchema((builder) => { + const notes = builder.table("notes", { + id: id(), + title: text().notNull(), + ownerId: text(), + views: bigint(), + secret: text().private(), + }); + const events = builder.table("events", { + id: id(), + noteId: fk(() => notes.id).notNull(), + action: text().notNull(), + }); + const tags = builder.table("tags", { label: text().notNull() }); + const stamps = builder.table("stamps", { id: id() }); + return { notes, events, tags, stamps }; +}); + +const exporter = new InMemorySpanExporter(); +const provider = new BasicTracerProvider({ + spanProcessors: [new SimpleSpanProcessor(exporter)], +}); +let workspace: Awaited>; +let serviceContext: Awaited>; +const servers: Server[] = []; +let logs: string[]; + +beforeAll(async () => { + context.setGlobalContextManager( + new AsyncLocalStorageContextManager().enable(), + ); + trace.setGlobalTracerProvider(provider); + workspace = await startWorkspace(); +}); + +afterAll(async () => { + await workspace.close(); + trace.disable(); + context.disable(); +}); + +beforeEach(async () => { + mocks.initializeLakebasePool.mockReset(); + mocks.createDrizzleDb.mockReset(); + mocks.createDrizzleDataPath.mockReset(); + serviceContext = await mockServiceContext(); + workspace.authorizations.length = 0; + exporter.reset(); + logs = []; + for (const level of ["log", "info", "warn", "error"] as const) { + vi.spyOn(console, level).mockImplementation((...args: unknown[]) => { + logs.push( + args + .map((arg) => + typeof arg === "string" ? arg : inspect(arg, { depth: 6 }), + ) + .join(" "), + ); + }); + } +}); + +afterEach(async () => { + vi.restoreAllMocks(); + serviceContext.restore(); + await Promise.all(servers.splice(0).map(close)); +}); + +/** Boot a real DatabasePlugin over the fake edge and serve operations over real HTTP. */ +async function boot( + operations: unknown[], + options: { resolveActor?: ActorResolver; parseAllBodies?: boolean } = {}, +) { + const database = memoryDatabase(Object.keys(schema.$tables)); + mocks.initializeLakebasePool.mockResolvedValue({ + end: async () => undefined, + }); + mocks.createDrizzleDb.mockReturnValue({}); + mocks.createDrizzleDataPath.mockReturnValue(database.path); + const plugin = new DatabasePlugin({ schema }); + const telemetry: ITelemetry = TelemetryManager.getProvider("database"); + (plugin as unknown as { telemetry: ITelemetry }).telemetry = telemetry; + await plugin.setup(); + + const located = operations.map((value, index) => ({ + file: "server/database/routes/notes.ts", + exportName: `operation${index}`, + value, + })); + const validated = validateOperationMetadata( + located, + new Set(Object.keys(schema.$tables)), + ); + expect(validated.skipped).toEqual([]); + expect(validated.errors).toEqual([]); + + const app = express(); + // The server plugin's global parser must skip the operation namespace. + app.use( + express.json({ + type: (req) => + options.parseAllBodies === true || + !req.url?.startsWith("/api/database/operations"), + }), + ); + const router = express.Router(); + router.use( + OPERATIONS_PATH, + createOperationHandler({ + schema, + operations: validated.operations, + resolveActor: + options.resolveActor ?? + createActorResolver({ kind: "apps", host: workspace.url }), + services: () => { + const db = plugin.exports(); + return { db, appkit: { database: db } }; + }, + telemetry, + }), + ); + app.use("/api/database", router); + const server = http.createServer(app); + servers.push(server); + const url = await listen(server); + + const call = ( + method: string, + name: string, + input?: unknown, + headers: Record = {}, + ): Promise => { + const target = new URL(`/api/database/operations/${name}`, url); + if (method === "GET" || method === "HEAD") { + if (input !== undefined) { + target.searchParams.set("input", JSON.stringify(input)); + } + return send(target.href, { method, headers }); + } + return send(target.href, { + method, + headers: { "content-type": "application/json", ...headers }, + body: input === undefined ? undefined : JSON.stringify(input), + }); + }; + return { url, database, db: () => plugin.exports() as any, call }; +} + +const asAda = { "x-forwarded-access-token": SENTINEL.token }; +const discriminator = (actorId: string) => + createHash("sha256").update(`${workspace.url}\0${actorId}`).digest("hex"); +const json = (reply: Reply) => JSON.parse(reply.body); + +/** A GET whose resource policy reads the note and admits only its owner. */ +function ownedNoteGet( + spies = { lookup: vi.fn(), allow: vi.fn(), execute: vi.fn() }, +) { + return { + spies, + operation: define({ + name: "notes.get", + entity: "notes", + method: "GET", + authorize: policy({ + resource: (ctx: Ctx) => { + spies.lookup(ctx); + return ctx.db.notes.find(ctx.input.id); + }, + allow: (ctx: Ctx) => { + spies.allow(ctx); + return ctx.resource.ownerId === ctx.actor.id; + }, + }), + execute: (ctx: Ctx) => { + spies.execute(ctx); + return { ...ctx.resource, events: [{ id: 9, action: "hidden" }] }; + }, + }), + }; +} + +async function seedNote(db: any, values: Record = {}) { + return db.notes.create({ + title: SENTINEL.row, + ownerId: SENTINEL.actorId, + views: 9007199254740993n, + secret: SENTINEL.row, + ...values, + }); +} + +describe("authentication comes first", () => { + test("answers 401 before reading input, loading, or executing when no verified token arrives", async () => { + const get = ownedNoteGet(); + const create = vi.fn(); + const app = await boot([ + get.operation, + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: () => true, + execute: create, + }), + ]); + const forged = { + "x-forwarded-user": SENTINEL.actorId, + "x-forwarded-email": SENTINEL.email, + }; + + const replies = [ + await send( + `${app.url}/api/database/operations/notes.get?input=%7Bnot-json`, + { + headers: forged, + }, + ), + await send(`${app.url}/api/database/operations/notes.create`, { + method: "POST", + headers: { ...forged, "content-type": "application/json" }, + body: "{not json", + }), + await app.call("GET", "missing.op", { id: 1 }, forged), + await app.call( + "GET", + "notes.get", + { id: 1 }, + { + "x-forwarded-access-token": "token-forged", + }, + ), + await app.call( + "GET", + "notes.get", + { id: 1 }, + { + "x-forwarded-access-token": "token-sp", + }, + ), + ]; + + for (const reply of replies) { + expect(reply.status).toBe(401); + expect(json(reply)).toEqual({ + error: "Database request is not authenticated", + }); + expect(reply.headers["cache-control"]).toBe("no-store"); + expect(reply.headers["x-appkit-database-actor"]).toBeUndefined(); + } + expect(get.spies.lookup).not.toHaveBeenCalled(); + expect(get.spies.execute).not.toHaveBeenCalled(); + expect(create).not.toHaveBeenCalled(); + // Only the two requests that carried a token reached the workspace. + expect(workspace.authorizations).toEqual([ + "Bearer token-forged", + "Bearer token-sp", + ]); + }); + + test("verifies only the forwarded token, never the service principal's credential", async () => { + vi.stubEnv("DATABRICKS_TOKEN", "service-principal-token"); + vi.stubEnv("DATABRICKS_CLIENT_ID", "service-client"); + vi.stubEnv("DATABRICKS_CLIENT_SECRET", "service-secret"); + const get = ownedNoteGet(); + const app = await boot([get.operation]); + const note = await seedNote(app.db()); + + const reply = await app.call("GET", "notes.get", { id: note.id }, asAda); + vi.unstubAllEnvs(); + + expect(reply.status).toBe(200); + expect(workspace.authorizations).toEqual([`Bearer ${SENTINEL.token}`]); + }); + + test("answers 503 without an actor when the workspace cannot verify the caller", async () => { + const get = ownedNoteGet(); + const app = await boot([get.operation]); + + const reply = await app.call( + "GET", + "notes.get", + { id: 1 }, + { + "x-forwarded-access-token": "token-outage", + }, + ); + + expect(reply.status).toBe(503); + expect(reply.headers["x-appkit-database-actor"]).toBeUndefined(); + expect(get.spies.lookup).not.toHaveBeenCalled(); + }); + + test("tags each authenticated response with a stable per-actor discriminator", async () => { + const get = ownedNoteGet(); + const app = await boot([get.operation]); + const note = await seedNote(app.db()); + + const first = await app.call("GET", "notes.get", { id: note.id }, asAda); + const again = await app.call("GET", "notes.get", { id: note.id }, asAda); + const bob = await app.call( + "GET", + "notes.get", + { id: note.id }, + { + "x-forwarded-access-token": "token-bob", + }, + ); + + expect(first.status).toBe(200); + expect(first.headers["x-appkit-database-actor"]).toBe( + discriminator(SENTINEL.actorId), + ); + expect(again.headers["x-appkit-database-actor"]).toBe( + first.headers["x-appkit-database-actor"], + ); + // Bob is authenticated but does not own the note: an opaque 404, his own tag. + expect(bob.status).toBe(404); + expect(bob.headers["x-appkit-database-actor"]).toBe( + discriminator("bob-2002"), + ); + expect(get.spies.execute).toHaveBeenCalledTimes(2); + }); +}); + +describe("routing", () => { + test("an authenticated request reaches only a declared name with its exact method", async () => { + const get = ownedNoteGet(); + const app = await boot([get.operation]); + const note = await seedNote(app.db()); + const path = (suffix: string) => + `${app.url}/api/database/operations/${suffix}?input=${encodeURIComponent(JSON.stringify({ id: note.id }))}`; + + const replies = [ + await app.call("GET", "notes.missing", { id: note.id }, asAda), + await app.call("POST", "notes.get", { id: note.id }, asAda), + await app.call("DELETE", "notes.get", { id: note.id }, asAda), + await app.call("HEAD", "notes.get", { id: note.id }, asAda), + await app.call("GET", "NOTES.GET", { id: note.id }, asAda), + await send(path("notes.get/extra"), { headers: asAda }), + await send(path("%E0%A4%A"), { headers: asAda }), + await send(path(""), { headers: asAda }), + ]; + + expect(replies.map((reply) => reply.status)).toEqual([ + 404, 404, 404, 404, 404, 404, 404, 404, + ]); + expect(get.spies.lookup).not.toHaveBeenCalled(); + expect(get.spies.execute).not.toHaveBeenCalled(); + expect( + (await app.call("GET", "notes.get", { id: note.id }, asAda)).status, + ).toBe(200); + }); +}); + +describe("input", () => { + test("a GET reads exactly one bounded JSON input parameter", async () => { + const get = ownedNoteGet(); + const app = await boot([get.operation]); + const base = `${app.url}/api/database/operations/notes.get`; + let deep: unknown = 1; + for (let depth = 0; depth < 40; depth++) deep = [deep]; + + const replies = await Promise.all( + [ + "", + "?input=1&input=2", + `?input=${encodeURIComponent('{"id":1}')}&other=1`, + `?input=${encodeURIComponent(SENTINEL.input)}`, + `?input=${encodeURIComponent(JSON.stringify({ id: 1, extra: SENTINEL.input }))}`, + `?input=${encodeURIComponent(JSON.stringify({ id: SENTINEL.input }))}`, + `?input=${encodeURIComponent(JSON.stringify({ id: deep }))}`, + `?input=${"a".repeat(9 * 1024)}`, + ].map((query) => send(base + query, { headers: asAda })), + ); + + for (const reply of replies) { + expect(reply.status).toBe(400); + expect(reply.body).not.toContain(SENTINEL.input); + } + expect(get.spies.lookup).not.toHaveBeenCalled(); + }); + + test("a write reads one JSON body of bounded size, only after authentication", async () => { + const execute = vi.fn(() => ({ id: 1, title: "t" })); + const app = await boot([ + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: () => true, + execute, + }), + ]); + const url = `${app.url}/api/database/operations/notes.create`; + + const text = await send(url, { + method: "POST", + headers: { ...asAda, "content-type": "text/plain" }, + body: JSON.stringify({ title: "t" }), + }); + const query = await send(`${url}?title=t`, { + method: "POST", + headers: { ...asAda, "content-type": "application/json" }, + body: JSON.stringify({ title: "t" }), + }); + const malformed = await send(url, { + method: "POST", + headers: { ...asAda, "content-type": "application/json" }, + body: `{"title": "${SENTINEL.input}"`, + }); + const large = await send(url, { + method: "POST", + headers: { ...asAda, "content-type": "application/json" }, + body: JSON.stringify({ title: "x".repeat(1024 * 1024 + 1) }), + }); + + expect([text.status, query.status, malformed.status, large.status]).toEqual( + [415, 400, 400, 413], + ); + expect(malformed.body).not.toContain(SENTINEL.input); + expect(execute).not.toHaveBeenCalled(); + }); + + test("fails closed when a body parser read the body before authentication", async () => { + const execute = vi.fn(() => ({ id: 1, title: "t" })); + const app = await boot( + [ + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: () => true, + execute, + }), + ], + { parseAllBodies: true }, + ); + + const reply = await app.call("POST", "notes.create", { title: "t" }, asAda); + + expect(reply.status).toBe(500); + expect(execute).not.toHaveBeenCalled(); + }); + + test("default inputs accept only the public contract of each method", async () => { + const seen: unknown[] = []; + const execute = (ctx: Ctx) => { + seen.push(ctx.input); + return null; + }; + const app = await boot([ + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: () => true, + execute: (ctx: Ctx) => { + seen.push(ctx.input); + return { id: 1, title: "t" }; + }, + }), + define({ + name: "notes.patch", + entity: "notes", + method: "PATCH", + authorize: () => true, + execute, + }), + define({ + name: "notes.put", + entity: "notes", + method: "PUT", + authorize: () => true, + execute, + }), + ]); + + const missing = await app.call( + "POST", + "notes.create", + { ownerId: "x" }, + asAda, + ); + const privateField = await app.call( + "POST", + "notes.create", + { title: "t", secret: "s" }, + asAda, + ); + const incomplete = await app.call( + "PUT", + "notes.put", + { id: 1, values: { title: "t" } }, + asAda, + ); + const readOnly = await app.call( + "PATCH", + "notes.patch", + { id: 1, changes: { id: 2 } }, + asAda, + ); + const bigintPatch = await app.call( + "PATCH", + "notes.patch", + { id: "1", changes: { views: "9007199254740993" } }, + asAda, + ); + const noChanges = await app.call( + "PATCH", + "notes.patch", + { id: 1, changes: {} }, + asAda, + ); + const notAnObject = await app.call( + "PATCH", + "notes.patch", + { id: 1, changes: "title" }, + asAda, + ); + + expect(missing.status).toBe(400); + expect(json(missing).details).toEqual([ + { path: ["title"], message: "Required" }, + ]); + expect(privateField.status).toBe(400); + expect(JSON.stringify(json(privateField))).not.toContain("secret"); + // Details name the field inside the member it belongs to. + expect([incomplete.status, json(incomplete).details]).toEqual([ + 400, + [{ path: ["values", "ownerId"], message: "Required" }], + ]); + expect([readOnly.status, json(readOnly).details]).toEqual([ + 400, + [{ path: ["changes", "id"], message: "Unknown or read-only field" }], + ]); + expect([noChanges.status, json(noChanges).details]).toEqual([ + 400, + [{ path: ["changes"], message: "Expected at least one change" }], + ]); + expect([notAnObject.status, json(notAnObject).details]).toEqual([ + 400, + [{ path: ["changes"], message: "Expected a JSON object" }], + ]); + // A missing row is a 404 even though the input decoded. + expect(bigintPatch.status).toBe(404); + expect(seen).toEqual([{ id: 1, changes: { views: 9007199254740993n } }]); + }); +}); + +describe("authorization", () => { + test("a policy without a resource admits only literal true", async () => { + const decisions: unknown[] = [ + false, + { deny: "forbidden", reason: SENTINEL.reason }, + { deny: "not-found", reason: SENTINEL.reason }, + "true", + 1, + {}, + Promise.resolve(false), + new Error(SENTINEL.error), + ]; + let next: unknown; + const execute = vi.fn(() => ({ id: 1, title: "t" })); + const app = await boot([ + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: () => { + if (next instanceof Error) throw next; + return next; + }, + execute, + }), + ]); + + const statuses: number[] = []; + for (const decision of decisions) { + next = decision; + const reply = await app.call( + "POST", + "notes.create", + { title: "t" }, + asAda, + ); + statuses.push(reply.status); + expect(reply.body).not.toContain(SENTINEL.reason); + expect(reply.body).not.toContain(SENTINEL.error); + } + expect(statuses).toEqual([403, 403, 404, 403, 403, 403, 403, 403]); + expect(execute).not.toHaveBeenCalled(); + + next = Promise.resolve(true); + expect( + (await app.call("POST", "notes.create", { title: "t" }, asAda)).status, + ).toBe(201); + expect(execute).toHaveBeenCalledTimes(1); + expect(logs.join("\n")).not.toContain(SENTINEL.reason); + expect(logs.join("\n")).not.toContain(SENTINEL.error); + }); + + test("a resource policy loads once, refuses a missing resource, and shares the admitted value", async () => { + let resource: unknown; + const lookup = vi.fn(() => resource); + const allow = vi.fn((ctx: Ctx) => ctx.resource !== "refuse"); + const execute = vi.fn((ctx: Ctx) => ({ + id: 1, + title: String(ctx.resource), + })); + const app = await boot([ + define({ + name: "notes.check", + entity: "notes", + method: "PATCH", + input: z.object({ id: z.number() }), + output: z.object({ id: z.number(), title: z.string() }), + authorize: policy({ resource: lookup, allow }), + execute, + }), + ]); + const call = () => app.call("PATCH", "notes.check", { id: 1 }, asAda); + + const refused: number[] = []; + for (const value of [null, undefined, "refuse"]) { + resource = value; + refused.push((await call()).status); + } + lookup.mockImplementationOnce(() => { + throw new Error(SENTINEL.error); + }); + refused.push((await call()).status); + expect(refused).toEqual([404, 404, 404, 404]); + expect(execute).not.toHaveBeenCalled(); + expect(allow).toHaveBeenCalledTimes(1); + + lookup.mockClear(); + allow.mockClear(); + for (const value of [0, false]) { + resource = value; + const reply = await call(); + expect(reply.status).toBe(200); + expect(json(reply)).toEqual({ id: 1, title: String(value) }); + } + expect(lookup).toHaveBeenCalledTimes(2); + expect(allow.mock.calls.map(([ctx]) => ctx.resource)).toEqual([0, false]); + expect(execute.mock.calls.map(([ctx]) => ctx.resource)).toEqual([0, false]); + }); + + test("callbacks receive the verified actor and services, never the request or its token", async () => { + const get = ownedNoteGet(); + const app = await boot([get.operation]); + const note = await seedNote(app.db()); + + await app.call("GET", "notes.get", { id: note.id }, asAda); + + const [lookupCtx] = get.spies.lookup.mock.calls[0]; + const [executeCtx] = get.spies.execute.mock.calls[0]; + expect(Object.keys(lookupCtx).sort()).toEqual([ + "actor", + "appkit", + "db", + "input", + ]); + expect(Object.keys(executeCtx).sort()).toEqual([ + "actor", + "appkit", + "db", + "input", + "resource", + ]); + expect(executeCtx.actor).toEqual(ADA); + expect(Object.isFrozen(executeCtx.actor)).toBe(true); + expect(executeCtx.resource).toBe(get.spies.allow.mock.calls[0][0].resource); + expect( + stringify({ ...executeCtx, db: undefined, appkit: undefined }), + ).not.toContain(SENTINEL.token); + }); +}); + +describe("execution and output", () => { + test("default replies project public columns and answer missing rows with 404", async () => { + const owned = ( + method: string, + name: string, + execute: (ctx: Ctx) => unknown, + ) => + define({ name, entity: "notes", method, authorize: () => true, execute }); + const get = ownedNoteGet(); + const app = await boot([ + get.operation, + owned("POST", "notes.create", ({ db, input, actor }) => + db.notes.create({ ...input, ownerId: actor.id, secret: SENTINEL.row }), + ), + owned("PATCH", "notes.patch", ({ db, input }) => + db.notes.update(input.id, input.changes), + ), + owned("PUT", "notes.put", ({ db, input }) => + db.notes.update(input.id, input.values), + ), + owned("DELETE", "notes.delete", ({ db, input }) => + db.notes.delete(input.id), + ), + ]); + + const created = await app.call( + "POST", + "notes.create", + { title: "first", views: "9007199254740993" }, + asAda, + ); + expect(created.status).toBe(201); + const row = json(created); + expect(row).toEqual({ + id: row.id, + title: "first", + ownerId: SENTINEL.actorId, + views: "9007199254740993", + }); + + const read = await app.call("GET", "notes.get", { id: row.id }, asAda); + expect(json(read)).toEqual(row); + expect(read.body).not.toContain("hidden"); + + const patched = await app.call( + "PATCH", + "notes.patch", + { id: row.id, changes: { title: "second" } }, + asAda, + ); + expect([patched.status, json(patched).title]).toEqual([200, "second"]); + const put = await app.call( + "PUT", + "notes.put", + { id: row.id, values: { title: "third", ownerId: null, views: null } }, + asAda, + ); + expect([put.status, json(put).title, json(put).views]).toEqual([ + 200, + "third", + null, + ]); + expect( + ( + await app.call( + "PATCH", + "notes.patch", + { id: 999, changes: { title: "x" } }, + asAda, + ) + ).status, + ).toBe(404); + + const deleted = await app.call( + "DELETE", + "notes.delete", + { id: row.id }, + asAda, + ); + expect([deleted.status, deleted.body]).toEqual([204, ""]); + expect(deleted.headers["cache-control"]).toBe("no-store"); + expect( + (await app.call("DELETE", "notes.delete", { id: row.id }, asAda)).status, + ).toBe(404); + expect(app.database.committed.notes).toEqual([]); + }); + + test("custom schemas decode input and validate the reply, with null only where the output accepts it", async () => { + const app = await boot([ + define({ + name: "notes.search", + entity: "notes", + method: "GET", + input: z.object({ + term: z.string().transform((term) => term.toUpperCase()), + }), + output: z.object({ term: z.string() }).nullable(), + authorize: () => true, + execute: ({ input }: Ctx) => + input.term === "NONE" ? null : { term: input.term }, + }), + define({ + name: "notes.summary", + entity: "notes", + method: "GET", + input: z.object({}), + output: z.object({ count: z.number() }), + authorize: () => true, + execute: () => null, + }), + define({ + name: "notes.broken", + entity: "notes", + method: "POST", + input: z.object({}), + output: z.object({ count: z.number() }), + authorize: () => true, + execute: () => ({ count: SENTINEL.row }), + }), + define({ + name: "notes.purge", + entity: "notes", + method: "DELETE", + input: z.object({}), + output: z.void(), + authorize: () => true, + execute: () => undefined, + }), + ]); + + const found = await app.call("GET", "notes.search", { term: "abc" }, asAda); + const none = await app.call("GET", "notes.search", { term: "none" }, asAda); + const invalid = await app.call("GET", "notes.search", { term: 1 }, asAda); + const missing = await app.call("GET", "notes.summary", {}, asAda); + const broken = await app.call("POST", "notes.broken", {}, asAda); + const purged = await app.call("DELETE", "notes.purge", {}, asAda); + + expect([found.status, json(found)]).toEqual([200, { term: "ABC" }]); + expect([none.status, none.body]).toEqual([200, "null"]); + expect(invalid.status).toBe(400); + expect(missing.status).toBe(404); + // The write ran before its answer failed, so it may have committed. + expect([broken.status, json(broken)]).toEqual([ + 500, + { error: "Database write may have completed", outcome: "unknown" }, + ]); + expect([purged.status, purged.body]).toEqual([204, ""]); + }); + + test("a write whose answer fails after it ran reports an unknown outcome", async () => { + const app = await boot([ + define({ + name: "notes.rename", + entity: "notes", + method: "PATCH", + authorize: () => true, + execute: () => null, + }), + define({ + name: "notes.drop", + entity: "notes", + method: "DELETE", + authorize: () => true, + execute: () => "deleted", + }), + define({ + name: "notes.peek", + entity: "notes", + method: "GET", + input: z.object({}), + output: z.object({ count: z.number() }), + authorize: () => true, + execute: () => ({ count: SENTINEL.row }), + }), + ]); + + const missing = await app.call( + "PATCH", + "notes.rename", + { id: 1, changes: { title: "t" } }, + asAda, + ); + const malformed = await app.call("DELETE", "notes.drop", { id: 1 }, asAda); + const read = await app.call("GET", "notes.peek", {}, asAda); + + // A missing row is the contract's definitive answer, not an unknown one. + expect([missing.status, json(missing)]).toEqual([ + 404, + { error: "Database record not found" }, + ]); + expect([malformed.status, json(malformed)]).toEqual([ + 500, + { error: "Database write may have completed", outcome: "unknown" }, + ]); + // A read changed nothing, so its failure stays definitive. + expect([read.status, json(read)]).toEqual([ + 500, + { error: "Database operation failed" }, + ]); + }); + + test("asynchronous refinements and transforms run on input and output", async () => { + const execute = vi.fn((ctx: Ctx) => ({ + title: ` ${String(Reflect.get(Object(ctx.input), "title"))} `, + })); + const app = await boot([ + define({ + name: "notes.checked", + entity: "notes", + method: "POST", + input: z.object({ + title: z.string().refine(async (title) => title !== "taken"), + }), + output: z.object({ + title: z.string().transform(async (title) => title.trim()), + }), + authorize: () => true, + execute, + }), + ]); + + const created = await app.call( + "POST", + "notes.checked", + { title: "t" }, + asAda, + ); + const refused = await app.call( + "POST", + "notes.checked", + { title: "taken" }, + asAda, + ); + + expect([created.status, json(created)]).toEqual([201, { title: "t" }]); + expect(refused.status).toBe(400); + expect(execute).toHaveBeenCalledTimes(1); + }); + + test("execution failures stay opaque and a write runs once", async () => { + let failure: unknown; + const execute = vi.fn(() => { + throw failure; + }); + const app = await boot([ + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: () => true, + execute, + }), + ]); + + const replies: Reply[] = []; + for (const error of [ + new Error(SENTINEL.error), + new DatabasePluginError("CONFLICT", "write"), + new DatabasePluginError("TRANSIENT", "write"), + new DatabasePluginError("UNAUTHENTICATED", "write"), + ]) { + failure = error; + replies.push( + await app.call("POST", "notes.create", { title: "t" }, asAda), + ); + } + + expect(replies.map((reply) => reply.status)).toEqual([500, 409, 503, 500]); + expect(json(replies[0])).toEqual({ error: "Database operation failed" }); + expect(execute).toHaveBeenCalledTimes(4); + expect(logs.join("\n")).not.toContain(SENTINEL.error); + }); + + test("a note and its event commit together or not at all", async () => { + const app = await boot([ + define({ + name: "notes.publish", + entity: "notes", + method: "POST", + authorize: () => true, + execute: ({ db, input, actor }: Ctx) => + db.transaction(async (tx: any) => { + const note = await tx.notes.create({ ...input, ownerId: actor.id }); + await tx.events.create({ + noteId: note.id, + action: input.title === "fail" ? "explode" : "created", + }); + return note; + }), + }), + ]); + + const ok = await app.call( + "POST", + "notes.publish", + { title: "kept" }, + asAda, + ); + const failed = await app.call( + "POST", + "notes.publish", + { title: "fail" }, + asAda, + ); + + expect(ok.status).toBe(201); + expect(failed.status).toBe(500); + expect(app.database.committed.notes.map((note) => note.title)).toEqual([ + "kept", + ]); + expect(app.database.committed.events).toHaveLength(1); + expect(app.database.log).toEqual([ + "begin", + "insert:notes", + "insert:events", + "commit", + "begin", + "insert:notes", + "rollback", + ]); + }); +}); + +describe("browser origin", () => { + test("refuses cross-site and same-site browser requests before reading input", async () => { + const execute = vi.fn(() => ({ id: 1, title: "t" })); + const get = ownedNoteGet(); + const app = await boot([ + get.operation, + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: () => true, + execute, + }), + ]); + const note = await seedNote(app.db()); + const host = new URL(app.url).host; + const write = (headers: Record) => + app.call( + "POST", + "notes.create", + { title: "t" }, + { ...asAda, ...headers }, + ); + + const refused = [ + await write({ "sec-fetch-site": "cross-site" }), + await write({ "sec-fetch-site": "same-site" }), + await write({ "sec-fetch-site": "none" }), + await write({ origin: "https://attacker.example" }), + await write({ origin: "null" }), + await app.call( + "GET", + "notes.get", + { id: note.id }, + { ...asAda, "sec-fetch-site": "cross-site" }, + ), + ]; + expect(refused.map((reply) => reply.status)).toEqual([ + 403, 403, 403, 403, 403, 403, + ]); + expect(execute).not.toHaveBeenCalled(); + expect(get.spies.lookup).not.toHaveBeenCalled(); + + expect((await write({ "sec-fetch-site": "same-origin" })).status).toBe(201); + expect((await write({ origin: `http://${host}` })).status).toBe(201); + expect( + ( + await app.call( + "GET", + "notes.get", + { id: note.id }, + { ...asAda, "sec-fetch-site": "none" }, + ) + ).status, + ).toBe(200); + }); +}); + +describe("telemetry", () => { + test("operation spans carry bounded metadata and lower spans are suppressed", async () => { + const get = ownedNoteGet(); + const app = await boot([ + get.operation, + define({ + name: "notes.create", + entity: "notes", + method: "POST", + authorize: policy({ + allow: () => ({ deny: "forbidden", reason: SENTINEL.reason }), + }), + execute: () => null, + }), + define({ + name: "notes.explode", + entity: "notes", + method: "POST", + authorize: () => true, + execute: ({ db }: Ctx) => + db.notes.create({ title: "explode", secret: SENTINEL.row }), + }), + ]); + const note = await seedNote(app.db()); + // The control: the same lower span outside an operation is exported. + expect( + stringify(exporter.getFinishedSpans().map((span) => span.attributes)), + ).toContain(SENTINEL.row); + exporter.reset(); + + await app.call("GET", "notes.get", { id: note.id }, asAda); + await app.call("GET", "notes.get", { id: SENTINEL.input }, asAda); + await app.call("POST", "notes.create", { title: SENTINEL.input }, asAda); + await app.call("POST", "notes.explode", { title: SENTINEL.input }, asAda); + await app.call( + "GET", + "notes.get", + { id: 1 }, + { "x-forwarded-access-token": "token-forged" }, + ); + + // The fake workspace runs in-process; its server spans are the workspace's, not the app's. + const workspaceHost = new URL(workspace.url).host; + const spans = exporter + .getFinishedSpans() + .filter((span) => span.attributes["http.host"] !== workspaceHost); + const operationSpans = spans.filter( + (span) => span.name === "database.operation", + ); + const serverSpans = spans.filter((span) => span.kind === SpanKind.SERVER); + // Only the request span and the operation span: every lower span was suppressed. + expect(spans).toHaveLength(10); + expect(serverSpans.map((span) => span.name)).toEqual([ + "GET /api/database/operations/notes.get", + "GET /api/database/operations/notes.get", + "POST /api/database/operations/notes.create", + "POST /api/database/operations/notes.explode", + "GET /api/database/operations/notes.get", + ]); + expect(serverSpans[1].attributes["http.target"]).toBe( + "/api/database/operations/notes.get", + ); + expect(serverSpans[1].attributes["http.url"]).toBe( + `${app.url}/api/database/operations/notes.get`, + ); + expect(operationSpans.map((span) => span.attributes)).toEqual([ + expect.objectContaining({ + "database.operation.name": "notes.get", + "database.operation.outcome": "success", + "http.response.status_code": 200, + }), + expect.objectContaining({ + "database.operation.outcome": "INVALID_REQUEST", + "database.operation.stage": "input", + }), + expect.objectContaining({ + "database.operation.name": "notes.create", + "database.operation.outcome": "FORBIDDEN", + "database.operation.stage": "allow", + }), + expect.objectContaining({ + "database.operation.name": "notes.explode", + "database.operation.outcome": "INTERNAL", + "database.operation.stage": "execute", + }), + expect.objectContaining({ + "database.operation.outcome": "UNAUTHENTICATED", + "database.operation.stage": "actor", + }), + ]); + const exported = JSON.stringify( + spans.map((span) => ({ + name: span.name, + attributes: span.attributes, + events: span.events, + status: span.status, + })), + ); + for (const value of Object.values(SENTINEL)) { + expect(exported).not.toContain(value); + expect(logs.join("\n")).not.toContain(value); + } + expect(exported).not.toContain(discriminator(SENTINEL.actorId)); + }); +}); + +describe("local development boundary", () => { + test("serves only the signed-in developer, on a same-origin loopback request", async () => { + const identity = vi.fn(async () => ({ + token: "local-token", + userId: "dev-user-1", + email: "dev@example.com", + })); + const execute = vi.fn((ctx: Ctx) => ({ term: ctx.actor.id })); + const app = await boot( + [ + define({ + name: "me.get", + entity: "notes", + method: "GET", + input: z.object({}), + output: z.object({ term: z.string() }), + authorize: () => true, + execute, + }), + ], + { + resolveActor: createActorResolver({ + kind: "local", + host: "https://dev.example", + identity, + }), + }, + ); + const target = `${app.url}/api/database/operations/me.get?input=%7B%7D`; + const host = new URL(app.url).host; + + const ok = await send(target, { + headers: { + "x-forwarded-access-token": SENTINEL.token, + "x-forwarded-user": "someone-else", + }, + }); + const foreignHost = await send(target, { + headers: { host: `localhost.attacker.example:${new URL(app.url).port}` }, + }); + const foreignOrigin = await send(target, { + headers: { host, origin: "http://evil.example" }, + }); + identity.mockRejectedValueOnce(new Error(SENTINEL.error)); + const signedOut = await send(target, { headers: { host } }); + + expect([ok.status, json(ok)]).toEqual([200, { term: "dev-user-1" }]); + expect(ok.headers["x-appkit-database-actor"]).toBe( + createHash("sha256") + .update("https://dev.example\0dev-user-1") + .digest("hex"), + ); + expect([ + foreignHost.status, + foreignOrigin.status, + signedOut.status, + ]).toEqual([401, 401, 401]); + expect(execute).toHaveBeenCalledTimes(1); + expect(workspace.authorizations).toEqual([]); + expect(logs.join("\n")).not.toContain(SENTINEL.error); + }); + + test("an unsupported deployment authenticates no one", async () => { + const get = ownedNoteGet(); + const app = await boot([get.operation], { + resolveActor: createActorResolver({ kind: "unsupported" }), + }); + + const reply = await app.call("GET", "notes.get", { id: 1 }, asAda); + + expect(reply.status).toBe(401); + expect(workspace.authorizations).toEqual([]); + expect(get.spies.lookup).not.toHaveBeenCalled(); + }); +}); + +describe("setup", () => { + const handler = + ( + operation: Record, + extra: Record[] = [], + ) => + () => { + const { operations } = validateOperationMetadata( + [operation, ...extra].map((value, index) => ({ + file: "routes.ts", + exportName: `op${index}`, + value: define(value), + })), + new Set(Object.keys(schema.$tables)), + ); + return createOperationHandler({ + schema, + operations, + resolveActor: createActorResolver({ kind: "unsupported" }), + services: () => { + throw new Error("unused"); + }, + telemetry: TelemetryManager.getProvider("database"), + }); + }; + const base = { authorize: () => true, execute: () => null }; + + test("refuses a default contract the table cannot honor", () => { + expect( + handler({ ...base, name: "tags.get", entity: "tags", method: "GET" }), + ).toThrow(/routes\.ts#op0 needs an explicit input/); + expect( + handler({ + ...base, + name: "stamps.patch", + entity: "stamps", + method: "PATCH", + }), + ).toThrow(/needs an explicit input/); + expect( + handler({ ...base, name: "stamps.put", entity: "stamps", method: "PUT" }), + ).toThrow(/needs an explicit input/); + expect( + handler({ + ...base, + name: "tags.find", + entity: "tags", + method: "GET", + input: z.object({ label: z.string() }), + }), + ).not.toThrow(); + }); +}); diff --git a/packages/appkit/src/plugins/server/dev-obo-middleware.ts b/packages/appkit/src/plugins/server/dev-obo-middleware.ts index 4647ba0b2..9ce1e1055 100644 --- a/packages/appkit/src/plugins/server/dev-obo-middleware.ts +++ b/packages/appkit/src/plugins/server/dev-obo-middleware.ts @@ -10,7 +10,8 @@ const logger = createLogger("server:dev-obo"); const loopbackHosts = new Set(["localhost", "127.0.0.1", "[::1]"]); const loopbackPeers = new Set(["127.0.0.1", "::1", "::ffff:127.0.0.1"]); -function isLocalRequest(req: Request): boolean { +/** A same-origin request to this loopback server, judged from the socket and browser metadata. */ +export function isLocalRequest(req: Request): boolean { // Check the socket, not Express's proxy-aware IP or forwarded headers. if (!loopbackPeers.has(req.socket.remoteAddress ?? "")) return false; try { diff --git a/packages/appkit/src/telemetry/instrumentations.ts b/packages/appkit/src/telemetry/instrumentations.ts index 08f8e71b5..3d383db97 100644 --- a/packages/appkit/src/telemetry/instrumentations.ts +++ b/packages/appkit/src/telemetry/instrumentations.ts @@ -3,6 +3,35 @@ import { ExpressInstrumentation } from "@opentelemetry/instrumentation-express"; import { HttpInstrumentation } from "@opentelemetry/instrumentation-http"; import { shouldIgnoreRequest } from "../utils/path-exclusions"; +import { routedPath } from "../utils/routed-path"; + +/** Named database operations carry their GET input in the query string. */ +const OPERATIONS_MOUNT = "/api/database/operations"; + +/** + * Keep an operation request's server span to its path, under either HTTP + * semantic convention. Express routes case-insensitively and serves the mount + * itself, so the match does too. Returns the path, or undefined for any other + * request. + */ +function redactOperationQuery(span: any, request: any): string | undefined { + const original: unknown = request.originalUrl; + if (typeof original !== "string") return undefined; + const path = routedPath(original); + const lower = path.toLowerCase(); + if (lower !== OPERATIONS_MOUNT && !lower.startsWith(`${OPERATIONS_MOUNT}/`)) { + return undefined; + } + const attributes = span.attributes ?? {}; + for (const key of ["http.url", "http.target", "url.full"]) { + const value = attributes[key]; + if (typeof value === "string") span.setAttribute(key, value.split("?")[0]); + } + if (typeof attributes["url.query"] === "string") { + span.setAttribute("url.query", "REDACTED"); + } + return path; +} /** * Registry of pre-configured instrumentations for common use cases. @@ -18,8 +47,11 @@ export const instrumentations: Record = { applyCustomAttributesOnSpan(span: any, request: any) { let spanName: string | null = null; + const operationPath = redactOperationQuery(span, request); - if (request.route) { + if (operationPath) { + spanName = `${request.method} ${operationPath}`; + } else if (request.route) { const baseUrl = request.baseUrl || ""; const url = request.url?.split("?")[0] || ""; const fullPath = baseUrl + url; diff --git a/packages/appkit/src/utils/routed-path.ts b/packages/appkit/src/utils/routed-path.ts new file mode 100644 index 000000000..f76d9596a --- /dev/null +++ b/packages/appkit/src/utils/routed-path.ts @@ -0,0 +1,11 @@ +/** + * The path Express routes a request target by: without its query, and without + * the scheme and authority of an absolute-form target, which a server must + * accept (RFC 9112, section 3.2.2). + */ +export function routedPath(target: string): string { + const path = target.split("?")[0]; + return path.startsWith("/") + ? path + : path.replace(/^[a-z][a-z\d+.-]*:\/\/[^/]*/i, ""); +} diff --git a/packages/appkit/src/utils/tests/routed-path.test.ts b/packages/appkit/src/utils/tests/routed-path.test.ts new file mode 100644 index 000000000..091818d64 --- /dev/null +++ b/packages/appkit/src/utils/tests/routed-path.test.ts @@ -0,0 +1,22 @@ +import { describe, expect, test } from "vitest"; + +import { routedPath } from "../routed-path"; + +describe("routedPath", () => { + test("drops the query of an origin-form target", () => { + expect(routedPath("/api/database/operations/notes.get?input=%7B%7D")).toBe( + "/api/database/operations/notes.get", + ); + }); + + test("drops the scheme and authority of an absolute-form target", () => { + expect( + routedPath("http://app.example.com:8000/API/Database/operations/x?a=1"), + ).toBe("/API/Database/operations/x"); + expect(routedPath("HTTPS://host/api")).toBe("/api"); + }); + + test("answers an empty path for a target without one", () => { + expect(routedPath("http://host")).toBe(""); + }); +}); diff --git a/packages/shared/src/database/errors.ts b/packages/shared/src/database/errors.ts index 5a109e94f..d2a441cc9 100644 --- a/packages/shared/src/database/errors.ts +++ b/packages/shared/src/database/errors.ts @@ -1,5 +1,6 @@ /** Stable failure category a generated database route answers with. */ export type DatabaseErrorCategory = + | "UNAUTHENTICATED" | "INVALID_REQUEST" | "VALIDATION_FAILED" | "NOT_FOUND" @@ -23,6 +24,7 @@ export interface DatabaseErrorDetail { */ const categoryByStatus: Readonly> = { 400: "INVALID_REQUEST", + 401: "UNAUTHENTICATED", 403: "FORBIDDEN", 404: "NOT_FOUND", 409: "CONFLICT", diff --git a/packages/shared/src/database/query-codec.test.ts b/packages/shared/src/database/query-codec.test.ts index 701e76c36..89f8d7448 100644 --- a/packages/shared/src/database/query-codec.test.ts +++ b/packages/shared/src/database/query-codec.test.ts @@ -157,11 +157,12 @@ describe("encodeDatabaseRecordQuery", () => { describe("databaseErrorCategoryForStatus", () => { it("reads every generated status back into its stable category", () => { expect( - [400, 403, 404, 409, 413, 415, 422, 503].map( + [400, 401, 403, 404, 409, 413, 415, 422, 503].map( databaseErrorCategoryForStatus, ), ).toEqual([ "INVALID_REQUEST", + "UNAUTHENTICATED", "FORBIDDEN", "NOT_FOUND", "CONFLICT", @@ -175,6 +176,5 @@ describe("databaseErrorCategoryForStatus", () => { it("treats any other status as INTERNAL", () => { expect(databaseErrorCategoryForStatus(500)).toBe("INTERNAL"); expect(databaseErrorCategoryForStatus(502)).toBe("INTERNAL"); - expect(databaseErrorCategoryForStatus(401)).toBe("INTERNAL"); }); }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index acead9e25..bfcf9ffe4 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -347,6 +347,9 @@ importers: '@opentelemetry/auto-instrumentations-node': specifier: 0.77.0 version: 0.77.0(@opentelemetry/api@1.9.0)(@opentelemetry/core@2.8.0(@opentelemetry/api@1.9.0)) + '@opentelemetry/core': + specifier: 2.8.0 + version: 2.8.0(@opentelemetry/api@1.9.0) '@opentelemetry/exporter-logs-otlp-proto': specifier: 0.219.0 version: 0.219.0(@opentelemetry/api@1.9.0)