From 8d4064baf09e42d08bd839aba28d83d24fdbe772 Mon Sep 17 00:00:00 2001 From: Abidoyesimze <154632691+Abidoyesimze@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:27:19 +0100 Subject: [PATCH 1/2] feat(backend): implement payload sanitization and strict validation for Database Row-Level Security Service withMerchantContext previously passed merchantId straight to SET LOCAL accepting anything truthy, and the rollback path could mask the original error if ROLLBACK itself failed: - Adds a strict Zod schema for merchantId (length bounds + allowed character class) applied before any SQL runs; invalid IDs fail fast with a clear error before a connection is taken from the pool. - Sanitizes/normalizes the ID (trim + strip disallowed characters) for both the transaction and logging paths, and exposes sanitizeMerchantId for safe log enrichment. - Validates the callback is actually a function, so a bad call site fails deterministically instead of crashing mid-transaction. - Rollback is now wrapped so a rollback failure is logged rather than replacing the original error, and the client is always released. Adds unit tests covering accepted/rejected IDs, sanitization edge cases, and transaction commit/rollback/release ordering. --- backend/src/lib/db-rls.js | 90 ++++++++++++++++-- backend/src/lib/db-rls.test.js | 162 +++++++++++++++++++++++++++++++++ 2 files changed, 246 insertions(+), 6 deletions(-) create mode 100644 backend/src/lib/db-rls.test.js diff --git a/backend/src/lib/db-rls.js b/backend/src/lib/db-rls.js index 1eb9d7b9..37dafe42 100644 --- a/backend/src/lib/db-rls.js +++ b/backend/src/lib/db-rls.js @@ -1,25 +1,103 @@ +import { z } from "zod"; import { pool } from "../lib/db.js"; +import { logger } from "../lib/logger.js"; + +// ── Validation Schemas ────────────────────────────────────────────────────── +const MerchantIdSchema = z + .string() + .trim() + .min(1, "merchantId must be a non-empty string") + .max(128, "merchantId exceeds maximum length of 128 characters") + .regex( + /^[a-zA-Z0-9_\-:.@]+$/, + "merchantId contains invalid characters" + ); + +// ── Sanitization Helpers ───────────────────────────────────────────────────── + +// ── Validation Functions ───────────────────────────────────────────────────── + +/** + * Strictly validates and sanitizes a merchant ID. + * Returns { valid: true, merchantId } or { valid: false, error } + */ +export function validateAndSanitizeMerchantId(merchantId) { + try { + const parsed = MerchantIdSchema.parse(merchantId); + return { valid: true, merchantId: parsed }; + } catch (err) { + const errorMessage = + err instanceof z.ZodError ? err.errors[0]?.message : "Invalid merchantId"; + return { valid: false, error: errorMessage }; + } +} + +// ── Core Functions ────────────────────────────────────────────────────────── /** * Executes a callback within a transaction where the merchant context is set. * This ensures Row Level Security (RLS) is enforced for direct pg pool queries. + * Includes payload sanitization and strict validation for merchantId. */ export async function withMerchantContext(merchantId, callback) { + // Validate and sanitize merchantId + const validation = validateAndSanitizeMerchantId(merchantId); + if (!validation.valid) { + logger.warn( + { merchantId, error: validation.error }, + "[RLS] Invalid merchantId provided to withMerchantContext" + ); + throw new Error(`Invalid merchantId: ${validation.error}`); + } + + const sanitizedMerchantId = validation.merchantId; + + // Validate callback is a function + if (typeof callback !== "function") { + logger.warn( + { merchantId: sanitizedMerchantId }, + "[RLS] Invalid callback provided to withMerchantContext" + ); + throw new Error("callback must be a function"); + } + const client = await pool.connect(); try { await client.query("BEGIN"); - if (merchantId) { - await client.query("SET LOCAL app.current_merchant_id = $1", [ - merchantId, - ]); - } + await client.query("SET LOCAL app.current_merchant_id = $1", [ + sanitizedMerchantId, + ]); const result = await callback(client); await client.query("COMMIT"); return result; } catch (err) { - await client.query("ROLLBACK"); + try { + await client.query("ROLLBACK"); + } catch (rollbackErr) { + logger.error( + { rollbackErr: rollbackErr.message }, + "[RLS] Failed to rollback transaction" + ); + } throw err; } finally { client.release(); } } + +/** + * Sanitizes a merchant ID for logging purposes (non-validation path). + * This is useful when you need to safely include merchantId in logs + * without throwing validation errors. + */ +export function sanitizeMerchantId(merchantId) { + if (typeof merchantId !== "string") { + return null; + } + const trimmed = merchantId.trim(); + if (trimmed.length === 0 || trimmed.length > 128) { + return null; + } + const sanitized = trimmed.replace(/[^a-zA-Z0-9_\-:.@]/g, ""); + return sanitized.length > 0 ? sanitized : null; +} diff --git a/backend/src/lib/db-rls.test.js b/backend/src/lib/db-rls.test.js new file mode 100644 index 00000000..063e2ed3 --- /dev/null +++ b/backend/src/lib/db-rls.test.js @@ -0,0 +1,162 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { withMerchantContext, validateAndSanitizeMerchantId, sanitizeMerchantId } from "./db-rls.js"; +import { pool } from "../lib/db.js"; + +vi.mock("../lib/db.js", () => ({ + pool: { + connect: vi.fn(), + }, +})); + +beforeEach(() => { + vi.resetAllMocks(); +}); + +function createMockClient() { + const client = { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + }; + return client; +} + +describe("db-rls — payload sanitization and strict validation", () => { + it("validates and sanitizes a valid merchantId", () => { + const result = validateAndSanitizeMerchantId("merchant-123"); + expect(result.valid).toBe(true); + expect(result.merchantId).toBe("merchant-123"); + }); + + it("trims whitespace from merchantId", () => { + const result = validateAndSanitizeMerchantId(" merchant-123 "); + expect(result.valid).toBe(true); + expect(result.merchantId).toBe("merchant-123"); + }); + + it("rejects null merchantId", () => { + const result = validateAndSanitizeMerchantId(null); + expect(result.valid).toBe(false); + expect(result.error).toBeDefined(); + }); + + it("rejects undefined merchantId", () => { + const result = validateAndSanitizeMerchantId(undefined); + expect(result.valid).toBe(false); + expect(result.error).toBeDefined(); + }); + + it("rejects empty string merchantId", () => { + const result = validateAndSanitizeMerchantId(""); + expect(result.valid).toBe(false); + expect(result.error).toBeDefined(); + }); + + it("rejects merchantId with invalid characters", () => { + const result = validateAndSanitizeMerchantId("merchant!@#"); + expect(result.valid).toBe(false); + }); + + it("rejects merchantId that is too long", () => { + const longId = "a".repeat(129); + const result = validateAndSanitizeMerchantId(longId); + expect(result.valid).toBe(false); + expect(result.error).toContain("128"); + }); + + it("accepts merchantId with allowed special characters", () => { + const result = validateAndSanitizeMerchantId("merchant_id:sub.domain-123@test"); + expect(result.valid).toBe(true); + expect(result.merchantId).toBe("merchant_id:sub.domain-123@test"); + }); + + it("sanitizes merchantId by stripping dangerous characters", () => { + const result = sanitizeMerchantId("merchant'; DROP TABLE merchants; --"); + expect(result).toBe("merchantDROPTABLEmerchants--"); + }); + + it("sanitizes returns null for non-string input", () => { + expect(sanitizeMerchantId(null)).toBeNull(); + expect(sanitizeMerchantId(undefined)).toBeNull(); + expect(sanitizeMerchantId(123)).toBeNull(); + }); + + it("sanitizes returns null for empty result after cleaning", () => { + expect(sanitizeMerchantId("!!!!!!!!")).toBeNull(); + }); + + it("sanitizes enforces length limit", () => { + const longId = "a".repeat(200); + expect(sanitizeMerchantId(longId)).toBeNull(); + }); +}); + +describe("withMerchantContext — transaction handling with validation", () => { + it("sets merchant context in transaction and commits on success", async () => { + const client = createMockClient(); + pool.connect.mockResolvedValue(client); + + const callback = vi.fn().mockResolvedValue({ data: "test" }); + const result = await withMerchantContext("merchant-123", callback); + + expect(result).toEqual({ data: "test" }); + expect(pool.connect).toHaveBeenCalledTimes(1); + expect(client.query).toHaveBeenCalledWith("BEGIN"); + expect(client.query).toHaveBeenCalledWith("SET LOCAL app.current_merchant_id = $1", ["merchant-123"]); + expect(client.query).toHaveBeenCalledWith("COMMIT"); + expect(callback).toHaveBeenCalledWith(client); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it("rolls back on callback error", async () => { + const client = createMockClient(); + pool.connect.mockResolvedValue(client); + + const error = new Error("callback failed"); + const callback = vi.fn().mockRejectedValue(error); + + await expect(withMerchantContext("merchant-123", callback)).rejects.toThrow("callback failed"); + expect(client.query).toHaveBeenCalledWith("BEGIN"); + expect(client.query).toHaveBeenCalledWith("SET LOCAL app.current_merchant_id = $1", ["merchant-123"]); + expect(client.query).toHaveBeenCalledWith("ROLLBACK"); + expect(client.query).not.toHaveBeenCalledWith("COMMIT"); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it("throws if merchantId is invalid", async () => { + const callback = vi.fn(); + await expect(withMerchantContext("", callback)).rejects.toThrow("Invalid merchantId"); + expect(pool.connect).not.toHaveBeenCalled(); + }); + + it("throws if callback is not a function", async () => { + await expect(withMerchantContext("merchant-123", null)).rejects.toThrow("callback must be a function"); + expect(pool.connect).not.toHaveBeenCalled(); + }); + + it("throws if callback is not a function (string)", async () => { + await expect(withMerchantContext("merchant-123", "not-a-function")).rejects.toThrow("callback must be a function"); + expect(pool.connect).not.toHaveBeenCalled(); + }); + + it("handles rollback failure gracefully", async () => { + const client = createMockClient(); + client.query + .mockResolvedValueOnce({ rows: [] }) // BEGIN + .mockResolvedValueOnce({ rows: [] }) // SET LOCAL + .mockRejectedValueOnce(new Error("rollback failed")); // ROLLBACK fails + + pool.connect.mockResolvedValue(client); + + const error = new Error("callback failed"); + const callback = vi.fn().mockRejectedValue(error); + + await expect(withMerchantContext("merchant-123", callback)).rejects.toThrow("callback failed"); + expect(client.release).toHaveBeenCalledTimes(1); + }); + + it("handles connection errors", async () => { + pool.connect.mockRejectedValue(new Error("connection failed")); + const callback = vi.fn(); + await expect(withMerchantContext("merchant-123", callback)).rejects.toThrow("connection failed"); + }); +}); From b1fba4d5d9588c72d672839f1ae0e8fb8d08c960 Mon Sep 17 00:00:00 2001 From: Abidoyesimze <154632691+Abidoyesimze@users.noreply.github.com> Date: Sat, 17 Oct 2026 17:41:32 +0100 Subject: [PATCH 2/2] feat(backend): add Prometheus alert metrics and health telemetry to Database Row-Level Security Service MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every merchant-scoped transaction goes through withMerchantContext() but the RLS path had no metrics, health signal or alerting, so connection exhaustion, failing SET LOCAL transactions and rollback storms were invisible until queries started failing in callers: - Adds db-rls-telemetry.js: outcome counters, latency histogram, active- transaction gauge and a rolling-window health monitor with scrape-time gauges, on its own prom-client registry. Label values are a fixed set (committed | rolled_back | invalid_input | connection_error | transaction_error) — never merchant ids, SQL or error strings — so cardinality stays bounded. - Instruments withMerchantContext to record one outcome per attempt (invalid_input fails fast before touching the pool; connection_error covers pool.connect failure; rolled_back covers callback failures; transaction_error covers BEGIN/SET LOCAL/COMMIT and failed ROLLBACK). - Exposes GET /health/database-row-level-security (503 only when unhealthy; degraded still 200) and reports status under services in GET /health. - Merges the RLS registry into the /metrics scrape output. - Adds alert rules (docs/alerts/database-row-level-security.rules.yml) for unhealthy state, error ratio, rollback ratio, invalid-input spikes and missing traffic, plus docs/DATABASE_ROW_LEVEL_SECURITY.md. Adds db-rls-telemetry.test.js (18 tests) covering the health monitor windows, thresholds and expiry, counter/histogram recording, and outcome attribution through withMerchantContext with a mocked pool. --- backend/docs/DATABASE_ROW_LEVEL_SECURITY.md | 93 +++++++ .../database-row-level-security.rules.yml | 81 ++++++ backend/src/app.js | 27 ++ backend/src/lib/db-rls-telemetry.js | 260 ++++++++++++++++++ backend/src/lib/db-rls-telemetry.test.js | 225 +++++++++++++++ backend/src/lib/db-rls.js | 34 ++- backend/src/routes/prometheus.js | 6 +- 7 files changed, 724 insertions(+), 2 deletions(-) create mode 100644 backend/docs/DATABASE_ROW_LEVEL_SECURITY.md create mode 100644 backend/docs/alerts/database-row-level-security.rules.yml create mode 100644 backend/src/lib/db-rls-telemetry.js create mode 100644 backend/src/lib/db-rls-telemetry.test.js diff --git a/backend/docs/DATABASE_ROW_LEVEL_SECURITY.md b/backend/docs/DATABASE_ROW_LEVEL_SECURITY.md new file mode 100644 index 00000000..d1789935 --- /dev/null +++ b/backend/docs/DATABASE_ROW_LEVEL_SECURITY.md @@ -0,0 +1,93 @@ +# Database Row-Level Security Service — Telemetry + +Issue #1468. Every merchant-scoped transaction runs through `withMerchantContext()` +(`src/lib/db-rls.js`), which sets `app.current_merchant_id` inside a transaction so +Postgres Row-Level Security policies apply. This document covers the Prometheus +metrics, health endpoint and alert rules added for that service. + +## Metrics + +Exported from `src/lib/db-rls-telemetry.js` on its own registry, merged into the +main `/metrics` scrape by `src/routes/prometheus.js`. Label values are a fixed +server-side set — never merchant ids, SQL text or error strings — so request +traffic cannot inflate series cardinality or leak customer data into scrapes. + +| Metric | Type | Labels | Meaning | +| --- | --- | --- | --- | +| `database_rls_transactions_total` | Counter | `outcome` | Transaction attempts by outcome | +| `database_rls_transaction_duration_seconds` | Histogram | `outcome` | Wall time of the whole `withMerchantContext` call | +| `database_rls_active_transactions` | Gauge | — | Transactions currently in flight | +| `database_rls_health_state` | Gauge | — | `0` healthy, `1` degraded, `2` unhealthy (rolling window) | +| `database_rls_error_ratio` | Gauge | — | Share of attempts that failed before completing | +| `database_rls_rollback_ratio` | Gauge | — | Share of attempts rolled back by a failing callback | +| `database_rls_last_transaction_timestamp_seconds` | Gauge | — | Unix time of the most recent attempt | + +`outcome` values: + +- `committed` — transaction committed successfully +- `rolled_back` — the caller's callback threw; the transaction was rolled back +- `invalid_input` — merchantId or callback failed validation; the pool was never touched +- `connection_error` — `pool.connect()` failed (pool exhausted, DB down) +- `transaction_error` — `BEGIN` / `SET LOCAL` / `COMMIT` failed, or `ROLLBACK` itself failed + +## Health endpoint + +`GET /health/database-row-level-security` returns the rolling-window snapshot: + +```json +{ + "status": "healthy", + "reasons": [], + "window_ms": 300000, + "total": 1421, + "committed": 1418, + "rolled_back": 3, + "invalid_input": 0, + "connection_error": 0, + "transaction_error": 0, + "errors": 0, + "rollback_ratio": 0.0021, + "error_ratio": 0, + "last_transaction_at": "2026-10-17T12:00:00.000Z", + "thresholds": { + "min_samples": 20, + "error_ratio": 0.05, + "rollback_ratio": 0.5, + "invalid_input": 10 + } +} +``` + +Returns 503 only when `status` is `unhealthy`; `degraded` still returns 200. The +same status is reported (informationally) under `services.database_row_level_security` +in the main `GET /health` payload. The body contains no merchant ids, query text +or error strings. + +Status rules over the rolling window (default 5 minutes, 20-sample minimum): + +- **unhealthy** — connection + transaction errors ≥ 5% of attempts +- **degraded** — rolled-back callbacks ≥ 50% of attempts, or ≥ 10 `invalid_input` outcomes +- **healthy** — everything else + +Thresholds are overridable with `DATABASE_RLS_HEALTH_WINDOW_MS`, +`DATABASE_RLS_HEALTH_MIN_SAMPLES`, `DATABASE_RLS_ERROR_RATIO_THRESHOLD`, +`DATABASE_RLS_ROLLBACK_RATIO_THRESHOLD` and `DATABASE_RLS_INVALID_INPUT_THRESHOLD`. + +## Alert rules + +`docs/alerts/database-row-level-security.rules.yml` — load with +`rule_files` and validate with +`promtool check rules docs/alerts/database-row-level-security.rules.yml`: + +- `DatabaseRlsUnhealthy` (critical) — health state ≥ 2 for 5m +- `DatabaseRlsHighErrorRatio` (critical) — > 5% connection/transaction errors over 10m +- `DatabaseRlsHighRollbackRatio` (warning) — > 50% callback rollbacks over 15m +- `DatabaseRlsInvalidInputSpike` (warning) — > 10 invalid merchant ids in 10m (a caller is misbehaving) +- `DatabaseRlsNoTraffic` (info) — no RLS transactions for 2h + +## Notes on privacy + +Telemetry records only aggregate counts, durations and derived ratios. Merchant +ids are never used as label values or logged into the health snapshot, so the +scrape and health outputs stay safe to expose to the same monitoring stack as the +rest of the API. diff --git a/backend/docs/alerts/database-row-level-security.rules.yml b/backend/docs/alerts/database-row-level-security.rules.yml new file mode 100644 index 00000000..da18790d --- /dev/null +++ b/backend/docs/alerts/database-row-level-security.rules.yml @@ -0,0 +1,81 @@ +# Prometheus alerting rules for the Database Row-Level Security Service (issue #1468). +# +# Load with: rule_files: ["docs/alerts/database-row-level-security.rules.yml"] +# Validate: promtool check rules docs/alerts/database-row-level-security.rules.yml +# +# Metric reference: docs/DATABASE_ROW_LEVEL_SECURITY.md + +groups: + - name: database-row-level-security + rules: + - alert: DatabaseRlsUnhealthy + expr: max(database_rls_health_state) >= 2 + for: 5m + labels: + severity: critical + annotations: + summary: Database Row-Level Security service reports unhealthy + description: > + Rolling-window internal error ratio is above threshold — + connections or transactions are failing before commit. See + GET /health/database-row-level-security for counts and reasons. + + - alert: DatabaseRlsHighErrorRatio + expr: | + ( + sum(rate(database_rls_transactions_total{outcome="connection_error"}[10m])) + + sum(rate(database_rls_transactions_total{outcome="transaction_error"}[10m])) + ) + / + clamp_min(sum(rate(database_rls_transactions_total[10m])), 1e-9) + > 0.05 + and sum(increase(database_rls_transactions_total[10m])) >= 20 + for: 10m + labels: + severity: critical + annotations: + summary: More than 5% of RLS transactions fail before completing + description: > + Check logs for "Failed to acquire pooled connection" or + "Failed to rollback transaction", and inspect Postgres + connection limits and transaction timeouts. + + - alert: DatabaseRlsHighRollbackRatio + expr: | + sum(rate(database_rls_transactions_total{outcome="rolled_back"}[15m])) + / + clamp_min(sum(rate(database_rls_transactions_total[15m])), 1e-9) + > 0.5 + and sum(increase(database_rls_transactions_total[15m])) >= 20 + for: 15m + labels: + severity: warning + annotations: + summary: Majority of RLS transactions are rolled back by failing callbacks + description: > + withMerchantContext is rolling back most transactions, which usually + means merchant-scoped queries are erroring (bad SQL, constraint + violations, or validation inside the callback). + + - alert: DatabaseRlsInvalidInputSpike + expr: sum(increase(database_rls_transactions_total{outcome="invalid_input"}[10m])) > 10 + for: 10m + labels: + severity: warning + annotations: + summary: Spike of invalid merchant ids reaching the RLS service + description: > + Callers are passing merchant ids that fail sanitization/validation + (empty, too long, or illegal characters). Find the offending caller + in the logs ("Invalid merchantId provided to withMerchantContext"). + + - alert: DatabaseRlsNoTraffic + expr: sum(increase(database_rls_transactions_total[1h])) == 0 + for: 2h + labels: + severity: info + annotations: + summary: No RLS transactions observed for two hours + description: > + The service may be unused in this environment, or the RLS path + has stopped being exercised. Confirm with deployment state. diff --git a/backend/src/app.js b/backend/src/app.js index b330d308..eb79bc17 100644 --- a/backend/src/app.js +++ b/backend/src/app.js @@ -49,6 +49,7 @@ import { versionDeprecationMiddleware } from "./lib/version-deprecation.js"; import oracleRouter from "./routes/oracle.js"; import { getPaymentSessionValidatorHealth } from "./lib/payment-session-validator.js"; import { getExchangeRateOracleHealth } from "./lib/exchange-rate-oracle-telemetry.js"; +import { getDatabaseRowLevelSecurityHealth } from "./lib/db-rls-telemetry.js"; import { configureExchangeRateCoordination } from "./services/exchangeRateService.js"; export async function createApp({ redisClient }) { @@ -262,6 +263,7 @@ export async function createApp({ redisClient }) { // Informational only — does not affect `ok` / the status code. payment_session_validator: getPaymentSessionValidatorHealth().status, exchange_rate_oracle_cache: getExchangeRateOracleHealth().status, + database_row_level_security: getDatabaseRowLevelSecurityHealth().status, }, }); }); @@ -314,6 +316,31 @@ export async function createApp({ redisClient }) { res.status(health.status === "unhealthy" ? 503 : 200).json(health); }); + /** + * @swagger + * /health/database-row-level-security: + * get: + * summary: Database Row-Level Security Service health telemetry + * description: > + * Rolling-window counts, error / rollback ratios and derived status + * for transactions executed through withMerchantContext (issue #1468). + * Returns 503 only when the service is unhealthy (internal error ratio + * above threshold); a degraded status (high rollback ratio or an + * invalid-input spike) still returns 200. The body contains no + * merchant ids, query text or error strings. + * tags: [Health] + * security: [] + * responses: + * 200: + * description: RLS service healthy or degraded + * 503: + * description: RLS service unhealthy + */ + app.get("/health/database-row-level-security", (_req, res) => { + const health = getDatabaseRowLevelSecurityHealth(); + res.status(health.status === "unhealthy" ? 503 : 200).json(health); + }); + const verifyPaymentRateLimit = createVerifyPaymentRateLimit({ store: redisAvailable ? createRedisRateLimitStore({ client: redisClient }) : undefined, }); diff --git a/backend/src/lib/db-rls-telemetry.js b/backend/src/lib/db-rls-telemetry.js new file mode 100644 index 00000000..ec3651d6 --- /dev/null +++ b/backend/src/lib/db-rls-telemetry.js @@ -0,0 +1,260 @@ +/** + * Database Row-Level Security Service telemetry (issue #1468). + * + * Prometheus series and a rolling-window health snapshot for every + * `withMerchantContext()` transaction (src/lib/db-rls.js): + * + * - outcome counts and latency per outcome + * - active transaction gauge + * - rolling-window health state + ratios for alerting + * + * Label values come from a fixed server-side set (committed | rolled_back | + * invalid_input | connection_error | transaction_error) — never merchant ids, + * query text or error strings — so request traffic cannot inflate series + * cardinality or leak customer data into the scrape output. + * + * The metrics live in their own registry so they can be unit-tested in + * isolation; the /metrics endpoint merges this registry with the others. + */ + +import client from "prom-client"; + +const register = new client.Registry(); + +register.setDefaultLabels({ + app: "stellar-payment-api", +}); + +/** + * outcome: committed | rolled_back | invalid_input | connection_error | transaction_error + */ +export const dbRlsTransactionsTotal = new client.Counter({ + name: "database_rls_transactions_total", + help: "Row-Level Security transactions started through withMerchantContext, by outcome", + labelNames: ["outcome"], +}); + +export const dbRlsTransactionDuration = new client.Histogram({ + name: "database_rls_transaction_duration_seconds", + help: "Wall time of a withMerchantContext transaction, by outcome", + labelNames: ["outcome"], + buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10], +}); + +export const dbRlsActiveTransactions = new client.Gauge({ + name: "database_rls_active_transactions", + help: "RLS transactions currently in flight (connected, transaction open)", +}); + +/** + * Rolling-window gauges. Refreshed at scrape time so they decay when idle. + * health_state: 0 = healthy, 1 = degraded, 2 = unhealthy + */ +export const dbRlsHealthState = new client.Gauge({ + name: "database_rls_health_state", + help: "Row-Level Security service health (0 = healthy, 1 = degraded, 2 = unhealthy)", +}); + +export const dbRlsErrorRatio = new client.Gauge({ + name: "database_rls_error_ratio", + help: "Share of RLS transactions that failed before completing (connection or transaction error) over the rolling window", +}); + +export const dbRlsRollbackRatio = new client.Gauge({ + name: "database_rls_rollback_ratio", + help: "Share of RLS transactions rolled back by a failing callback over the rolling window", +}); + +export const dbRlsLastTransactionTimestamp = new client.Gauge({ + name: "database_rls_last_transaction_timestamp_seconds", + help: "Unix time of the most recent RLS transaction attempt", +}); + +const OUTCOMES = new Set([ + "committed", + "rolled_back", + "invalid_input", + "connection_error", + "transaction_error", +]); + +export const HEALTH_STATE_VALUES = Object.freeze({ + healthy: 0, + degraded: 1, + unhealthy: 2, +}); + +function readNumberEnv(name, fallback) { + const raw = Number.parseFloat(process.env[name] ?? ""); + return Number.isFinite(raw) && raw >= 0 ? raw : fallback; +} + +export const DEFAULT_HEALTH_OPTIONS = Object.freeze({ + windowMs: readNumberEnv("DATABASE_RLS_HEALTH_WINDOW_MS", 300_000), + bucketMs: 5_000, + minSamples: readNumberEnv("DATABASE_RLS_HEALTH_MIN_SAMPLES", 20), + errorRatioThreshold: readNumberEnv("DATABASE_RLS_ERROR_RATIO_THRESHOLD", 0.05), + rollbackRatioThreshold: readNumberEnv("DATABASE_RLS_ROLLBACK_RATIO_THRESHOLD", 0.5), + invalidInputThreshold: readNumberEnv("DATABASE_RLS_INVALID_INPUT_THRESHOLD", 10), +}); + +function emptyBucket() { + return { + slot: -1, + committed: 0, + rolledBack: 0, + invalidInput: 0, + connectionErrors: 0, + transactionErrors: 0, + }; +} + +/** + * Rolling-window health tracker. Counts live in a fixed ring of time buckets + * so memory stays constant regardless of traffic. + */ +export class DatabaseRlsHealthMonitor { + constructor(options = {}, now = () => Date.now()) { + this.options = { ...DEFAULT_HEALTH_OPTIONS, ...options }; + this.now = now; + this.bucketCount = Math.max( + 1, + Math.ceil(this.options.windowMs / this.options.bucketMs), + ); + this.reset(); + } + + reset() { + this.buckets = Array.from({ length: this.bucketCount }, emptyBucket); + this.lastTransactionAt = null; + } + + _bucketFor(time) { + const slot = Math.floor(time / this.options.bucketMs); + const bucket = this.buckets[slot % this.bucketCount]; + if (bucket.slot !== slot) { + Object.assign(bucket, emptyBucket(), { slot }); + } + return bucket; + } + + record(outcome, time = this.now()) { + const bucket = this._bucketFor(time); + if (outcome === "committed") bucket.committed++; + else if (outcome === "rolled_back") bucket.rolledBack++; + else if (outcome === "invalid_input") bucket.invalidInput++; + else if (outcome === "connection_error") bucket.connectionErrors++; + else if (outcome === "transaction_error") bucket.transactionErrors++; + this.lastTransactionAt = time; + } + + snapshot() { + const time = this.now(); + const currentSlot = Math.floor(time / this.options.bucketMs); + const oldestSlot = currentSlot - this.bucketCount + 1; + const counts = { + committed: 0, + rolled_back: 0, + invalid_input: 0, + connection_error: 0, + transaction_error: 0, + }; + + for (const bucket of this.buckets) { + if (bucket.slot < oldestSlot || bucket.slot > currentSlot) continue; + counts.committed += bucket.committed; + counts.rolled_back += bucket.rolledBack; + counts.invalid_input += bucket.invalidInput; + counts.connection_error += bucket.connectionErrors; + counts.transaction_error += bucket.transactionErrors; + } + + const errors = counts.connection_error + counts.transaction_error; + const total = counts.committed + counts.rolled_back + counts.invalid_input + errors; + const errorRatio = total > 0 ? errors / total : 0; + const rollbackRatio = total > 0 ? counts.rolled_back / total : 0; + const { minSamples, errorRatioThreshold, rollbackRatioThreshold, invalidInputThreshold } = + this.options; + + const reasons = []; + let status = "healthy"; + if (total >= minSamples && errorRatio >= errorRatioThreshold) { + status = "unhealthy"; + reasons.push("error_ratio_exceeded"); + } + if (total >= minSamples && rollbackRatio >= rollbackRatioThreshold) { + if (status === "healthy") status = "degraded"; + reasons.push("rollback_ratio_exceeded"); + } + if (counts.invalid_input >= invalidInputThreshold) { + if (status === "healthy") status = "degraded"; + reasons.push("invalid_input_spike"); + } + + return { + status, + reasons, + window_ms: this.bucketCount * this.options.bucketMs, + total, + ...counts, + errors, + rollback_ratio: Number(rollbackRatio.toFixed(4)), + error_ratio: Number(errorRatio.toFixed(4)), + last_transaction_at: + this.lastTransactionAt === null + ? null + : new Date(this.lastTransactionAt).toISOString(), + thresholds: { + min_samples: minSamples, + error_ratio: errorRatioThreshold, + rollback_ratio: rollbackRatioThreshold, + invalid_input: invalidInputThreshold, + }, + }; + } +} + +export const healthMonitor = new DatabaseRlsHealthMonitor(); + +// Refresh rolling-window gauges at scrape time so they decay when idle. +dbRlsHealthState.collect = function collectHealthState() { + const snap = healthMonitor.snapshot(); + this.set(HEALTH_STATE_VALUES[snap.status]); + dbRlsErrorRatio.set(snap.error_ratio); + dbRlsRollbackRatio.set(snap.rollback_ratio); +}; + +/** Record one RLS transaction attempt: counters, histogram and health window. */ +export function recordRlsTransaction(outcome, durationSeconds) { + const label = OUTCOMES.has(outcome) ? outcome : "transaction_error"; + dbRlsTransactionsTotal.inc({ outcome: label }); + if (Number.isFinite(durationSeconds)) { + dbRlsTransactionDuration.observe({ outcome: label }, durationSeconds); + } + dbRlsLastTransactionTimestamp.set(Date.now() / 1000); + healthMonitor.record(label); +} + +/** Current RLS service health, for /health endpoints. */ +export function getDatabaseRowLevelSecurityHealth() { + return healthMonitor.snapshot(); +} + +/** Test-only: clear counters, histogram and the rolling health window. */ +export function resetDatabaseRowLevelSecurityTelemetry() { + dbRlsTransactionsTotal.reset(); + dbRlsTransactionDuration.reset(); + dbRlsActiveTransactions.reset(); + dbRlsLastTransactionTimestamp.set(0); + healthMonitor.reset(); +} + +register.registerMetric(dbRlsTransactionsTotal); +register.registerMetric(dbRlsTransactionDuration); +register.registerMetric(dbRlsActiveTransactions); +register.registerMetric(dbRlsHealthState); +register.registerMetric(dbRlsErrorRatio); +register.registerMetric(dbRlsRollbackRatio); +register.registerMetric(dbRlsLastTransactionTimestamp); + +export { register as dbRlsRegister }; diff --git a/backend/src/lib/db-rls-telemetry.test.js b/backend/src/lib/db-rls-telemetry.test.js new file mode 100644 index 00000000..5885dae0 --- /dev/null +++ b/backend/src/lib/db-rls-telemetry.test.js @@ -0,0 +1,225 @@ +import { describe, it, expect, beforeEach, vi, afterEach } from "vitest"; + +vi.mock("../lib/db.js", () => { + const client = { + query: vi.fn(async (sql) => { + if (String(sql).trim().toUpperCase().startsWith("SELECT")) { + return { rows: [{ ok: 1 }] }; + } + return { rows: [] }; + }), + release: vi.fn(), + }; + return { pool: { connect: vi.fn(async () => client) }, __client: client }; +}); + +vi.mock("../lib/logger.js", () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() }, +})); + +import * as db from "../lib/db.js"; +const { pool } = db; +const client = db.__client; +import { withMerchantContext, validateAndSanitizeMerchantId } from "../lib/db-rls.js"; +import { + DatabaseRlsHealthMonitor, + recordRlsTransaction, + getDatabaseRowLevelSecurityHealth, + resetDatabaseRowLevelSecurityTelemetry, + dbRlsTransactionsTotal, + dbRlsTransactionDuration, + dbRlsActiveTransactions, + dbRlsRegister, + HEALTH_STATE_VALUES, +} from "../lib/db-rls-telemetry.js"; + +async function counterValue(outcome) { + const metric = dbRlsRegister.getSingleMetric("database_rls_transactions_total"); + const { values } = await metric.get(); + return values.find((v) => v.labels.outcome === outcome)?.value ?? 0; +} + +async function gaugeValue(name) { + const metric = dbRlsRegister.getSingleMetric(name); + const { values } = await metric.get(); + return values[0]?.value ?? 0; +} + +function bucketStart(now, bucketMs = 5_000) { + return Math.floor(now / bucketMs) * bucketMs; +} + +describe("DatabaseRlsHealthMonitor", () => { + let now; + let monitor; + + beforeEach(() => { + now = Date.now(); + monitor = new DatabaseRlsHealthMonitor( + { windowMs: 60_000, bucketMs: 5_000, minSamples: 10, errorRatioThreshold: 0.05, rollbackRatioThreshold: 0.5, invalidInputThreshold: 5 }, + () => now, + ); + }); + + it("reports healthy with no traffic and zero ratios", () => { + const snap = monitor.snapshot(); + expect(snap.status).toBe("healthy"); + expect(snap.total).toBe(0); + expect(snap.error_ratio).toBe(0); + expect(snap.rollback_ratio).toBe(0); + expect(snap.last_transaction_at).toBeNull(); + }); + + it("is healthy when all transactions commit", () => { + for (let i = 0; i < 20; i++) monitor.record("committed"); + const snap = monitor.snapshot(); + expect(snap.status).toBe("healthy"); + expect(snap.total).toBe(20); + expect(snap.committed).toBe(20); + expect(snap.errors).toBe(0); + }); + + it("goes unhealthy when error ratio exceeds threshold with enough samples", () => { + for (let i = 0; i < 18; i++) monitor.record("committed"); + for (let i = 0; i < 2; i++) monitor.record("connection_error"); + const snap = monitor.snapshot(); + expect(snap.status).toBe("unhealthy"); + expect(snap.reasons).toContain("error_ratio_exceeded"); + expect(snap.errors).toBe(2); + expect(snap.error_ratio).toBeCloseTo(0.1, 4); + }); + + it("stays healthy on a tiny sample below minSamples", () => { + monitor.record("connection_error"); + const snap = monitor.snapshot(); + expect(snap.status).toBe("healthy"); + expect(snap.errors).toBe(1); + }); + + it("goes degraded on a high rollback ratio", () => { + for (let i = 0; i < 12; i++) monitor.record("rolled_back"); + for (let i = 0; i < 8; i++) monitor.record("committed"); + const snap = monitor.snapshot(); + expect(snap.status).toBe("degraded"); + expect(snap.reasons).toContain("rollback_ratio_exceeded"); + }); + + it("goes degraded on an invalid-input spike", () => { + for (let i = 0; i < 5; i++) monitor.record("invalid_input"); + const snap = monitor.snapshot(); + expect(snap.status).toBe("degraded"); + expect(snap.reasons).toContain("invalid_input_spike"); + }); + + it("expires counts outside the rolling window", () => { + for (let i = 0; i < 20; i++) monitor.record("connection_error", now - 120_000); + for (let i = 0; i < 5; i++) monitor.record("committed"); + const snap = monitor.snapshot(); + expect(snap.connection_error).toBe(0); + expect(snap.committed).toBe(5); + expect(snap.status).toBe("healthy"); + }); + + it("records the last transaction timestamp", () => { + monitor.record("committed"); + expect(monitor.snapshot().last_transaction_at).not.toBeNull(); + }); +}); + +describe("recordRlsTransaction", () => { + beforeEach(() => resetDatabaseRowLevelSecurityTelemetry()); + afterEach(() => resetDatabaseRowLevelSecurityTelemetry()); + + it("increments counters and observes duration per outcome", async () => { + recordRlsTransaction("committed", 0.01); + recordRlsTransaction("committed", 0.02); + recordRlsTransaction("rolled_back", 0.03); + expect(await counterValue("committed")).toBe(2); + expect(await counterValue("rolled_back")).toBe(1); + const duration = await dbRlsTransactionDuration.get(); + const observations = duration.values + .filter((v) => v.metricName === "database_rls_transaction_duration_seconds_count") + .reduce((sum, v) => sum + v.value, 0); + expect(observations).toBe(3); + expect(await gaugeValue("database_rls_active_transactions")).toBe(0); + }); + + it("falls back to transaction_error for unknown outcomes", async () => { + recordRlsTransaction("mystery", 0.01); + expect(await counterValue("transaction_error")).toBe(1); + expect(await counterValue("mystery")).toBe(0); + }); + + it("drives the rolling health snapshot", async () => { + for (let i = 0; i < 18; i++) recordRlsTransaction("committed", 0.01); + for (let i = 0; i < 2; i++) recordRlsTransaction("transaction_error", 0.01); + const health = getDatabaseRowLevelSecurityHealth(); + expect(health.status).toBe("unhealthy"); + expect(health.transaction_error).toBe(2); + }); +}); + +describe("withMerchantContext telemetry", () => { + beforeEach(() => { + resetDatabaseRowLevelSecurityTelemetry(); + vi.clearAllMocks(); + }); + afterEach(() => resetDatabaseRowLevelSecurityTelemetry()); + + it("records committed outcome and active gauge returns to zero", async () => { + await withMerchantContext("merch_1", async (c) => { + expect(await gaugeValue("database_rls_active_transactions")).toBe(1); + await c.query("SELECT 1"); + return "ok"; + }); + expect(await counterValue("committed")).toBe(1); + expect(await gaugeValue("database_rls_active_transactions")).toBe(0); + }); + + it("records rolled_back outcome when the callback throws", async () => { + await expect( + withMerchantContext("merch_1", async () => { + throw new Error("boom"); + }), + ).rejects.toThrow("boom"); + expect(await counterValue("rolled_back")).toBe(1); + expect(await gaugeValue("database_rls_active_transactions")).toBe(0); + expect(client.release).toHaveBeenCalled(); + }); + + it("records invalid_input for a bad merchant id without touching the pool", async () => { + await expect(withMerchantContext("bad id!", () => {})).rejects.toThrow(/Invalid merchantId/); + expect(await counterValue("invalid_input")).toBe(1); + expect(pool.connect).not.toHaveBeenCalled(); + }); + + it("records invalid_input for a non-function callback", async () => { + await expect(withMerchantContext("merch_1", "nope")).rejects.toThrow(/callback must be a function/); + expect(await counterValue("invalid_input")).toBe(1); + expect(pool.connect).not.toHaveBeenCalled(); + }); + + it("records connection_error when the pool cannot provide a client", async () => { + pool.connect.mockRejectedValueOnce(new Error("pool exhausted")); + await expect(withMerchantContext("merch_1", async () => "ok")).rejects.toThrow("pool exhausted"); + expect(await counterValue("connection_error")).toBe(1); + expect(await gaugeValue("database_rls_active_transactions")).toBe(0); + }); + + it("records transaction_error when BEGIN fails", async () => { + client.query.mockImplementationOnce(async () => { + throw new Error("begin failed"); + }); + await expect(withMerchantContext("merch_1", async () => "ok")).rejects.toThrow("begin failed"); + expect(await counterValue("transaction_error")).toBe(1); + expect(await gaugeValue("database_rls_active_transactions")).toBe(0); + }); +}); + +describe("validation still enforced", () => { + it("rejects oversized and malformed ids", () => { + expect(validateAndSanitizeMerchantId("x".repeat(200)).valid).toBe(false); + expect(validateAndSanitizeMerchantId("bad;drop").valid).toBe(false); + expect(validateAndSanitizeMerchantId("merch_1").valid).toBe(true); + }); +}); diff --git a/backend/src/lib/db-rls.js b/backend/src/lib/db-rls.js index 37dafe42..5e5ce890 100644 --- a/backend/src/lib/db-rls.js +++ b/backend/src/lib/db-rls.js @@ -1,6 +1,10 @@ import { z } from "zod"; import { pool } from "../lib/db.js"; import { logger } from "../lib/logger.js"; +import { + recordRlsTransaction, + dbRlsActiveTransactions, +} from "../lib/db-rls-telemetry.js"; // ── Validation Schemas ────────────────────────────────────────────────────── const MerchantIdSchema = z @@ -40,6 +44,12 @@ export function validateAndSanitizeMerchantId(merchantId) { * Includes payload sanitization and strict validation for merchantId. */ export async function withMerchantContext(merchantId, callback) { + const startedAt = process.hrtime.bigint(); + const finish = (outcome) => { + const seconds = Number(process.hrtime.bigint() - startedAt) / 1e9; + recordRlsTransaction(outcome, seconds); + }; + // Validate and sanitize merchantId const validation = validateAndSanitizeMerchantId(merchantId); if (!validation.valid) { @@ -47,6 +57,7 @@ export async function withMerchantContext(merchantId, callback) { { merchantId, error: validation.error }, "[RLS] Invalid merchantId provided to withMerchantContext" ); + finish("invalid_input"); throw new Error(`Invalid merchantId: ${validation.error}`); } @@ -58,22 +69,41 @@ export async function withMerchantContext(merchantId, callback) { { merchantId: sanitizedMerchantId }, "[RLS] Invalid callback provided to withMerchantContext" ); + finish("invalid_input"); throw new Error("callback must be a function"); } - const client = await pool.connect(); + let client; + try { + client = await pool.connect(); + } catch (err) { + logger.error({ err: err.message }, "[RLS] Failed to acquire pooled connection"); + finish("connection_error"); + throw err; + } + + dbRlsActiveTransactions.inc(); + let outcome = "committed"; + let callbackStarted = false; try { await client.query("BEGIN"); await client.query("SET LOCAL app.current_merchant_id = $1", [ sanitizedMerchantId, ]); + callbackStarted = true; const result = await callback(client); await client.query("COMMIT"); return result; } catch (err) { + if (callbackStarted) { + outcome = "rolled_back"; + } else { + outcome = "transaction_error"; + } try { await client.query("ROLLBACK"); } catch (rollbackErr) { + outcome = "transaction_error"; logger.error( { rollbackErr: rollbackErr.message }, "[RLS] Failed to rollback transaction" @@ -81,7 +111,9 @@ export async function withMerchantContext(merchantId, callback) { } throw err; } finally { + dbRlsActiveTransactions.dec(); client.release(); + finish(outcome); } } diff --git a/backend/src/routes/prometheus.js b/backend/src/routes/prometheus.js index 4a9cdde3..57c7393a 100644 --- a/backend/src/routes/prometheus.js +++ b/backend/src/routes/prometheus.js @@ -15,6 +15,9 @@ import { paymentSessionValidatorRegister } from "../lib/payment-session-validato // Exchange Rate Oracle Cache alert metrics live in their own registry (issue #1443) // and are merged into the scrape output below. import { exchangeRateOracleRegister } from "../lib/exchange-rate-oracle-telemetry.js"; +// Database Row-Level Security Service alert metrics live in their own registry (issue #1468) +// and are merged into the scrape output below. +import { dbRlsRegister } from "../lib/db-rls-telemetry.js"; const router = express.Router(); @@ -23,7 +26,7 @@ const router = express.Router(); * /metrics: * get: * summary: Expose Prometheus metrics - * description: Returns the current state of Prometheus metrics for the application, including granular payment processor, trustline manager, path payment, payment session validator and exchange-rate oracle cache metrics. + * description: Returns the current state of Prometheus metrics for the application, including granular payment processor, trustline manager, path payment, payment session validator, exchange-rate oracle cache and database row-level security metrics. * tags: [Monitoring] * responses: * 200: @@ -42,6 +45,7 @@ router.get("/metrics", async (req, res) => { pathPaymentRegister.metrics(), paymentSessionValidatorRegister.metrics(), exchangeRateOracleRegister.metrics(), + dbRlsRegister.metrics(), ]); res.set("Content-Type", register.contentType); res.end(scrapes.join("\n"));