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)